Consumer disconnect from Cluster after long time of inactivity

[Flooze-IT]

Hi everyone,

I am currently working on a project where we need to set up an architecture with Kafka at the center.

The problem I have once my consumers are deployed is that once they have been running for a while (1 day), after a while it disconnects from my cluster.

Clarification:

• I use OAUTHBEARER in protocol to protect my cluster (sometimes I can't refresh my token so expiration error -> Set up retry after login exception)

Here is my way to set up a listener:

    ContainerProperties containerProperties = new ContainerProperties(kafkaConnectionProperties.getTopicName());
    Map<String, Object> props = getInitConsumerConfig(connectorName,kafkaConnectionProperties, valueDeserializer,exchangeType);
    containerProperties.setMessageListener(messageListener);
    containerProperties.setAuthExceptionRetryInterval(Duration.ofSeconds(10L));

    ConsumerFactory<String, T> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
    if(exchangeType.equals("JSON")){
        try {
            consumerFactory = new DefaultKafkaConsumerFactory<>(props, new ErrorHandlingDeserializer(new KafkaKeyGenericDeserializer()), new ErrorHandlingDeserializer(new KafkaGenericDeserializer(objectMapper,impl)));
        } catch (NoSuchMethodException e) {
            throw new RuntimeException(e);
        }
    }
    KafkaMessageListenerContainer<String, T> listenerContainer = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
    listenerContainer.setAutoStartup(true);
    listenerContainer.setCommonErrorHandler(errorHandler());

    // Consumer Traceability (only with SpringKafka)
    SpringKafkaTelemetry springKafkaTelemetry = SpringKafkaTelemetry.create(opentelemetry);
    listenerContainer.setBatchInterceptor(springKafkaTelemetry.createBatchInterceptor());
    listenerContainer.setRecordInterceptor(springKafkaTelemetry.createRecordInterceptor());

    // bean name is the prefix of kafka consumer thread name
    listenerContainer.setBeanName("kafka-"+exchangeType+"-listener-"+connectorName);
    beanFactory.registerSingleton("kafka-"+exchangeType+"-listener-"+connectorName, listenerContainer);

And the recursive problem that occur in my console is :
Node 2147483647 disconnected. Group coordinator kafka-cluster-human-ressources-kafka-0.kafka-cluster-human-ressources-kafka-brokers.human-ressources.svc:9092 (id: 2147483647 rack: null) is unavailable or invalid due to cause: coordinator unavailable.isDisconnected: true. Rediscovery will be attempted Discovered group coordinator kafka-cluster-human-ressources-kafka-0.kafka-cluster-human-ressources-kafka-brokers.human-ressources.svc:9092 (id: 2147483647 rack: null)

But nothing happened, no reconnection.

Thanks for your help

[garyrussell]

We are not Kafka experts here; especially with regards to authentication/authorization and connection management. I suggest you ask the wider Kafka community, either on Stack Overflow and/or via their user mailing list.