You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2018/11/16 16:56:24 UTC
[flink] branch release-1.5 updated: [FLINK-10455][Kafka Tx] Close
transactional producers in case of failure and termination (#6989)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.5
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.5 by this push:
new e493d83 [FLINK-10455][Kafka Tx] Close transactional producers in case of failure and termination (#6989)
e493d83 is described below
commit e493d8361e685a84af971aef2d889d8162569ca4
Author: azagrebin <az...@users.noreply.github.com>
AuthorDate: Wed Nov 14 10:08:39 2018 +0100
[FLINK-10455][Kafka Tx] Close transactional producers in case of failure and termination (#6989)
This commit addresses the problem of potential leak of resources associated with unclosed Kafka transactional producers in case of commitment failure or task shutdown.
1. always close producer even if commit fails in TwoPhaseCommitSinkFunction#notifyCheckpointComplete
2. close pending transactions in close method of Kafka Flink function in case of task shutdown
3. continue trying to commit other transactions in TwoPhaseCommitSinkFunction#notifyCheckpointComplete if any of them failed
---
.../connectors/kafka/FlinkKafkaProducer011.java | 11 ++++++++--
.../functions/sink/TwoPhaseCommitSinkFunction.java | 24 +++++++++++++++++++++-
2 files changed, 32 insertions(+), 3 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
index 8497372..fe383ad 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -45,6 +45,7 @@ import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartiti
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.IOUtils;
import org.apache.flink.util.NetUtils;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
@@ -671,6 +672,9 @@ public class FlinkKafkaProducer011<IN>
}
// make sure we propagate pending errors
checkErroneous();
+ pendingTransactions().forEach(transaction ->
+ IOUtils.closeQuietly(transaction.getValue().producer)
+ );
}
// ------------------- Logic for handling checkpoint flushing -------------------------- //
@@ -714,8 +718,11 @@ public class FlinkKafkaProducer011<IN>
protected void commit(KafkaTransactionState transaction) {
switch (semantic) {
case EXACTLY_ONCE:
- transaction.producer.commitTransaction();
- recycleTransactionalProducer(transaction.producer);
+ try {
+ transaction.producer.commitTransaction();
+ } finally {
+ recycleTransactionalProducer(transaction.producer);
+ }
break;
case AT_LEAST_ONCE:
case NONE:
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
index 2ffb6d5..6256262 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
@@ -38,20 +38,24 @@ import org.apache.flink.runtime.state.CheckpointListener;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.util.FlinkRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
import java.time.Clock;
+import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.stream.Stream;
import static java.util.Objects.requireNonNull;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -149,6 +153,12 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
return currentTransactionHolder == null ? null : currentTransactionHolder.handle;
}
+ @Nonnull
+ protected Stream<Map.Entry<Long, TXN>> pendingTransactions() {
+ return pendingCommitTransactions.entrySet().stream()
+ .map(e -> new AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().handle));
+ }
+
// ------ methods that should be implemented in child class to support two phase commit algorithm ------
/**
@@ -256,6 +266,7 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
Iterator<Map.Entry<Long, TransactionHolder<TXN>>> pendingTransactionIterator = pendingCommitTransactions.entrySet().iterator();
checkState(pendingTransactionIterator.hasNext(), "checkpoint completed, but no transaction pending");
+ Throwable firstError = null;
while (pendingTransactionIterator.hasNext()) {
Map.Entry<Long, TransactionHolder<TXN>> entry = pendingTransactionIterator.next();
@@ -269,12 +280,23 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
name(), checkpointId, pendingTransaction, pendingTransactionCheckpointId);
logWarningIfTimeoutAlmostReached(pendingTransaction);
- commit(pendingTransaction.handle);
+ try {
+ commit(pendingTransaction.handle);
+ } catch (Throwable t) {
+ if (firstError == null) {
+ firstError = t;
+ }
+ }
LOG.debug("{} - committed checkpoint transaction {}", name(), pendingTransaction);
pendingTransactionIterator.remove();
}
+
+ if (firstError != null) {
+ throw new FlinkRuntimeException("Committing one of transactions failed, logging first encountered failure",
+ firstError);
+ }
}
@Override