[KAFKA-6127] Streams should never block infinitely

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.0.0
• Component/s: streams
• Labels: exactly-once
• Assignee: Unassigned
• Reporter: Matthias J. Sax
• Created: 25/Oct/17 23:51
• Updated: 18/Sep/20 22:06
• Resolved: 18/Sep/20 22:06

Description

Streams uses three consumer APIs that can block infinite: commitSync(), committed(), and position(). Also KafkaProducer#send() can block. If EOS is enabled, KafkaProducer#initTransactions() also used to block (fixed in KAFKA-6446) and we should double check the code if we handle this case correctly.

If we block within one operation, the whole StreamThread would block, and the instance does not make any progress, becomes unresponsive (for example, KafkaStreams#close() suffers), and we also might drop out of the consumer group.

Thanks to [KIP-266|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=75974886,] the Consumer now has non-blocking variants that we can use, but the same is not true of Producer. We can add non-blocking variants to Producer as well, or set the appropriate config options to set the max timeout.

Of course, we'd also need to be sure the catch the appropriate timeout exceptions.

Issue Links

• is related to: KAFKA-6446 KafkaProducer with transactionId endless waits when bootstrap server is down
• is related to: KAFKA-4879 KafkaConsumer.position may hang forever when deleting a topic
• relates to: KAFKA-6446 KafkaProducer with transactionId endless waits when bootstrap server is down
• relates to: KAFKA-7932 Streams needs to handle new Producer exceptions
• relates to: KAFKA-6705 producer.send() should not block due to metadata not available
• links to: GitHub Pull Request #5333