flink job进行cancel后kafka producer报错

Thu, 18 Nov 2021 18:01:29 -0800 

Hi All：
   有一个flink的运行正常的job进行cancel后，flink的log里面打印了一些信息，请问这种异常如何排查根因？

TY-APP-DATA-REAL-COMPUTATION - [2021-11-18 22:10:32.465] - INFO  
[RecordComputeOperator -> Sink: wi-data-sink (3/10)] 
org.apache.flink.runtime.taskmanager.Task  - RecordComputeOperator ->
Sink: wi-data-sink (3/10) (b2166a7da829182804f4557a16eabc58) switched from 
CANCELING to CANCELED.
TY-APP-DATA-REAL-COMPUTATION - [2021-11-18 22:10:32.466] - INFO  
[RecordComputeOperator -> Sink: wi-data-sink (9/10)] 
org.apache.kafka.clients.producer.KafkaProducer  - Closing the Kafka p
roducer with timeoutMillis = 9223372036854775807 ms.
TY-APP-DATA-REAL-COMPUTATION - [2021-11-18 22:10:32.467] - ERROR 
[RecordComputeOperator -> Sink: wi-data-sink (9/10)] 
org.apache.flink.streaming.runtime.tasks.StreamTask  - Error during di
sposal of stream operator.
org.apache.flink.streaming.connectors.kafka.FlinkKafka011Exception: Failed to 
send data to Kafka: Failed to close kafka producer
        at 
org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.checkErroneous(FlinkKafkaProducer011.java:1026)
        at 
org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.close(FlinkKafkaProducer011.java:691)
        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:651)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.cleanUpInvoke(StreamTask.java:562)
        at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:480)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
        at java.lang.Thread.run(Thread.java:745)
Caused by: 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:195)
        at 
org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.close(FlinkKafkaProducer011.java:674)
        ... 8 common frames omitted
Caused by: java.lang.InterruptedException: null
        at java.lang.Object.wait(Native Method)
        at java.lang.Thread.join(Thread.java:1253)
        at 
org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1031)
        ... 12 common frames omitted

回复