azure-sdk-for-java: [BUG] EventHub Consumer stops consuming messages until we restart

Describe the bug EventHub consumer stops consuming messages until we restart the consumer. Have deployed my consumer in Azure Kubernetes (AKS). Initially, it consumes a few messages, all of a sudden it stops consuming. If we restart the consumer, it works like charm. Until we restart it, all the messages sit in EH. Even it is 2-3 days, none of the messages is consumed.

Exception or Stack Trace This is the consumer log just captured before restarting: consumerlogmessages.txt Picked up this stack trace from log:

2020-12-03 14:56:02.126 [ERROR] [Loggers$Slf4JLogger:319] Scheduler worker in group main failed with an uncaught exception java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 60000ms in 'takeUntil' (and no fallback has been configured)
	at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.handleTimeout(FluxTimeout.java:289)
	at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.doTimeout(FluxTimeout.java:274)
	at reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.onNext(FluxTimeout.java:396)

To Reproduce This has happened very often.

Code Snippet

import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventProcessorClient;
import com.azure.messaging.eventhubs.EventProcessorClientBuilder;
import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
import com.azure.storage.blob.BlobContainerAsyncClient;
import com.azure.storage.blob.BlobContainerClientBuilder;
import com.roo.connect.soa.config.IEventHubReceiverConfig;
import com.roo.connect.soa.log.ServiceLogger;
import org.springframework.beans.factory.annotation.Autowired;

public abstract class IEventHubReceiver implements Runnable {

    private EventProcessorClient eventProcessorClient;
    private String consumerGroupName="";
    private static final String IEVENTHUBRECIEVER = "IEventHubReceiver : ";

    @Autowired
    private ServiceLogger log;

    public IEventHubReceiver(IEventHubReceiverConfig config){

        BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
                .connectionString(config.getStorageConnectionString())
                .containerName(config.getStorageContainerName())
                .buildAsyncClient();

        eventProcessorClient = new EventProcessorClientBuilder()
                .consumerGroup(config.getConsumerGroup())
                .connectionString(config.getConnectionString())
                .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
                .processEvent(eventContext -> {
                    EventData eventData = eventContext.getEventData();
                    String receivedData = eventData.getBodyAsString();
                    consumerGroupName = config.getConsumerGroup();
                    log.info("IEventHubReceiver - received from : " + consumerGroupName + " : receivedData : " + receivedData);
                    onEvents(eventData);
                    if(isAlive()) {
                        //if onEvent is success, context updates the 'checkpoint'
                        eventContext.updateCheckpoint();
                    }
                })
                .processError(errorContext -> {
                    log.error(IEVENTHUBRECIEVER + consumerGroupName + " - Error occurred while processing events :: " + errorContext.getThrowable().getMessage());
                })
                .buildEventProcessorClient();
    }

    //This method is implemented in child classes with the logic to process the eventData
    public abstract void onEvents(EventData eventData);

    public void stop() {

        eventProcessorClient.stop();
        log.info(IEVENTHUBRECIEVER + consumerGroupName + " - Client stopped...");
    }

    public boolean isAlive() {

        boolean isAlive = eventProcessorClient.isRunning();
        log.info(IEVENTHUBRECIEVER + consumerGroupName + " - isAlive:: " + isAlive);
        return isAlive;
    }

    @Override
    public void run() {

        // This will start the processor. It will start processing events from all partitions.
        eventProcessorClient.start();
        log.info(IEVENTHUBRECIEVER + consumerGroupName + " - Client started...");
    }
}

Expected behavior I could see a few connection exception errors in logs, but I am wondering why it stopped consuming messages forever. Immediately after restarting messages started consuming.

Setup (please complete the following information):

• OS: Ubuntu (AKS)
• IDE : IntelliJ
• Version of the Library used

    <dependency>
      <groupId>com.azure</groupId>
      <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
      <version>1.3.1</version>
    </dependency>

Information Checklist Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• Bug Description Added
• Repro Steps Added
• Setup information Added