KafkaListener(Springboot 2.3.6 and kafka-client 2.5.1) got stuck, Not able to get any messages

[vimalsign]

Hi
I am using @KafkaListener(Springboot 2.3.6 and kafka-client 2.5.1) to consume the messages from topic with 30 partition (concurrency 30). This particular service is running in k8s cluster.

Here the issue is,
After starting the springboot application, KafkaListener is starting to consume the message from kafka. But 1 or 2 days later KafkaListener is not getting any messages (it is a kind of stuck). When we analysis the threaddump. WE found that heartbeat threads(30) are in blocked state.

Please assist me to solve this issue

[vimalsign]

Configuration
max.poll.records : 25
max.poll.interval.ms : 1200000 ms (20 mins)
session.timeout.ms : 30000 ms (30 seconds)

ThreadDump

"product_spring_aggregator_tunnelstats_groupid_prod-0-C-1" #22 prio=5 os_prio=0 cpu=642857.70ms elapsed=709536.40s tid=0x00007f5285017800 nid=0x19 runnable  [0x00007f523eef9000]
   java.lang.Thread.State: RUNNABLE
	at java.net.SocketInputStream.socketRead0(java.base@11.0.4/Native Method)
	at java.net.SocketInputStream.socketRead(java.base@11.0.4/Unknown Source)
	at java.net.SocketInputStream.read(java.base@11.0.4/Unknown Source)
	at java.net.SocketInputStream.read(java.base@11.0.4/Unknown Source)
	at com.mongodb.internal.connection.SocketStream.read(SocketStream.java:109)
	at com.mongodb.internal.connection.InternalStreamConnection.receiveResponseBuffers(InternalStreamConnection.java:579)
	at com.mongodb.internal.connection.InternalStreamConnection.receiveMessage(InternalStreamConnection.java:444)
	at com.mongodb.internal.connection.InternalStreamConnection.receiveCommandMessageResponse(InternalStreamConnection.java:298)
	at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:258)
	at com.mongodb.internal.connection.UsageTrackingInternalConnection.sendAndReceive(UsageTrackingInternalConnection.java:99)
	at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection.sendAndReceive(DefaultConnectionPool.java:500)
	at com.mongodb.internal.connection.CommandProtocolImpl.execute(CommandProtocolImpl.java:71)
	at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:224)
	at com.mongodb.internal.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:202)
	at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:118)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.executeCommand(MixedBulkWriteOperation.java:419)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.executeBulkWriteBatch(MixedBulkWriteOperation.java:245)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.access$700(MixedBulkWriteOperation.java:71)
	at com.mongodb.internal.operation.MixedBulkWriteOperation$1.call(MixedBulkWriteOperation.java:189)
	at com.mongodb.internal.operation.MixedBulkWriteOperation$1.call(MixedBulkWriteOperation.java:180)
	at com.mongodb.internal.operation.OperationHelper.withReleasableConnection(OperationHelper.java:500)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:180)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:71)
	at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.execute(MongoClientDelegate.java:207)
	at com.mongodb.client.internal.MongoCollectionImpl.executeSingleWriteRequest(MongoCollectionImpl.java:1008)
	at com.mongodb.client.internal.MongoCollectionImpl.executeInsertOne(MongoCollectionImpl.java:469)
	at com.mongodb.client.internal.MongoCollectionImpl.insertOne(MongoCollectionImpl.java:452)
	at com.mongodb.client.internal.MongoCollectionImpl.insertOne(MongoCollectionImpl.java:446)
	at org.springframework.data.mongodb.core.MongoTemplate.lambda$insertDocument$15(MongoTemplate.java:1442)
	at org.springframework.data.mongodb.core.MongoTemplate$$Lambda$1353/0x0000000100732840.doInCollection(Unknown Source)
	at org.springframework.data.mongodb.core.MongoTemplate.execute(MongoTemplate.java:566)
	at org.springframework.data.mongodb.core.MongoTemplate.insertDocument(MongoTemplate.java:1436)
	at org.springframework.data.mongodb.core.MongoTemplate.doInsert(MongoTemplate.java:1236)
	at org.springframework.data.mongodb.core.MongoTemplate.insert(MongoTemplate.java:1168)
	at com.test.product.aggregator.service.impl.TunnelStatsServiceImpl.lambda$saveTunnelStats$0(TunnelStatsServiceImpl.java:74)
	at com.test.product.aggregator.service.impl.TunnelStatsServiceImpl$$Lambda$1270/0x0000000100707440.accept(Unknown Source)
	at java.util.ArrayList.forEach(java.base@11.0.4/Unknown Source)
	at com.test.product.aggregator.service.impl.TunnelStatsServiceImpl.saveTunnelStats(TunnelStatsServiceImpl.java:71)
	at com.test.product.aggregator.listener.TunnelStatsListener.consumeTunnelStatsData(TunnelStatsListener.java:31)
	at jdk.internal.reflect.GeneratedMethodAccessor103.invoke(Unknown Source)
	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(java.base@11.0.4/Unknown Source)
	at java.lang.reflect.Method.invoke(java.base@11.0.4/Unknown Source)
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:171)
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:120)
	at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48)
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:329)
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:86)
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2004)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1986)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1923)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1863)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1757)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1478)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1141)
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1044)
	at java.util.concurrent.Executors$RunnableAdapter.call(java.base@11.0.4/Unknown Source)
	at java.util.concurrent.FutureTask.run(java.base@11.0.4/Unknown Source)
	at java.lang.Thread.run(java.base@11.0.4/Unknown Source)


"kafka-coordinator-heartbeat-thread | product_spring_aggregator_tunnelstats_groupid_prod" #94 daemon prio=5 os_prio=0 cpu=124064.52ms elapsed=709536.01s tid=0x00007f51fc01b000 nid=0x61 in Object.wait()  [0x00007f5179fe0000]
   java.lang.Thread.State: WAITING (on object monitor)
	at java.lang.Object.wait(java.base@11.0.4/Native Method)
	- waiting on <no object reference available>
	at java.lang.Object.wait(java.base@11.0.4/Unknown Source)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:1269)
	- waiting to re-lock in wait() <0x00000000f9596978> (a org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)

[garyrussell]

This has nothing to do with this project; it looks like your thread is stuck reading from mongo

Socket reads are not interruptible; I am not familiar with the mongo client; look to see if there is a socket timeout configuration property; socket reads don't timeout by default.

[vimalsign]

Hi garyrussell, Thanks for your input.

When I see the thread dump, all my consumers( product_spring_aggregator_tunnelstats_groupid_prod-0-C-1 ) are running fine,

But heartbeat-thread ( "kafka-coordinator-heartbeat-thread | product_spring_aggregator_tunnelstats_groupid_prod " ) is in waiting state.

I thought that, Since heartbeat-thread is not sending notification to kafka, So My Kafka Listener is stopped to consume messages.
(Nearly 3 million messages are pending to consume from topic "tunnel_topic")

I have added full threaddump and my kafkalistener code below.

Please assist me on this.

ThreadDump

threaddump.txt

My kafkaListner code

@KafkaListener(id = "product_spring_aggregator_tunnelstats_groupid_prod" , topics = "tunnel_topic", autoStartup = "true", 
                                                              concurrency="30", containerFactory = "kafkaListenerContainerFactory" )
public void consumeTunnelStatsData(@Payload String message, Acknowledgment acknowledgment) {
        boolean tunnelStatSaveFlag = tunnelStatsService.saveTunnelStats(message);

        messageCount.incrementAndGet();

        if(messageCount.get() % 3000 == 0){
            acknowledgment.acknowledge();
        }
}

Configuration
max.poll.records : 25
max.poll.interval.ms : 1200000 ms (20 mins)
session.timeout.ms : 30000 ms (30 seconds)

ThreadDump Report

[garyrussell]

Looks like the heartbeats are not enabled; the threads are stuck here

                        if (!enabled) {
                            AbstractCoordinator.this.wait();
                            continue;
                        }

Waiting for something to enable the thread.

We don't know kafka-clients internals here; you would be better asking the wider Kafka community.

[garyrussell]

The thread gets disabled under several circumstances, such as when a rebalance occurs

                        if (state != MemberState.STABLE) {
                            // the group is not stable (perhaps because we left the group or because the coordinator
                            // kicked us out), so disable heartbeats and wait for the main thread to rejoin.
                            disable();
                            continue;
                        }

I suggest you look in the logs to see if there are any hints.

I don't think we can help further here.

[vimalsign]

Thanks garyrussell for your input.