IllegalStateException: Cannot Subscrie.Processor is already terminated

I created a new eventhub and trying to publish messages to eventHubA. When I am trying to send messages to eventhub , I get the following error:

java.lang.IllegalStateException: namespace[xxxxx] entityPath[xxxxx]: Cannot subscribe. Processor is already terminated at com.azure.core.amqp.implementation.AmqpChannelProcessor.subscribe(AmqpChannelProcessor.java:217)

Below is the code snippet i am using:

public void send(Response response) {
        String responseInString = JsonHandlingUtil.objectToJsonString(response);

        EventData eventData = new EventData(responseInString);
        // create a batch
        EventDataBatch eventDataBatch = producer.createBatch();

        // try to add the event from the array to the batch
        if (!eventDataBatch.tryAdd(eventData)) {
            // if the batch is full, send it and then create a new batch
            producer.send(eventDataBatch);
            eventDataBatch = producer.createBatch();

            // Try to add that event that couldn't fit before.
            if (!eventDataBatch.tryAdd(eventData)) {
                throw new IllegalArgumentException("Event is too large for an empty batch. Max size: "
                        + eventDataBatch.getMaxSizeInBytes());
            }
        }
        // send the last batch of remaining events
        if (eventDataBatch.getCount() > 0) {
            producer.send(eventDataBatch);
        }
        producer.close();
    }

JavaScript
​x
 
public void send(Response response) {        String responseInString = JsonHandlingUtil.objectToJsonString(response);​        EventData eventData = new EventData(responseInString);        // create a batch        EventDataBatch eventDataBatch = producer.createBatch();​        // try to add the event from the array to the batch        if (!eventDataBatch.tryAdd(eventData)) {            // if the batch is full, send it and then create a new batch            producer.send(eventDataBatch);            eventDataBatch = producer.createBatch();​            // Try to add that event that couldn't fit before.            if (!eventDataBatch.tryAdd(eventData)) {                throw new IllegalArgumentException("Event is too large for an empty batch. Max size: "                        + eventDataBatch.getMaxSizeInBytes());            }        }        // send the last batch of remaining events        if (eventDataBatch.getCount() > 0) {            producer.send(eventDataBatch);        }        producer.close();    }​

I have defined the eventhubProducerClient as a Bean.

 @Bean
    public EventHubProducerClient eventHubProducerClient() {
        return new EventHubClientBuilder()
                .transportType(AmqpTransportType.AMQP_WEB_SOCKETS)
                .connectionString(connectionString, eventHubName)
                .buildProducerClient();
    }

JavaScript
 
 @Bean    public EventHubProducerClient eventHubProducerClient() {        return new EventHubClientBuilder()                .transportType(AmqpTransportType.AMQP_WEB_SOCKETS)                .connectionString(connectionString, eventHubName)                .buildProducerClient();    }​

Below is my gradle dependencies

>    //eventhub
>         implementation 'com.azure:azure-messaging-eventhubs:5.7.0'
>         implementation group: 'io.projectreactor', name: 'reactor-core', version: '3.4.6'

JavaScript
 
>    //eventhub>         implementation 'com.azure:azure-messaging-eventhubs:5.7.0'>         implementation group: 'io.projectreactor', name: 'reactor-core', version: '3.4.6'​

Answer

From the follow-up question, it would appear that the root cause has been confirmed as the producer.close() call in the send method.

Since it appears that he producer is being managed as a singleton by the application, the mitigation would be to call close at the point when events are no longer going to be published, such as when the application is shutting down.

Advertisement

Answer