[FLINK-10953] InterruptedException on KafkaProducer

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Cannot Reproduce
• Affects Version/s: 1.6.2
• Component/s: Connectors / Kafka
• Assignee: Unassigned
• Reporter: Avi Levi
• Created: 20/Nov/18 16:39
• Updated: 02/Oct/19 17:45
• Resolved: 26/Dec/18 16:21

Description

I saw that this was handled in this [pr|http://example.com] however I am still getting this exception 

2018-11-20 17:52:40,577 ERROR org.apache.kafka.clients.producer.KafkaProducer - Interrupted while joining ioThread
java.lang.InterruptedException
at java.lang.Object.wait(Native Method)
at java.lang.Thread.join(Thread.java:1260)
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1031)
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1010)
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:989)
at org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer.close(FlinkKafkaProducer.java:168)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.close(FlinkKafkaProducer011.java:662)
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43)
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117)
at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:477)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:378)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
at java.lang.Thread.run(Thread.java:748)
2018-11-20 17:52:40,578 ERROR org.apache.flink.streaming.runtime.tasks.StreamTask - Error during disposal of stream operator.
org.apache.kafka.common.KafkaException: Failed to close kafka producer
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1062)
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1010)
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:989)
at org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer.close(FlinkKafkaProducer.java:168)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.close(FlinkKafkaProducer011.java:662)
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43)
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117)
at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:477)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:378)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.InterruptedException
at java.lang.Object.wait(Native Method)
at java.lang.Thread.join(Thread.java:1260)
at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1031)
... 10 more