[KAFKA-10790] Detect/Prevent Deadlock on Producer Network Thread

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 2.6.0, 2.7.0
• Component/s: clients
• Assignee: GeordieMai
• Reporter: Gary Russell
• Created: 01/Dec/20 14:42
• Updated: 07/Jan/21 11:54

Description

I realize this is contrived, but I stumbled across the problem while testing some library code with 2.7.0 RC3 (although the issue is not limited to 2.7).

For example, calling flush() on the producer callback deadlocks the network thread (and any attempt to close the producer thereafter).

producer.send(new ProducerRecord("foo", "bar"), (rm, ex) -> {
	producer.flush();
});
Thread.sleep(1000);
producer.close();

It took some time to figure out why the close was blocked.

There is existing logic in close() to avoid it blocking if called from the callback; perhaps similar logic could be added to flush() (and any other methods that might block), even if it means throwing an exception to make it clear that you can't call flush() from the callback. 

These stack traces are with the 2.6.0 client.

"main" #1 prio=5 os_prio=31 cpu=1333.10ms elapsed=13.05s tid=0x00007ff259012800 nid=0x2803 in Object.wait()  [0x000070000fda5000]
   java.lang.Thread.State: TIMED_WAITING (on object monitor)
	at java.lang.Object.wait(java.base@14.0.2/Native Method)
	- waiting on <0x0000000700d00000> (a org.apache.kafka.common.utils.KafkaThread)
	at java.lang.Thread.join(java.base@14.0.2/Thread.java:1297)
	- locked <0x0000000700d00000> (a org.apache.kafka.common.utils.KafkaThread)
	at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1205)
	at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1182)
	at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1158)
	at com.example.demo.Rk1Application.lambda$2(Rk1Application.java:55)


"kafka-producer-network-thread | producer-1" #24 daemon prio=5 os_prio=31 cpu=225.80ms elapsed=11.64s tid=0x00007ff256963000 nid=0x7103 waiting on condition  [0x0000700011d04000]
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(java.base@14.0.2/Native Method)
	- parking to wait for  <0x00000007020b27e0> (a java.util.concurrent.CountDownLatch$Sync)
	at java.util.concurrent.locks.LockSupport.park(java.base@14.0.2/LockSupport.java:211)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(java.base@14.0.2/AbstractQueuedSynchronizer.java:714)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(java.base@14.0.2/AbstractQueuedSynchronizer.java:1046)
	at java.util.concurrent.CountDownLatch.await(java.base@14.0.2/CountDownLatch.java:232)
	at org.apache.kafka.clients.producer.internals.ProduceRequestResult.await(ProduceRequestResult.java:76)
	at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:712)
	at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1111)
	at com.example.demo.Rk1Application.lambda$3(Rk1Application.java:52)
	at com.example.demo.Rk1Application$$Lambda$528/0x0000000800e28840.onCompletion(Unknown Source)
	at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1363)
	at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:228)
	at org.apache.kafka.clients.producer.internals.ProducerBatch.done(ProducerBatch.java:197)
	at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:653)
	at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:634)
	at org.apache.kafka.clients.producer.internals.Sender.handleProduceResponse(Sender.java:554)
	at org.apache.kafka.clients.producer.internals.Sender.lambda$sendProduceRequest$0(Sender.java:743)
	at org.apache.kafka.clients.producer.internals.Sender$$Lambda$642/0x0000000800ea2040.onComplete(Unknown Source)
	at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
	at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:566)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:558)
	at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:325)
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
	at java.lang.Thread.run(java.base@14.0.2/Thread.java:832)

Issue Links

• links to: GitHub Pull Request #9707