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:32 UTC
[flink] 09/11: [FLINK-24131][connectors/kafka] Improve threading
model of KafkaWriter.
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