Kafka Streams producer thread hung or infinite wait state

[nsnmurthyk]

We noticed an issue where one producer thread is hung or infinite wait.

KAFKA: strimzi/kafka:0.22.1-kafka-2.5.0

KAFKA Client: 3.6.2

We recovered the same by restarting our application. Please let me know your comments on this.

java.lang.Thread.State: TIMED_WAITING (on object monitor) at java.lang.Object.wait(java.base@17.0.8.1/Native Method)

• waiting on at org.apache.kafka.common.utils.SystemTime.waitObject(SystemTime.java:55)
• locked <0x00001000ad72c348> (a org.apache.kafka.clients.producer.internals.ProducerMetadata) at org.apache.kafka.clients.producer.internals.ProducerMetadata.awaitUpdate(ProducerMetadata.java:119)
• locked <0x00001000ad72c348> (a org.apache.kafka.clients.producer.internals.ProducerMetadata) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:1132) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:981) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:947) at org.apache.kafka.streams.processor.internals.StreamsProducer.send(StreamsProducer.java:260) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:253) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:175) at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:85) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:215) at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:44) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:154) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:48) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:152) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:215) at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:44) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:154) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:215) at org.apache.kafka.streams.kstream.internals.KStreamFlatMapValues$KStreamFlatMapValuesProcessor.process(KStreamFlatMapValues.java:44) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:154) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:290) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:269) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:228) at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84) at org.apache.kafka.streams.processor.internals.StreamTask.lambda$doProcess$1(StreamTask.java:810) at org.apache.kafka.streams.processor.internals.StreamTask$$Lambda$2360/0x00007f1eb7106af8.run(Unknown Source) at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:872) at org.apache.kafka.streams.processor.internals.StreamTask.doProcess(StreamTask.java:810) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:741) at org.apache.kafka.streams.processor.internals.TaskExecutor.processTask(TaskExecutor.java:97) at org.apache.kafka.streams.processor.internals.TaskExecutor.process(TaskExecutor.java:78) at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1765) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:807) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:617) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:579)

[quarkus-bot[bot]]

/cc @alesj (kafka), @cescoffier (kafka), @ozangunalp (kafka)

[ozangunalp]

I don't think this is Quarkus-related. The producer blocking timeout can be configured with max.block.ms property, and by default is 60 seconds. I don't know how it can block there indefinitely.