Disconnect exception when listening to multiple topics

[venkatasreekanth]

Hi,

I have requirement to populate a datastore from multiple topics(15+), I have each kafkalistener setup with its on topic and when I launch the project, except for one time all others continuously fail with timeoutexception or disconnect exception. When the topics were lesser in number setting request timeout to 60seconds helped, but with increased number of topics its no longer working. Asking the admins to increase the timeout is causing some pushback. Is there a way to start the kafkalisteners in sequence?

below is an example of how my listener/container is setup

@KafkaListener(autoStartup = "${listener.autostartup.*****.name}",
            id = "${listener.id.***.name}",
            groupId = "${listener.group.id.****.name}",
            containerFactory = "****ContainerFactory",
            topics = {"${listener.topic.****.name}"})
    public void ***Consumer(List<ConsumerRecord<String,String>> recordsList, Acknowledgment acknowledgment){
        storeInDB(mongoClient,mongoDatabaseName,*******,recordsList,acknowledgment);
    }

@Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    ******ContainerFactory(@Autowired KafkaProperties kafkaProperties) {

        Map<String,Object> consumerProperties = kafkaProperties.buildConsumerProperties();
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProperties));
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(60000);
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

[garyrussell]

Set autoStartup to false.

Then use the KafkaListenerEndpointRegistry to start each container - get a reference to each container from the registry using its id (or get them all and iterate over them).

If you implement SmartLifecycle you can put your bean in the last phase Integer.MAX_VALUE and put the code to start the containers (with a delay) in the start() method.

[venkatasreekanth]

@garyrussell is there some documentation or sample for implementing smart lifecycle. I did try setting autostartup to false trying to invoke start by creating rest endpoints, seems to have not effect, I am probably doing something wrong.

When starting all the containers I see this exception, could you let us know the context behind this exception

org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition tst_procurement_pim_item-12 could be determined
	at org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler.handle(SeekToCurrentBatchErrorHandler.java:79)
	at org.springframework.kafka.listener.RecoveringBatchErrorHandler.handle(RecoveringBatchErrorHandler.java:124)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1603)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1210)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.lang.Thread.run(Thread.java:834)

[garyrussell]

Your SmartLifecycle must return true to isAutoStartup; set the listener autoStartup to false so that the framework doesn't start them - since your bean is going to do so on a schedule.

I have never seen that exception; I have no idea what might cause it - you should ask the wider Kafka community; it really has nothing to do with Spring.