[FLINK-36569] flink kafka connector do not close kafka produer when it checkpoint success

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.19.0, 1.20.0
• Component/s: Connectors / Kafka

Environment

flink: 1.20

flink kafka connector: 3.3.0-1.20 

• Assignee: Unassigned
• Reporter: Jake.zhang
• Due: 18/Oct/24
• Created: 18/Oct/24 07:07
• Updated: 06/Dec/24 01:43

Description

flink kafka connector do't close FlinkKafkaInternalProducer when flink checkpoint success in flink 1.20/1.19 .  it will create one  FlinkKafkaInternalProducer per checkpoint.

 

FlinkKafkaInternalProducer do not close automatic. so  kafka producer network thread will more and more 

 it create `getRecoveryProducer` each time,   `recyclable` object always null, so  `recyclable.ifPresent(Recyclable::close)` not work.

`org.apache.flink.connector.kafka.sink.KafkaCommitter`

producer =
recyclable
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
producer.flush();
recyclable.ifPresent(Recyclable::close);