You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/09/06 10:47:10 UTC
[flink] branch release-1.14 updated: [FLINK-23839][kafka] Improve
warnings on InvalidTxnState-/ProducerFencedException
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new f675aca [FLINK-23839][kafka] Improve warnings on InvalidTxnState-/ProducerFencedException
f675aca is described below
commit f675acaaa89e75573525b4a45e6d9abfed115162
Author: David Moravek <dm...@apache.org>
AuthorDate: Mon Sep 6 12:45:35 2021 +0200
[FLINK-23839][kafka] Improve warnings on InvalidTxnState-/ProducerFencedException
---
docs/content/docs/connectors/datastream/kafka.md | 7 ++++
.../flink/connector/kafka/sink/KafkaCommitter.java | 27 +++++++++---
.../connectors/kafka/FlinkKafkaProducer.java | 48 +++++++++++++---------
3 files changed, 57 insertions(+), 25 deletions(-)
diff --git a/docs/content/docs/connectors/datastream/kafka.md b/docs/content/docs/connectors/datastream/kafka.md
index 58e0342..6b2cb94 100644
--- a/docs/content/docs/connectors/datastream/kafka.md
+++ b/docs/content/docs/connectors/datastream/kafka.md
@@ -543,4 +543,11 @@ It also can be circumvented by changing `retries` property in the producer setti
However this might cause reordering of messages,
which in turn if undesired can be circumvented by setting `max.in.flight.requests.per.connection` to 1.
+### ProducerFencedException
+
+The reason for this exception is most likely a transaction timeout on the broker side. With the implementation of
+[KAFKA-6119](https://issues.apache.org/jira/browse/KAFKA-6119), the `(producerId, epoch)` will be fenced off
+after a transaction timeout and all of its pending transactions are aborted (each `transactional.id` is
+mapped to a single `producerId`; this is described in more detail in the following [blog post](https://www.confluent.io/blog/simplified-robust-exactly-one-semantics-in-kafka-2-5/)).
+
{{< top >}}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
index 24d077d..8b27980 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
@@ -18,7 +18,9 @@
package org.apache.flink.connector.kafka.sink;
import org.apache.flink.api.connector.sink.Committer;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.errors.InvalidTxnStateException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.slf4j.Logger;
@@ -65,13 +67,26 @@ class KafkaCommitter implements Committer<KafkaCommittable> {
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (InvalidTxnStateException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
+ "Unable to commit recovered transaction ({}) because it's in an invalid state. "
+ + "Most likely the transaction has been aborted for some reason. Please check the Kafka logs for more details.",
+ committable,
+ e);
+ recyclable.ifPresent(Recyclable::close);
+ } catch (ProducerFencedException e) {
+ LOG.warn(
+ "Unable to commit recovered transaction ({}) because its producer is already fenced."
+ + " This means that you either have a different producer with the same '{}' (this is"
+ + " unlikely with the '{}' as all generated ids are unique and shouldn't be reused)"
+ + " or recovery took longer than '{}' ({}ms). In both cases this most likely signals data loss,"
+ + " please consult the Flink documentation for more details.",
+ committable,
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG,
+ KafkaSink.class.getSimpleName(),
+ ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
+ FlinkKafkaProducer.getTransactionTimeout(kafkaProducerConfig),
+ e);
recyclable.ifPresent(Recyclable::close);
} catch (Throwable e) {
LOG.warn("Cannot commit Kafka transaction, retrying.", e);
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index 4893760..5970158 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -733,19 +733,7 @@ public class FlinkKafkaProducer<IN>
// See KAFKA-6119 (affects versions 0.11.0.0 and 0.11.0.1):
// The KafkaProducer may not throw an exception if the transaction failed to commit
if (semantic == FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
- final Object object =
- this.producerConfig.get(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG);
- final long transactionTimeout;
- if (object instanceof String && StringUtils.isNumeric((String) object)) {
- transactionTimeout = Long.parseLong((String) object);
- } else if (object instanceof Number) {
- transactionTimeout = ((Number) object).longValue();
- } else {
- throw new IllegalArgumentException(
- ProducerConfig.TRANSACTION_TIMEOUT_CONFIG
- + " must be numeric, was "
- + object);
- }
+ final long transactionTimeout = getTransactionTimeout(producerConfig);
super.setTransactionTimeout(transactionTimeout);
super.enableTransactionTimeoutWarnings(0.8);
}
@@ -1052,13 +1040,23 @@ public class FlinkKafkaProducer<IN>
producer = initTransactionalProducer(transaction.transactionalId, false);
producer.resumeTransaction(transaction.producerId, transaction.epoch);
producer.commitTransaction();
- } catch (InvalidTxnStateException | ProducerFencedException ex) {
- // That means we have committed this transaction before.
+ } catch (InvalidTxnStateException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- ex,
- transaction);
+ "Unable to commit recovered transaction ({}) because it's in an invalid state. "
+ + "Most likely the transaction has been aborted for some reason. Please check the Kafka logs for more details.",
+ transaction,
+ e);
+ } catch (ProducerFencedException e) {
+ LOG.warn(
+ "Unable to commit recovered transaction ({}) because its producer is already fenced."
+ + " This means that you either have a different producer with the same '{}' or"
+ + " recovery took longer than '{}' ({}ms). In both cases this most likely signals data loss,"
+ + " please consult the Flink documentation for more details.",
+ transaction,
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG,
+ ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
+ getTransactionTimeout(producerConfig),
+ e);
} finally {
if (producer != null) {
producer.close(0, TimeUnit.SECONDS);
@@ -1492,6 +1490,18 @@ public class FlinkKafkaProducer<IN>
return partitions;
}
+ public static long getTransactionTimeout(Properties producerConfig) {
+ final Object object = producerConfig.get(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG);
+ if (object instanceof String && StringUtils.isNumeric((String) object)) {
+ return Long.parseLong((String) object);
+ } else if (object instanceof Number) {
+ return ((Number) object).longValue();
+ } else {
+ throw new IllegalArgumentException(
+ ProducerConfig.TRANSACTION_TIMEOUT_CONFIG + " must be numeric, was " + object);
+ }
+ }
+
/** State for handling transactions. */
@VisibleForTesting
@Internal