[FLINK-21228] Deadlock in KinesisProducer

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.12.1
• Component/s: Connectors / Kinesis
• Labels: auto-deprioritized-major auto-deprioritized-minor auto-unassigned
• Assignee: Unassigned
• Reporter: Danny Cranmer
• Created: 01/Feb/21 10:55
• Updated: 14/Oct/22 16:19

Description

Background
Kinesis sink failed and resulted in deadlock:

• Indefinite backpressure being applied
• Exception never thrown causing job to fail

Application running with:

flinkKinesisProducer.setQueueLimit(1);
flinkKinesisProducer.setFailOnError(true); 

• KinesisProducer is waiting for queue to empty before sending the next record (code)
• KPL ran out of memory, which raised an error, however this is processed async (code)
• KinesisProducer would have rethrown the error and restarted the job, however operator stuck in an infinite loop enforcing the queue limit (which never clears) (code)

Proposal

• checkAndPropagateAsyncError() while enforcing queue limit in enforceQueueLimit() to break deadlock