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 2021/09/07 10:18:23 UTC
[flink] branch master updated (e55e664 -> 85684e4)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from e55e664 [FLINK-23614][table-planner] The resulting scale of TRUNCATE(DECIMAL) is not correct
new da0e7dd [FLINK-24131][tests] Harden leak check in KafkaSinkITCase.
new d84671c [FLINK-24131][datastream] Ensure that SinkWriter#prepareCommit is not called twice for final commit.
new b6da334 [FLINK-24131][datastream] Recommit recovered transactions as quickly as possible.
new 184198c [FLINK-24131][datastream] Ensure writer and committer are closed correctly even with Interruptions.
new 3bf4a01 [FLINK-24131][connectors/kafka] Ensure kafka writer and producer closed correctly even with Interruptions.
new ad9903e [FLINK-24131][runtime] Introduce CheckpointIDCounter#INITIAL_CHECKPOINT_ID.
new 94f000a [FLINK-24131][connectors/kafka] Improve debuggability of KafkaWriter.
new c8ee1b1 [FLINK-24131][connectors/kafka] Improve handling of committer errors in KafkaCommitter.
new 93c7104 [FLINK-24131][connectors/kafka] Improve threading model of KafkaWriter.
new 346172c [FLINK-24131][connectors/kafka] Fix KafkaWriter currentSendTime metric.
new 85684e4 [FLINK-24151][connectors/kafka] Add concurrent checkpoint test to KafkaSinkITCase.
The 11 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../kafka/sink/FlinkKafkaInternalProducer.java | 31 ++++-
.../flink/connector/kafka/sink/KafkaCommitter.java | 53 ++++---
.../flink/connector/kafka/sink/KafkaWriter.java | 155 +++++++++++----------
.../flink/connector/kafka/sink/Recyclable.java | 6 +-
.../connector/kafka/sink/KafkaCommitterTest.java | 62 ++++++++-
.../connector/kafka/sink/KafkaSinkITCase.java | 67 +++++++--
.../connector/kafka/sink/KafkaWriterITCase.java | 13 +-
.../KubernetesCheckpointIDCounter.java | 3 +-
.../runtime/checkpoint/CheckpointIDCounter.java | 1 +
.../checkpoint/StandaloneCheckpointIDCounter.java | 2 +-
.../checkpoint/ZooKeeperCheckpointIDCounter.java | 2 +-
.../runtime/operators/sink/CommitterOperator.java | 6 +-
.../runtime/operators/sink/SinkOperator.java | 16 ++-
13 files changed, 292 insertions(+), 125 deletions(-)
[flink] 05/11: [FLINK-24131][connectors/kafka] Ensure kafka writer
and producer closed correctly even with Interruptions.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3bf4a01a96497c34f1af4716a0ff684f5513d280
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:48:01 2021 +0200
[FLINK-24131][connectors/kafka] Ensure kafka writer and producer closed correctly even with Interruptions.
---
.../kafka/sink/FlinkKafkaInternalProducer.java | 12 ++++++++--
.../flink/connector/kafka/sink/KafkaWriter.java | 26 +++++++++++++++++-----
2 files changed, 30 insertions(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
index c04f73e..4856222 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
@@ -101,8 +101,16 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
@Override
public void close() {
closed = true;
- flush();
- super.close(Duration.ZERO);
+ if (inTransaction) {
+ // This is state is most likely reached in case of a failure.
+ // If this producer is still in transaction, it should be committing.
+ // However, at this point, we cannot decide that and we shouldn't prolong cancellation.
+ // So hard kill this producer with all resources.
+ super.close(Duration.ZERO);
+ } else {
+ // If this is outside of a transaction, we should be able to cleanly shutdown.
+ super.close(Duration.ofHours(1));
+ }
}
@Override
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index 8be56f8..b13a51f 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -38,6 +38,7 @@ import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.errors.ProducerFencedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,6 +55,7 @@ import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
+import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -206,14 +208,26 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
@Override
public void close() throws Exception {
+ closed = true;
+ closeAll(
+ this::abortCurrentProducer,
+ closer,
+ producerPool::clear,
+ () -> {
+ checkState(currentProducer.isClosed());
+ currentProducer = null;
+ });
+ }
+
+ private void abortCurrentProducer() {
if (currentProducer.isInTransaction()) {
- currentProducer.abortTransaction();
+ try {
+ currentProducer.abortTransaction();
+ } catch (ProducerFencedException e) {
+ LOG.debug(
+ "Producer {} fenced while aborting", currentProducer.getTransactionalId());
+ }
}
- closed = true;
- closer.close();
- producerPool.clear();
- checkState(currentProducer.isClosed());
- currentProducer = null;
}
@VisibleForTesting
[flink] 08/11: [FLINK-24131][connectors/kafka] Improve handling of
committer errors in KafkaCommitter.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c8ee1b118c625c7653a45c9e7adb9f1def6ea2b7
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sun Sep 5 23:19:46 2021 +0200
[FLINK-24131][connectors/kafka] Improve handling of committer errors in KafkaCommitter.
---
.../flink/connector/kafka/sink/KafkaCommitter.java | 53 ++++++++++++------
.../flink/connector/kafka/sink/Recyclable.java | 6 ++-
.../connector/kafka/sink/KafkaCommitterTest.java | 62 +++++++++++++++++++---
3 files changed, 96 insertions(+), 25 deletions(-)
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 8b27980..b511126 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,16 +18,18 @@
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.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.UnknownProducerIdException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -39,9 +41,12 @@ import java.util.Properties;
*
* <p>The committer is responsible to finalize the Kafka transactions by committing them.
*/
-class KafkaCommitter implements Committer<KafkaCommittable> {
+class KafkaCommitter implements Committer<KafkaCommittable>, Closeable {
private static final Logger LOG = LoggerFactory.getLogger(KafkaCommitter.class);
+ public static final String UNKNOWN_PRODUCER_ID_ERROR_MESSAGE =
+ "because of a bug in the Kafka broker (KAFKA-9310). Please upgrade to Kafka 2.5+. If you are running with concurrent checkpoints, you also may want to try without them.\n"
+ + "To avoid data loss, the application will restart.";
private final Properties kafkaProducerConfig;
@@ -66,17 +71,16 @@ class KafkaCommitter implements Committer<KafkaCommittable> {
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (InvalidTxnStateException e) {
+ producer.flush();
+ } catch (RetriableException e) {
LOG.warn(
- "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);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
+ retryableCommittables.add(committable);
+ continue;
} catch (ProducerFencedException e) {
- LOG.warn(
- "Unable to commit recovered transaction ({}) because its producer is already fenced."
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Unable to commit 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,"
@@ -85,19 +89,34 @@ class KafkaCommitter implements Committer<KafkaCommittable> {
ProducerConfig.TRANSACTIONAL_ID_CONFIG,
KafkaSink.class.getSimpleName(),
ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
- FlinkKafkaProducer.getTransactionTimeout(kafkaProducerConfig),
+ kafkaProducerConfig.getProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG),
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Unable to commit 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);
+ } catch (UnknownProducerIdException e) {
+ LOG.error(
+ "Unable to commit transaction ({}) " + UNKNOWN_PRODUCER_ID_ERROR_MESSAGE,
+ committable,
+ e);
+ } catch (Exception e) {
+ LOG.error(
+ "Transaction ({}) encountered error and data has been potentially lost.",
+ committable,
e);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
- retryableCommittables.add(committable);
}
+ recyclable.ifPresent(Recyclable::close);
}
return retryableCommittables;
}
@Override
- public void close() throws Exception {
+ public void close() {
if (recoveryProducer != null) {
recoveryProducer.close();
}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/Recyclable.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/Recyclable.java
index 163592a..012fa99 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/Recyclable.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/Recyclable.java
@@ -33,10 +33,14 @@ class Recyclable<T> implements Closeable {
}
public T getObject() {
- checkState(object != null, "Already recycled");
+ checkState(!isRecycled(), "Already recycled");
return object;
}
+ boolean isRecycled() {
+ return object == null;
+ }
+
@Override
public void close() {
recycler.accept(object);
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaCommitterTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaCommitterTest.java
index 62f0cec..6ce4f24 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaCommitterTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaCommitterTest.java
@@ -19,6 +19,9 @@ package org.apache.flink.connector.kafka.sink;
import org.apache.flink.util.TestLogger;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;
import java.io.IOException;
@@ -26,17 +29,62 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
-import static org.junit.Assert.assertEquals;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
/** Tests for {@link KafkaCommitter}. */
public class KafkaCommitterTest extends TestLogger {
+ private static final int PRODUCER_ID = 0;
+ private static final short EPOCH = 0;
+ private static final String TRANSACTIONAL_ID = "transactionalId";
+
+ /** Causes a network error by inactive broker and tests that a retry will happen. */
+ @Test
+ public void testRetryCommittableOnRetriableError() throws IOException {
+ Properties properties = getProperties();
+ try (final KafkaCommitter committer = new KafkaCommitter(properties);
+ FlinkKafkaInternalProducer<Object, Object> producer =
+ new FlinkKafkaInternalProducer<>(properties, TRANSACTIONAL_ID);
+ Recyclable<FlinkKafkaInternalProducer<Object, Object>> recyclable =
+ new Recyclable<>(producer, p -> {})) {
+ final List<KafkaCommittable> committables =
+ Collections.singletonList(
+ new KafkaCommittable(PRODUCER_ID, EPOCH, TRANSACTIONAL_ID, recyclable));
+ producer.resumeTransaction(PRODUCER_ID, EPOCH);
+ List<KafkaCommittable> recovered = committer.commit(committables);
+ assertThat(recovered, contains(committables.toArray()));
+ assertThat(recyclable.isRecycled(), equalTo(false));
+ }
+ }
+
@Test
- public void testRetryCommittableOnFailure() throws IOException {
- final KafkaCommitter committer = new KafkaCommitter(new Properties());
- final short epoch = 0;
- final List<KafkaCommittable> committables =
- Collections.singletonList(new KafkaCommittable(0, epoch, "transactionalId", null));
- assertEquals(committables, committer.commit(committables));
+ public void testRetryCommittableOnFatalError() throws IOException {
+ Properties properties = getProperties();
+ try (final KafkaCommitter committer = new KafkaCommitter(properties);
+ FlinkKafkaInternalProducer<Object, Object> producer =
+ new FlinkKafkaInternalProducer(properties, TRANSACTIONAL_ID);
+ Recyclable<FlinkKafkaInternalProducer<Object, Object>> recyclable =
+ new Recyclable<>(producer, p -> {})) {
+ final List<KafkaCommittable> committables =
+ Collections.singletonList(
+ new KafkaCommittable(PRODUCER_ID, EPOCH, TRANSACTIONAL_ID, recyclable));
+ // will fail because transaction not started
+ List<KafkaCommittable> recovered = committer.commit(committables);
+ assertThat(recovered, empty());
+ assertThat(recyclable.isRecycled(), equalTo(true));
+ }
+ }
+
+ Properties getProperties() {
+ Properties properties = new Properties();
+ properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "http://localhost:1");
+ // Low timeout will fail commitTransaction quicker
+ properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "100");
+ properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ return properties;
}
}
[flink] 10/11: [FLINK-24131][connectors/kafka] Fix KafkaWriter
currentSendTime metric.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 346172cf986454ed70eb88fa9dc4feffa034ac34
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sun Sep 5 22:08:07 2021 +0200
[FLINK-24131][connectors/kafka] Fix KafkaWriter currentSendTime metric.
The metric can only be registered once and should simply take the currentProducer to calculate.
---
.../flink/connector/kafka/sink/KafkaWriter.java | 29 ++++++++++++++--------
1 file changed, 19 insertions(+), 10 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index 95a096f..1540e62 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -36,7 +36,6 @@ import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
import org.apache.flink.shaded.guava30.com.google.common.io.Closer;
import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Metric;
@@ -74,6 +73,7 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
private static final String KEY_DISABLE_METRICS = "flink.disable-metrics";
private static final String KEY_REGISTER_METRICS = "register.producer.metrics";
+ private static final String KAFKA_PRODUCER_METRICS = "producer-metrics";
private final DeliveryGuarantee deliveryGuarantee;
private final Properties kafkaProducerConfig;
@@ -163,12 +163,13 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
|| deliveryGuarantee == DeliveryGuarantee.NONE) {
this.currentProducer = new FlinkKafkaInternalProducer<>(this.kafkaProducerConfig, null);
closer.register(this.currentProducer);
- initMetrics(this.currentProducer);
+ initKafkaMetrics(this.currentProducer);
} else {
throw new UnsupportedOperationException(
"Unsupported Kafka writer semantic " + this.deliveryGuarantee);
}
- registerMetricSync();
+
+ initFlinkMetrics();
}
@Override
@@ -298,18 +299,22 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
producer = new FlinkKafkaInternalProducer<>(kafkaProducerConfig, transactionalId);
closer.register(producer);
producer.initTransactions();
- initMetrics(producer);
+ initKafkaMetrics(producer);
} else {
producer.initTransactionId(transactionalId);
}
return producer;
}
- private void initMetrics(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ private void initFlinkMetrics() {
+ metricGroup.setCurrentSendTimeGauge(this::computeSendTime);
+ registerMetricSync();
+ }
+
+ private void initKafkaMetrics(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
byteOutMetric =
MetricUtil.getKafkaMetric(
- producer.metrics(), "producer-metrics", "outgoing-byte-total");
- metricGroup.setCurrentSendTimeGauge(() -> computeSendTime(producer));
+ producer.metrics(), KAFKA_PRODUCER_METRICS, "outgoing-byte-total");
if (disabledMetrics) {
return;
}
@@ -333,13 +338,17 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
};
}
- private static long computeSendTime(Producer<?, ?> producer) {
+ private long computeSendTime() {
+ FlinkKafkaInternalProducer<byte[], byte[]> producer = this.currentProducer;
+ if (producer == null) {
+ return -1L;
+ }
final Metric sendTime =
MetricUtil.getKafkaMetric(
- producer.metrics(), "producer-metrics", "request-latency-avg");
+ producer.metrics(), KAFKA_PRODUCER_METRICS, "request-latency-avg");
final Metric queueTime =
MetricUtil.getKafkaMetric(
- producer.metrics(), "producer-metrics", "record-queue-time-avg");
+ producer.metrics(), KAFKA_PRODUCER_METRICS, "record-queue-time-avg");
return ((Number) sendTime.metricValue()).longValue()
+ ((Number) queueTime.metricValue()).longValue();
}
[flink] 03/11: [FLINK-24131][datastream] Recommit recovered
transactions as quickly as possible.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b6da3342a15457ee4934ce1501c690c45ea80300
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:35:41 2021 +0200
[FLINK-24131][datastream] Recommit recovered transactions as quickly as possible.
---
.../flink/streaming/runtime/operators/sink/CommitterOperator.java | 2 ++
.../org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java | 1 +
2 files changed, 3 insertions(+)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index 125ec06..f72f1de 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -72,6 +72,8 @@ class CommitterOperator<InputT, OutputT> extends AbstractStreamOperator<byte[]>
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
committerHandler.initializeState(context);
+ // try to re-commit recovered transactions as quickly as possible
+ commitRetrier.retryWithDelay();
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
index 015e90c..dec4036 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
@@ -149,6 +149,7 @@ class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<b
checkpointId.isPresent() ? checkpointId.getAsLong() : null),
sinkWriterStateHandler.initializeState(context));
committerHandler.initializeState(context);
+ commitRetrier.retryWithDelay();
}
@Override
[flink] 04/11: [FLINK-24131][datastream] Ensure writer and
committer are closed correctly even with Interruptions.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 184198cfa276dfdc42f45069d8463631e3404646
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:44:33 2021 +0200
[FLINK-24131][datastream] Ensure writer and committer are closed correctly even with Interruptions.
---
.../flink/streaming/runtime/operators/sink/CommitterOperator.java | 4 ++--
.../apache/flink/streaming/runtime/operators/sink/SinkOperator.java | 5 ++---
2 files changed, 4 insertions(+), 5 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index f72f1de..ee0a436 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -29,6 +29,7 @@ import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import java.util.Collections;
+import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -106,7 +107,6 @@ class CommitterOperator<InputT, OutputT> extends AbstractStreamOperator<byte[]>
@Override
public void close() throws Exception {
- committerHandler.close();
- super.close();
+ closeAll(committerHandler, super::close);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
index dec4036..72108aa 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
@@ -48,6 +48,7 @@ import java.util.Collection;
import java.util.List;
import java.util.OptionalLong;
+import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -212,9 +213,7 @@ class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<b
@Override
public void close() throws Exception {
- committerHandler.close();
- sinkWriter.close();
- super.close();
+ closeAll(committerHandler, sinkWriter, super::close);
}
private Sink.InitContext createInitContext(@Nullable Long restoredCheckpointId) {
[flink] 06/11: [FLINK-24131][runtime] Introduce
CheckpointIDCounter#INITIAL_CHECKPOINT_ID.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ad9903eb088709fced1451be508785b75a9efce6
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Mon Sep 6 13:50:55 2021 +0200
[FLINK-24131][runtime] Introduce CheckpointIDCounter#INITIAL_CHECKPOINT_ID.
---
.../kubernetes/highavailability/KubernetesCheckpointIDCounter.java | 3 +--
.../java/org/apache/flink/runtime/checkpoint/CheckpointIDCounter.java | 1 +
.../apache/flink/runtime/checkpoint/StandaloneCheckpointIDCounter.java | 2 +-
.../apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java | 2 +-
4 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java
index c55508e..ede797e 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java
@@ -163,8 +163,7 @@ public class KubernetesCheckpointIDCounter implements CheckpointIDCounter {
if (configMap.getData().containsKey(CHECKPOINT_COUNTER_KEY)) {
return Long.valueOf(configMap.getData().get(CHECKPOINT_COUNTER_KEY));
} else {
- // Initial checkpoint id
- return 1;
+ return INITIAL_CHECKPOINT_ID;
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointIDCounter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointIDCounter.java
index b2f2211..81e0471 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointIDCounter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointIDCounter.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobStatus;
/** A checkpoint ID counter. */
public interface CheckpointIDCounter {
+ int INITIAL_CHECKPOINT_ID = 1;
/** Starts the {@link CheckpointIDCounter} service down. */
void start() throws Exception;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointIDCounter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointIDCounter.java
index e2c0c64..30b6539 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointIDCounter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointIDCounter.java
@@ -31,7 +31,7 @@ import java.util.concurrent.atomic.AtomicLong;
*/
public class StandaloneCheckpointIDCounter implements CheckpointIDCounter {
- private final AtomicLong checkpointIdCounter = new AtomicLong(1);
+ private final AtomicLong checkpointIdCounter = new AtomicLong(INITIAL_CHECKPOINT_ID);
@Override
public void start() throws Exception {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java
index 1e4180a..c2806b8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java
@@ -83,7 +83,7 @@ public class ZooKeeperCheckpointIDCounter implements CheckpointIDCounter {
CuratorFramework client, LastStateConnectionStateListener connectionStateListener) {
this.client = checkNotNull(client, "Curator client");
this.counterPath = ZooKeeperUtils.getCheckpointIdCounterPath();
- this.sharedCount = new SharedCount(client, counterPath, 1);
+ this.sharedCount = new SharedCount(client, counterPath, INITIAL_CHECKPOINT_ID);
this.connectionStateListener = connectionStateListener;
}
[flink] 07/11: [FLINK-24131][connectors/kafka] Improve
debuggability of KafkaWriter.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 94f000a9331d05535deea5361febd0188066e966
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:49:24 2021 +0200
[FLINK-24131][connectors/kafka] Improve debuggability of KafkaWriter.
This also aligns the transaction ids with checkpoint ids starting at 1.
---
.../kafka/sink/FlinkKafkaInternalProducer.java | 19 +++++++++++++++++++
.../flink/connector/kafka/sink/KafkaWriter.java | 9 +++++++--
.../flink/connector/kafka/sink/KafkaWriterITCase.java | 10 +++++-----
3 files changed, 31 insertions(+), 7 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
index 4856222..aec1edf 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
@@ -84,12 +84,16 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
@Override
public void abortTransaction() throws ProducerFencedException {
+ LOG.debug("abortTransaction {}", transactionalId);
+ checkState(inTransaction, "Transaction was not started");
super.abortTransaction();
inTransaction = false;
}
@Override
public void commitTransaction() throws ProducerFencedException {
+ LOG.debug("commitTransaction {}", transactionalId);
+ checkState(inTransaction, "Transaction was not started");
super.commitTransaction();
inTransaction = false;
}
@@ -156,6 +160,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
public void setTransactionId(String transactionalId) {
if (!transactionalId.equals(this.transactionalId)) {
checkState(!inTransaction);
+ LOG.debug("Change transaction id from {} to {}", this.transactionalId, transactionalId);
Object transactionManager = getTransactionManager();
synchronized (transactionManager) {
setField(transactionManager, "transactionalId", transactionalId);
@@ -274,6 +279,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
* https://github.com/apache/kafka/commit/5d2422258cb975a137a42a4e08f03573c49a387e#diff-f4ef1afd8792cd2a2e9069cd7ddea630
*/
public void resumeTransaction(long producerId, short epoch) {
+ checkState(!inTransaction, "Already in transaction %s", transactionalId);
checkState(
producerId >= 0 && epoch >= 0,
"Incorrect values for producerId %s and epoch %s",
@@ -302,6 +308,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
transitionTransactionManagerStateTo(transactionManager, "IN_TRANSACTION");
setField(transactionManager, "transactionStarted", true);
+ this.inTransaction = true;
}
}
@@ -358,4 +365,16 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
Object transactionManager, String state) {
invoke(transactionManager, "transitionTo", getTransactionManagerState(state));
}
+
+ @Override
+ public String toString() {
+ return "FlinkKafkaInternalProducer{"
+ + "transactionalId='"
+ + transactionalId
+ + "', inTransaction="
+ + inTransaction
+ + ", closed="
+ + closed
+ + '}';
+ }
}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index b13a51f..dfe17f6 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -26,6 +26,7 @@ import org.apache.flink.connector.kafka.MetricUtil;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
import org.apache.flink.streaming.connectors.kafka.internals.metrics.KafkaMetricMutableWrapper;
import org.apache.flink.util.FlinkRuntimeException;
@@ -157,7 +158,10 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
}
this.kafkaWriterState = new KafkaWriterState(transactionalIdPrefix);
- this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(-1);
+ this.lastCheckpointId =
+ sinkInitContext
+ .getRestoredCheckpointId()
+ .orElse(CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1);
if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
abortLingeringTransactions(
checkNotNull(recoveredStates, "recoveredStates"), lastCheckpointId + 1);
@@ -191,7 +195,7 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
final List<KafkaCommittable> committables =
Collections.singletonList(
KafkaCommittable.of(currentProducer, producerPool::add));
- LOG.info("Committing {} committables.", committables);
+ LOG.debug("Committing {} committables, final commit={}.", committables, flush);
return committables;
}
return Collections.emptyList();
@@ -209,6 +213,7 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
@Override
public void close() throws Exception {
closed = true;
+ LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
closer,
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index 62b382a..5f8d552 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -191,9 +191,9 @@ public class KafkaWriterITCase extends TestLogger {
// create two lingering transactions
failedWriter.prepareCommit(false);
- failedWriter.snapshotState(0);
- failedWriter.prepareCommit(false);
failedWriter.snapshotState(1);
+ failedWriter.prepareCommit(false);
+ failedWriter.snapshotState(2);
try (final KafkaWriter<Integer> recoveredWriter =
createWriterWithConfiguration(
@@ -201,7 +201,7 @@ public class KafkaWriterITCase extends TestLogger {
recoveredWriter.write(1, SINK_WRITER_CONTEXT);
List<KafkaCommittable> committables = recoveredWriter.prepareCommit(false);
- recoveredWriter.snapshotState(0);
+ recoveredWriter.snapshotState(1);
assertThat(committables, hasSize(1));
assertThat(committables.get(0).getProducer().isPresent(), equalTo(true));
@@ -248,7 +248,7 @@ public class KafkaWriterITCase extends TestLogger {
assertThat(writer.getProducerPool(), hasSize(0));
List<KafkaCommittable> committables0 = writer.prepareCommit(false);
- writer.snapshotState(0);
+ writer.snapshotState(1);
assertThat(committables0, hasSize(1));
assertThat(committables0.get(0).getProducer().isPresent(), equalTo(true));
@@ -266,7 +266,7 @@ public class KafkaWriterITCase extends TestLogger {
assertThat(writer.getProducerPool(), hasSize(1));
List<KafkaCommittable> committables1 = writer.prepareCommit(false);
- writer.snapshotState(1);
+ writer.snapshotState(2);
assertThat(committables1, hasSize(1));
assertThat(committables1.get(0).getProducer().isPresent(), equalTo(true));
[flink] 02/11: [FLINK-24131][datastream] Ensure that
SinkWriter#prepareCommit is not called twice for final commit.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d84671c9b651a8df796698b3534f976d9530b19e
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:34:51 2021 +0200
[FLINK-24131][datastream] Ensure that SinkWriter#prepareCommit is not called twice for final commit.
With FLIP-147 (final checkpoint) and the respective opt-in option, a sink would invoke prepareCommit twice for the final commit.
---
.../flink/streaming/runtime/operators/sink/SinkOperator.java | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
index d01434b..015e90c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
@@ -104,6 +104,9 @@ class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<b
private final MailboxExecutor mailboxExecutor;
private Counter numRecordsOutCounter;
+ // record endOfInput state to avoid duplicate prepareCommit on final notifyCheckpointComplete
+ // once FLIP-147 is fully operational all endOfInput processing needs to be removed
+ private boolean endOfInput = false;
SinkOperator(
ProcessingTimeService processingTimeService,
@@ -165,8 +168,10 @@ class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<b
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
super.prepareSnapshotPreBarrier(checkpointId);
- emitCommittables(
- committerHandler.processCommittables(() -> sinkWriter.prepareCommit(false)));
+ if (!endOfInput) {
+ emitCommittables(
+ committerHandler.processCommittables(() -> sinkWriter.prepareCommit(false)));
+ }
}
@Override
@@ -186,6 +191,7 @@ class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<b
@Override
public void endInput() throws Exception {
+ endOfInput = true;
emitCommittables(
committerHandler.processCommittables(() -> sinkWriter.prepareCommit(true)));
emitCommittables(committerHandler.endOfInput());
[flink] 09/11: [FLINK-24131][connectors/kafka] Improve threading
model of KafkaWriter.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 93c7104dade420374d1d3674f2e0d2eb715adb57
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sat Sep 4 22:56:24 2021 +0200
[FLINK-24131][connectors/kafka] Improve threading model of KafkaWriter.
Removed pending records as it doesn't add anything to KafkaWriter#flush (same post-condition as per JavaDoc) but introduces instabilities because of concurrency.
---
.../flink/connector/kafka/sink/KafkaWriter.java | 91 +++++++++-------------
.../connector/kafka/sink/KafkaWriterITCase.java | 3 +-
2 files changed, 40 insertions(+), 54 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index dfe17f6..95a096f 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.kafka.sink;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.connector.sink.Sink;
import org.apache.flink.api.connector.sink.SinkWriter;
@@ -37,14 +38,14 @@ import org.apache.flink.shaded.guava30.com.google.common.io.Closer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.errors.UnknownProducerIdException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Collections;
@@ -53,7 +54,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import static org.apache.flink.util.IOUtils.closeAll;
@@ -80,7 +80,6 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
private final String transactionalIdPrefix;
private final KafkaRecordSerializationSchema<IN> recordSerializer;
private final Callback deliveryCallback;
- private final AtomicLong pendingRecords = new AtomicLong();
private final KafkaRecordSerializationSchema.KafkaSinkContext kafkaSinkContext;
private final Map<String, KafkaMetricMutableWrapper> previouslyCreatedMetrics = new HashMap<>();
private final SinkWriterMetricGroup metricGroup;
@@ -95,7 +94,6 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
private final Deque<FlinkKafkaInternalProducer<byte[], byte[]>> producerPool =
new ArrayDeque<>();
private final Closer closer = Closer.create();
- @Nullable private volatile Exception producerAsyncException;
private long lastCheckpointId;
private boolean closed = false;
@@ -128,13 +126,7 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
this.kafkaProducerConfig = checkNotNull(kafkaProducerConfig, "kafkaProducerConfig");
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
this.recordSerializer = checkNotNull(recordSerializer, "recordSerializer");
- this.deliveryCallback =
- (metadata, exception) -> {
- if (exception != null && producerAsyncException == null) {
- producerAsyncException = exception;
- }
- acknowledgeMessage();
- };
+ this.deliveryCallback = new WriterCallback(sinkInitContext.getMailboxExecutor());
this.disabledMetrics =
kafkaProducerConfig.containsKey(KEY_DISABLE_METRICS)
&& Boolean.parseBoolean(
@@ -181,16 +173,16 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
@Override
public void write(IN element, Context context) throws IOException {
- checkErroneous();
final ProducerRecord<byte[], byte[]> record =
recordSerializer.serialize(element, kafkaSinkContext, context.timestamp());
- pendingRecords.incrementAndGet();
currentProducer.send(record, deliveryCallback);
}
@Override
public List<KafkaCommittable> prepareCommit(boolean flush) {
- flushRecords(flush);
+ if (deliveryGuarantee != DeliveryGuarantee.NONE || flush) {
+ currentProducer.flush();
+ }
if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
final List<KafkaCommittable> committables =
Collections.singletonList(
@@ -270,44 +262,6 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
}
}
- private void acknowledgeMessage() {
- pendingRecords.decrementAndGet();
- }
-
- private void checkErroneous() {
- Exception e = producerAsyncException;
- if (e != null) {
- // prevent double throwing
- producerAsyncException = null;
- throw new RuntimeException("Failed to send data to Kafka: " + e.getMessage(), e);
- }
- }
-
- private void flushRecords(boolean finalFlush) {
- switch (deliveryGuarantee) {
- case EXACTLY_ONCE:
- case AT_LEAST_ONCE:
- currentProducer.flush();
- final long pendingRecordsCount = pendingRecords.get();
- if (pendingRecordsCount != 0) {
- throw new IllegalStateException(
- "Pending record count must be zero at this point: "
- + pendingRecordsCount);
- }
- break;
- case NONE:
- if (finalFlush) {
- currentProducer.flush();
- }
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported Kafka writer semantic " + deliveryGuarantee);
- }
- // if the flushed requests has errors, we should propagate it also and fail the checkpoint
- checkErroneous();
- }
-
/**
* For each checkpoint we create new {@link FlinkKafkaInternalProducer} so that new transactions
* will not clash with transactions created during previous checkpoints ({@code
@@ -402,4 +356,35 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
registerMetricSync();
});
}
+
+ private class WriterCallback implements Callback {
+ private final MailboxExecutor mailboxExecutor;
+
+ public WriterCallback(MailboxExecutor mailboxExecutor) {
+ this.mailboxExecutor = mailboxExecutor;
+ }
+
+ @Override
+ public void onCompletion(RecordMetadata metadata, Exception exception) {
+ if (exception != null) {
+ FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ KafkaWriter.this.currentProducer;
+ mailboxExecutor.execute(
+ () -> throwException(metadata, exception, producer),
+ "Failed to send data to Kafka");
+ }
+ }
+
+ private void throwException(
+ RecordMetadata metadata,
+ Exception exception,
+ FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ String message =
+ String.format("Failed to send data to Kafka %s with %s ", metadata, producer);
+ if (exception instanceof UnknownProducerIdException) {
+ message += KafkaCommitter.UNKNOWN_PRODUCER_ID_ERROR_MESSAGE;
+ }
+ throw new FlinkRuntimeException(message, exception);
+ }
+ }
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index 5f8d552..145534c 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -29,6 +29,7 @@ import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.util.TestLogger;
@@ -371,7 +372,7 @@ public class KafkaWriterITCase extends TestLogger {
@Override
public MailboxExecutor getMailboxExecutor() {
- throw new UnsupportedOperationException("Not implemented");
+ return new SyncMailboxExecutor();
}
@Override
[flink] 11/11: [FLINK-24151][connectors/kafka] Add concurrent
checkpoint test to KafkaSinkITCase.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 85684e4aa91208a7cc06f1425bc5b7d375a48637
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sun Sep 5 23:39:32 2021 +0200
[FLINK-24151][connectors/kafka] Add concurrent checkpoint test to KafkaSinkITCase.
---
.../flink/connector/kafka/sink/KafkaSinkITCase.java | 21 ++++++++++++++++++++-
1 file changed, 20 insertions(+), 1 deletion(-)
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
index 765f2ec..fe4ae3c 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
@@ -191,6 +191,7 @@ public class KafkaSinkITCase extends TestLogger {
public void testRecoveryWithAtLeastOnceGuarantee() throws Exception {
testRecoveryWithAssertion(
DeliveryGuarantee.AT_LEAST_ONCE,
+ 1,
(records) ->
assertThat(records, hasItems(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L)));
}
@@ -199,6 +200,21 @@ public class KafkaSinkITCase extends TestLogger {
public void testRecoveryWithExactlyOnceGuarantee() throws Exception {
testRecoveryWithAssertion(
DeliveryGuarantee.EXACTLY_ONCE,
+ 1,
+ (records) ->
+ assertThat(
+ records,
+ contains(
+ LongStream.range(1, lastCheckpointedRecord.get().get() + 1)
+ .boxed()
+ .toArray())));
+ }
+
+ @Test
+ public void testRecoveryWithExactlyOnceGuaranteeAndConcurrentCheckpoints() throws Exception {
+ testRecoveryWithAssertion(
+ DeliveryGuarantee.EXACTLY_ONCE,
+ 2,
(records) ->
assertThat(
records,
@@ -302,12 +318,15 @@ public class KafkaSinkITCase extends TestLogger {
}
private void testRecoveryWithAssertion(
- DeliveryGuarantee guarantee, java.util.function.Consumer<List<Long>> recordsAssertion)
+ DeliveryGuarantee guarantee,
+ int maxConcurrentCheckpoints,
+ java.util.function.Consumer<List<Long>> recordsAssertion)
throws Exception {
Configuration config = new Configuration();
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
final StreamExecutionEnvironment env = new LocalStreamEnvironment(config);
env.enableCheckpointing(300L);
+ env.getCheckpointConfig().setMaxConcurrentCheckpoints(maxConcurrentCheckpoints);
DataStreamSource<Long> source = env.fromSequence(1, 10);
DataStream<Long> stream =
source.map(new FailingCheckpointMapper(failed, lastCheckpointedRecord));
[flink] 01/11: [FLINK-24131][tests] Harden leak check in
KafkaSinkITCase.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit da0e7dd05f486fef8922f6677d25e1fe00ea90a5
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Fri Sep 3 15:32:10 2021 +0200
[FLINK-24131][tests] Harden leak check in KafkaSinkITCase.
---
.../connector/kafka/sink/KafkaSinkITCase.java | 46 +++++++++++++++++-----
1 file changed, 37 insertions(+), 9 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
index 33ce934..765f2ec 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
@@ -48,6 +48,8 @@ import org.apache.flink.testutils.junit.SharedObjects;
import org.apache.flink.testutils.junit.SharedReference;
import org.apache.flink.util.TestLogger;
+import org.apache.flink.shaded.guava30.com.google.common.base.Joiner;
+
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -198,11 +200,12 @@ public class KafkaSinkITCase extends TestLogger {
testRecoveryWithAssertion(
DeliveryGuarantee.EXACTLY_ONCE,
(records) ->
- assertEquals(
+ assertThat(
records,
- LongStream.range(1, lastCheckpointedRecord.get().get() + 1)
- .boxed()
- .collect(Collectors.toList())));
+ contains(
+ LongStream.range(1, lastCheckpointedRecord.get().get() + 1)
+ .boxed()
+ .toArray())));
}
@Test
@@ -278,6 +281,7 @@ public class KafkaSinkITCase extends TestLogger {
Configuration config,
@Nullable String transactionalIdPrefix)
throws Exception {
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
final StreamExecutionEnvironment env = new LocalStreamEnvironment(config);
env.enableCheckpointing(100L);
env.setRestartStrategy(RestartStrategies.noRestart());
@@ -300,7 +304,9 @@ public class KafkaSinkITCase extends TestLogger {
private void testRecoveryWithAssertion(
DeliveryGuarantee guarantee, java.util.function.Consumer<List<Long>> recordsAssertion)
throws Exception {
- final StreamExecutionEnvironment env = new LocalStreamEnvironment();
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ final StreamExecutionEnvironment env = new LocalStreamEnvironment(config);
env.enableCheckpointing(300L);
DataStreamSource<Long> source = env.fromSequence(1, 10);
DataStream<Long> stream =
@@ -588,12 +594,34 @@ public class KafkaSinkITCase extends TestLogger {
public void initializeState(FunctionInitializationContext context) throws Exception {}
}
- private void checkProducerLeak() {
- for (Thread t : Thread.getAllStackTraces().keySet()) {
- if (t.getName().contains("kafka-producer-network-thread")) {
- fail("Detected producer leak. Thread name: " + t.getName());
+ private void checkProducerLeak() throws InterruptedException {
+ List<Map.Entry<Thread, StackTraceElement[]>> leaks = null;
+ for (int tries = 0; tries < 10; tries++) {
+ leaks =
+ Thread.getAllStackTraces().entrySet().stream()
+ .filter(this::findAliveKafkaThread)
+ .collect(Collectors.toList());
+ if (leaks.isEmpty()) {
+ return;
}
+ Thread.sleep(1000);
}
+
+ for (Map.Entry<Thread, StackTraceElement[]> leak : leaks) {
+ leak.getKey().stop();
+ }
+ fail(
+ "Detected producer leaks:\n"
+ + leaks.stream().map(this::format).collect(Collectors.joining("\n\n")));
+ }
+
+ private String format(Map.Entry<Thread, StackTraceElement[]> leak) {
+ return leak.getKey().getName() + ":\n" + Joiner.on("\n").join(leak.getValue());
+ }
+
+ private boolean findAliveKafkaThread(Map.Entry<Thread, StackTraceElement[]> threadStackTrace) {
+ return threadStackTrace.getKey().getState() != Thread.State.TERMINATED
+ && threadStackTrace.getKey().getName().contains("kafka-producer-network-thread");
}
/**