You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2025/04/04 11:34:50 UTC
(flink-connector-kafka) branch main updated: [FLINK-37613] Fix resource leak during abortion
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new a84c2b8a [FLINK-37613] Fix resource leak during abortion
a84c2b8a is described below
commit a84c2b8a768392d7974523ae875497e83e6fdfdc
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Thu Apr 3 15:45:23 2025 +0200
[FLINK-37613] Fix resource leak during abortion
Transaction abortion can take a while. Apparently, task cleanup may not trigger correctly if this abortion is interrupted (for some unrelated reason). This leaks producers. The fix is to close() manually during initialization of the EOS writer until we fixed the Flink bug.
---
.../kafka/sink/ExactlyOnceKafkaWriter.java | 16 ++++++--
.../flink/connector/kafka/sink/KafkaCommitter.java | 5 +++
.../flink/connector/kafka/sink/KafkaWriter.java | 14 ++++++-
.../sink/FlinkKafkaInternalProducerITCase.java | 48 ++++++++++++++++++++++
4 files changed, 78 insertions(+), 5 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java
index ed5b95e9..741d00f7 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java
@@ -131,9 +131,19 @@ class ExactlyOnceKafkaWriter<IN> extends KafkaWriter<IN> {
@Override
public void initialize() {
- abortLingeringTransactions(
- checkNotNull(recoveredStates, "recoveredStates"), restoredCheckpointId + 1);
- this.currentProducer = startTransaction(restoredCheckpointId + 1);
+ // Workaround for FLINK-37612: ensure that we are not leaking producers
+ try {
+ abortLingeringTransactions(
+ checkNotNull(recoveredStates, "recoveredStates"), restoredCheckpointId + 1);
+ this.currentProducer = startTransaction(restoredCheckpointId + 1);
+ } catch (Throwable t) {
+ try {
+ close();
+ } catch (Exception e) {
+ t.addSuppressed(e);
+ }
+ throw t;
+ }
}
private FlinkKafkaInternalProducer<byte[], byte[]> startTransaction(long checkpointId) {
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
index 9ba6de1f..70d67150 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
@@ -139,6 +139,7 @@ class KafkaCommitter implements Committer<KafkaCommittable>, Closeable {
"Transaction ({}) encountered error and data has been potentially lost.",
request,
e);
+ closeCommitterProducer(producer);
// cause failover
request.signalFailedWithUnknownReason(e);
}
@@ -150,6 +151,10 @@ class KafkaCommitter implements Committer<KafkaCommittable>, Closeable {
return;
}
backchannel.send(TransactionFinished.erroneously(producer.getTransactionalId()));
+ closeCommitterProducer(producer);
+ }
+
+ private void closeCommitterProducer(FlinkKafkaInternalProducer<?, ?> producer) {
if (producer == this.committingProducer) {
this.committingProducer.close();
this.committingProducer = null;
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index 83a400d7..8e0bf3ba 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -148,8 +148,18 @@ class KafkaWriter<IN>
}
public void initialize() {
- this.currentProducer = new FlinkKafkaInternalProducer<>(this.kafkaProducerConfig);
- initKafkaMetrics(this.currentProducer);
+ // Workaround for FLINK-37612: ensure that we are not leaking producers
+ try {
+ this.currentProducer = new FlinkKafkaInternalProducer<>(this.kafkaProducerConfig);
+ initKafkaMetrics(this.currentProducer);
+ } catch (Throwable t) {
+ try {
+ close();
+ } catch (Exception e) {
+ t.addSuppressed(e);
+ }
+ throw t;
+ }
}
@Override
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
index efd92dc9..6bd6d880 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
@@ -34,11 +34,13 @@ import org.apache.kafka.common.errors.InvalidTxnStateException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
+import org.assertj.core.api.AbstractThrowableAssert;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.junit.jupiter.Container;
@@ -54,6 +56,7 @@ import java.util.stream.Collectors;
import static org.apache.flink.connector.kafka.testutils.KafkaUtil.checkProducerLeak;
import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@Testcontainers
@@ -174,6 +177,51 @@ class FlinkKafkaInternalProducerITCase {
}
}
+ @ParameterizedTest
+ @CsvSource({"true,true", "true,false", "false,true", "false,false"})
+ void testDoubleCommitAndAbort(boolean firstCommit, boolean secondCommit) {
+ final String topic = "test-double-commit-transaction-" + firstCommit + secondCommit;
+ final String transactionIdPrefix = "testDoubleCommitTransaction-";
+ final String transactionalId = transactionIdPrefix + "id";
+
+ KafkaCommittable committable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(topic, "test-value"));
+ producer.flush();
+ committable = KafkaCommittable.of(producer);
+ if (firstCommit) {
+ producer.commitTransaction();
+ } else {
+ producer.abortTransaction();
+ }
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ resumedProducer.resumeTransaction(committable.getProducerId(), committable.getEpoch());
+ AbstractThrowableAssert<?, ? extends Throwable> secondOp =
+ assertThatCode(
+ () -> {
+ if (secondCommit) {
+ resumedProducer.commitTransaction();
+ } else {
+ resumedProducer.abortTransaction();
+ }
+ });
+ if (firstCommit == secondCommit) {
+ secondOp.doesNotThrowAnyException();
+ } else {
+ secondOp.isInstanceOf(InvalidTxnStateException.class);
+ }
+ }
+
+ assertNumTransactions(1, transactionIdPrefix);
+ assertThat(readRecords(topic).count()).isEqualTo(firstCommit ? 1 : 0);
+ }
+
private static Properties getProperties() {
Properties properties = new Properties();
properties.put(