You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2023/04/04 17:50:15 UTC
[flink-connector-kafka] branch v4.0 updated (0b2bf49 -> a5df3c7)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
from 0b2bf49 [hotfix] Allow manual triggering of nightly builds
new 68e9c56 [FLINK-31319][connectors/kafka] Fix kafka new source partitionDiscoveryIntervalMs error condition check cause bounded source can not quit
new 15b51f1 [FLINK-31620] [upsert-kafka] Flush wrapped writer
new 7e3c5a1 [FLINK-31305] Propagate producer exceptions outside of mailbox executor so that checkpoints can correctly fail
new dd09fa9 [FLINK-31305] fix error propagation bug in WriterCallback and use TestSinkInitContext general purpose sink testing tool
new a5df3c7 [FLINK-31363] [kafka] Add hasDataInTransaction flag in FlinkKafkaInternalProducer
The 5 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:
flink-connector-kafka/pom.xml | 8 +
.../kafka/sink/FlinkKafkaInternalProducer.java | 29 ++++
.../flink/connector/kafka/sink/KafkaWriter.java | 60 ++++++-
.../source/enumerator/KafkaSourceEnumerator.java | 2 +-
.../kafka/table/ReducingUpsertWriter.java | 9 +-
.../sink/FlinkKafkaInternalProducerITCase.java | 59 ++++++-
.../connector/kafka/sink/KafkaWriterITCase.java | 192 +++++++++++++++++----
.../source/enumerator/KafkaEnumeratorTest.java | 113 +++++++++++-
.../kafka/table/ReducingUpsertWriterTest.java | 7 +-
9 files changed, 422 insertions(+), 57 deletions(-)
[flink-connector-kafka] 01/05: [FLINK-31319][connectors/kafka] Fix kafka new source partitionDiscoveryIntervalMs error condition check cause bounded source can not quit
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 68e9c56d73197d0bf03e4119de236433913f9555
Author: Ran Tao <ch...@gmail.com>
AuthorDate: Sat Mar 18 17:55:01 2023 +0800
[FLINK-31319][connectors/kafka] Fix kafka new source partitionDiscoveryIntervalMs error condition check cause bounded source can not quit
This closes #8.
---
.../source/enumerator/KafkaSourceEnumerator.java | 2 +-
.../source/enumerator/KafkaEnumeratorTest.java | 113 +++++++++++++++++++--
2 files changed, 108 insertions(+), 7 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java
index 9cf233c..137f420 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java
@@ -298,7 +298,7 @@ public class KafkaSourceEnumerator
if (t != null) {
throw new FlinkRuntimeException("Failed to initialize partition splits due to ", t);
}
- if (partitionDiscoveryIntervalMs < 0) {
+ if (partitionDiscoveryIntervalMs <= 0) {
LOG.debug("Partition discovery is disabled.");
noMoreNewPartitionSplits = true;
}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
index 8c39b54..8d0d3fc 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
@@ -98,6 +98,10 @@ public class KafkaEnumeratorTest {
assertThat(context.getOneTimeCallables())
.as("A one time partition discovery callable should have been scheduled")
.hasSize(1);
+
+ // enumerator just start noMoreNewPartitionSplits will be false
+ assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+ .isFalse();
}
}
@@ -115,6 +119,10 @@ public class KafkaEnumeratorTest {
assertThat(context.getPeriodicCallables())
.as("A periodic partition discovery callable should have been scheduled")
.hasSize(1);
+
+ // enumerator just start noMoreNewPartitionSplits will be false
+ assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+ .isFalse();
}
}
@@ -166,6 +174,78 @@ public class KafkaEnumeratorTest {
}
}
+ @Test
+ public void testRunWithDiscoverPartitionsOnceToCheckNoMoreSplit() throws Throwable {
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(context, DISABLE_PERIODIC_PARTITION_DISCOVERY)) {
+
+ // Start the enumerator and it should schedule a one time task to discover and assign
+ // partitions.
+ enumerator.start();
+ assertThat(context.getOneTimeCallables())
+ .as("A one time partition discovery callable should have been scheduled")
+ .hasSize(1);
+ assertThat(context.getPeriodicCallables()).isEmpty();
+ // Run the partition discover callable and check the partition assignment.
+ runOneTimePartitionDiscovery(context);
+
+ // enumerator noMoreNewPartitionSplits first will be false, when execute
+ // handlePartitionSplitChanges will be set true
+ assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+ .isTrue();
+ }
+ }
+
+ @Test
+ public void testRunWithPeriodicPartitionDiscoveryOnceToCheckNoMoreSplit() throws Throwable {
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(context, ENABLE_PERIODIC_PARTITION_DISCOVERY)) {
+
+ // Start the enumerator and it should schedule a one time task to discover and assign
+ // partitions.
+ enumerator.start();
+ assertThat(context.getOneTimeCallables()).isEmpty();
+ assertThat(context.getPeriodicCallables())
+ .as("A periodic partition discovery callable should have been scheduled")
+ .hasSize(1);
+ // Run the partition discover callable and check the partition assignment.
+ runPeriodicPartitionDiscovery(context);
+
+ // enumerator noMoreNewPartitionSplits first will be false, even when execute
+ // handlePartitionSplitChanges it still be false
+ assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+ .isFalse();
+ }
+ }
+
+ @Test
+ public void testRunWithDiscoverPartitionsOnceWithZeroMsToCheckNoMoreSplit() throws Throwable {
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ // set partitionDiscoveryIntervalMs = 0
+ KafkaSourceEnumerator enumerator = createEnumerator(context, 0L)) {
+
+ // Start the enumerator, and it should schedule a one time task to discover and assign
+ // partitions.
+ enumerator.start();
+ assertThat(context.getOneTimeCallables())
+ .as("A one time partition discovery callable should have been scheduled")
+ .hasSize(1);
+ assertThat(context.getPeriodicCallables()).isEmpty();
+ // Run the partition discover callable and check the partition assignment.
+ runOneTimePartitionDiscovery(context);
+
+ // enumerator noMoreNewPartitionSplits first will be false, when execute
+ // handlePartitionSplitChanges will be set true
+ assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+ .isTrue();
+ }
+ }
+
@Test(timeout = 30000L)
public void testDiscoverPartitionsPeriodically() throws Throwable {
try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
@@ -261,7 +341,7 @@ public class KafkaEnumeratorTest {
KafkaSourceEnumerator enumerator =
createEnumerator(
context2,
- ENABLE_PERIODIC_PARTITION_DISCOVERY,
+ ENABLE_PERIODIC_PARTITION_DISCOVERY ? 1 : -1,
PRE_EXISTING_TOPICS,
preexistingAssignments,
new Properties())) {
@@ -290,7 +370,7 @@ public class KafkaEnumeratorTest {
KafkaSourceEnumerator enumerator =
createEnumerator(
context,
- ENABLE_PERIODIC_PARTITION_DISCOVERY,
+ ENABLE_PERIODIC_PARTITION_DISCOVERY ? 1 : -1,
PRE_EXISTING_TOPICS,
Collections.emptySet(),
properties)) {
@@ -403,6 +483,12 @@ public class KafkaEnumeratorTest {
enumContext, enablePeriodicPartitionDiscovery, EXCLUDE_DYNAMIC_TOPIC);
}
+ private KafkaSourceEnumerator createEnumerator(
+ MockSplitEnumeratorContext<KafkaPartitionSplit> enumContext,
+ long partitionDiscoveryIntervalMs) {
+ return createEnumerator(enumContext, partitionDiscoveryIntervalMs, EXCLUDE_DYNAMIC_TOPIC);
+ }
+
private KafkaSourceEnumerator createEnumerator(
MockSplitEnumeratorContext<KafkaPartitionSplit> enumContext,
boolean enablePeriodicPartitionDiscovery,
@@ -413,7 +499,23 @@ public class KafkaEnumeratorTest {
}
return createEnumerator(
enumContext,
- enablePeriodicPartitionDiscovery,
+ enablePeriodicPartitionDiscovery ? 1 : -1,
+ topics,
+ Collections.emptySet(),
+ new Properties());
+ }
+
+ private KafkaSourceEnumerator createEnumerator(
+ MockSplitEnumeratorContext<KafkaPartitionSplit> enumContext,
+ long partitionDiscoveryIntervalMs,
+ boolean includeDynamicTopic) {
+ List<String> topics = new ArrayList<>(PRE_EXISTING_TOPICS);
+ if (includeDynamicTopic) {
+ topics.add(DYNAMIC_TOPIC_NAME);
+ }
+ return createEnumerator(
+ enumContext,
+ partitionDiscoveryIntervalMs,
topics,
Collections.emptySet(),
new Properties());
@@ -425,7 +527,7 @@ public class KafkaEnumeratorTest {
*/
private KafkaSourceEnumerator createEnumerator(
MockSplitEnumeratorContext<KafkaPartitionSplit> enumContext,
- boolean enablePeriodicPartitionDiscovery,
+ long partitionDiscoveryIntervalMs,
Collection<String> topicsToSubscribe,
Set<TopicPartition> assignedPartitions,
Properties overrideProperties) {
@@ -442,10 +544,9 @@ public class KafkaEnumeratorTest {
Properties props =
new Properties(KafkaSourceTestEnv.getConsumerProperties(StringDeserializer.class));
KafkaSourceEnumerator.deepCopyProperties(overrideProperties, props);
- String partitionDiscoverInterval = enablePeriodicPartitionDiscovery ? "1" : "-1";
props.setProperty(
KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
- partitionDiscoverInterval);
+ String.valueOf(partitionDiscoveryIntervalMs));
return new KafkaSourceEnumerator(
subscriber,
[flink-connector-kafka] 03/05: [FLINK-31305] Propagate producer exceptions outside of mailbox executor so that checkpoints can correctly fail
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 7e3c5a16458cced88360fb318e5b6e8b368a79d1
Author: mas-chen <ma...@berkeley.edu>
AuthorDate: Fri Mar 10 00:42:21 2023 -0800
[FLINK-31305] Propagate producer exceptions outside of mailbox executor so that checkpoints can correctly fail
This closes #19.
---
.../flink/connector/kafka/sink/KafkaWriter.java | 58 +++++++++++++++++++---
.../connector/kafka/sink/KafkaWriterITCase.java | 26 ++++++++++
2 files changed, 77 insertions(+), 7 deletions(-)
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 ba2cb4e..adc05b8 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
@@ -89,7 +89,7 @@ class KafkaWriter<IN>
private final KafkaRecordSerializationSchema<IN> recordSerializer;
private final Callback deliveryCallback;
private final KafkaRecordSerializationSchema.KafkaSinkContext kafkaSinkContext;
-
+ private volatile Exception asyncProducerException;
private final Map<String, KafkaMetricMutableWrapper> previouslyCreatedMetrics = new HashMap<>();
private final SinkWriterMetricGroup metricGroup;
private final boolean disabledMetrics;
@@ -139,6 +139,7 @@ class KafkaWriter<IN>
this.kafkaProducerConfig = checkNotNull(kafkaProducerConfig, "kafkaProducerConfig");
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
this.recordSerializer = checkNotNull(recordSerializer, "recordSerializer");
+ checkNotNull(sinkInitContext, "sinkInitContext");
this.deliveryCallback =
new WriterCallback(
sinkInitContext.getMailboxExecutor(),
@@ -150,7 +151,6 @@ class KafkaWriter<IN>
|| kafkaProducerConfig.containsKey(KEY_REGISTER_METRICS)
&& !Boolean.parseBoolean(
kafkaProducerConfig.get(KEY_REGISTER_METRICS).toString());
- checkNotNull(sinkInitContext, "sinkInitContext");
this.timeService = sinkInitContext.getProcessingTimeService();
this.metricGroup = sinkInitContext.metricGroup();
this.numBytesOutCounter = metricGroup.getIOMetricGroup().getNumBytesOutCounter();
@@ -192,6 +192,8 @@ class KafkaWriter<IN>
@Override
public void write(@Nullable IN element, Context context) throws IOException {
+ checkAsyncException();
+
final ProducerRecord<byte[], byte[]> record =
recordSerializer.serialize(element, kafkaSinkContext, context.timestamp());
if (record != null) {
@@ -206,6 +208,8 @@ class KafkaWriter<IN>
LOG.debug("final flush={}", endOfInput);
currentProducer.flush();
}
+
+ checkAsyncException();
}
@Override
@@ -241,6 +245,9 @@ class KafkaWriter<IN>
checkState(currentProducer.isClosed());
currentProducer = null;
});
+
+ // Rethrow exception for the case in which close is called before writer() and flush().
+ checkAsyncException();
}
private void abortCurrentProducer() {
@@ -264,6 +271,16 @@ class KafkaWriter<IN>
return currentProducer;
}
+ @VisibleForTesting
+ Exception getAsyncProducerException() {
+ return asyncProducerException;
+ }
+
+ @VisibleForTesting
+ void setAsyncProducerException(Exception asyncProducerException) {
+ this.asyncProducerException = asyncProducerException;
+ }
+
void abortLingeringTransactions(
Collection<KafkaWriterState> recoveredStates, long startCheckpointId) {
List<String> prefixesToAbort = Lists.newArrayList(transactionalIdPrefix);
@@ -397,6 +414,18 @@ class KafkaWriter<IN>
});
}
+ /** This logic needs to be invoked by write AND flush since we support various semantics. */
+ private void checkAsyncException() throws IOException {
+ // reset this exception since we could close the writer later on
+ Exception e = asyncProducerException;
+ if (e != null) {
+
+ asyncProducerException = null;
+ throw new IOException(
+ "One or more Kafka Producer send requests have encountered exception", e);
+ }
+ }
+
private class WriterCallback implements Callback {
private final MailboxExecutor mailboxExecutor;
@Nullable private final Consumer<RecordMetadata> metadataConsumer;
@@ -413,12 +442,27 @@ class KafkaWriter<IN>
if (exception != null) {
FlinkKafkaInternalProducer<byte[], byte[]> producer =
KafkaWriter.this.currentProducer;
- mailboxExecutor.execute(
+
+ // Propagate the first exception since amount of exceptions could be large. Need to
+ // do this in Producer IO thread since flush() guarantees that the future will
+ // complete. The same guarantee does not hold for tasks executed in separate
+ // executor e.g. mailbox executor. flush() needs to have the exception immediately
+ // available to fail the checkpoint.
+ if (asyncProducerException != null) {
+ asyncProducerException = decorateException(metadata, exception, producer);
+ }
+
+ mailboxExecutor.submit(
() -> {
+ // Need to send metrics through mailbox thread since we are in the
+ // producer io
+ // thread
numRecordsOutErrorsCounter.inc();
- throwException(metadata, exception, producer);
+
+ // Checking for exceptions from previous writes
+ checkAsyncException();
},
- "Failed to send data to Kafka");
+ "Update error metric");
}
if (metadataConsumer != null) {
@@ -426,7 +470,7 @@ class KafkaWriter<IN>
}
}
- private void throwException(
+ private FlinkRuntimeException decorateException(
RecordMetadata metadata,
Exception exception,
FlinkKafkaInternalProducer<byte[], byte[]> producer) {
@@ -435,7 +479,7 @@ class KafkaWriter<IN>
if (exception instanceof UnknownProducerIdException) {
message += KafkaCommitter.UNKNOWN_PRODUCER_ID_ERROR_MESSAGE;
}
- throw new FlinkRuntimeException(message, exception);
+ return new FlinkRuntimeException(message, exception);
}
}
}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index 98f64ed..e7627b2 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -75,6 +75,7 @@ import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaCo
import static org.apache.flink.connector.kafka.testutils.KafkaUtil.drainAllRecordsFromTopic;
import static org.apache.flink.util.DockerImageVersions.KAFKA;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatCode;
/** Tests for the standalone KafkaWriter. */
@ExtendWith(TestLoggerExtension.class)
@@ -223,6 +224,9 @@ public class KafkaWriterITCase {
}
writer.write(3, SINK_WRITER_CONTEXT);
+ // this doesn't throw exception because the exception is thrown in the Producer IO
+ // thread in unit tests due to the mock mailbox executor, while it would be thrown in
+ // flush() when the real mailbox executor is configured
writer.flush(false);
writer.prepareCommit();
assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
@@ -383,6 +387,28 @@ public class KafkaWriterITCase {
}
}
+ @Test
+ public void testErrorPropagation() {
+ Properties properties = getKafkaClientConfiguration();
+ final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(properties, DeliveryGuarantee.AT_LEAST_ONCE);
+ try {
+ writer.setAsyncProducerException(
+ new IOException("previous send request encountered error."));
+ assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
+ .hasRootCauseExactlyInstanceOf(IOException.class);
+
+ writer.setAsyncProducerException(
+ new IOException("previous send request encountered error."));
+ assertThatCode(() -> writer.flush(false))
+ .hasRootCauseExactlyInstanceOf(IOException.class);
+ } finally {
+ writer.setAsyncProducerException(
+ new IOException("previous send request encountered error."));
+ assertThatCode(writer::close).hasRootCauseExactlyInstanceOf(IOException.class);
+ }
+ }
+
private void assertKafkaMetricNotPresent(
DeliveryGuarantee guarantee, String configKey, String configValue) throws Exception {
final Properties config = getKafkaClientConfiguration();
[flink-connector-kafka] 05/05: [FLINK-31363] [kafka] Add hasDataInTransaction flag in FlinkKafkaInternalProducer
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit a5df3c7c7f4925ff70114f862bca7588b819ac21
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Thu Mar 23 14:18:08 2023 -0700
[FLINK-31363] [kafka] Add hasDataInTransaction flag in FlinkKafkaInternalProducer
This closes #15.
---
.../kafka/sink/FlinkKafkaInternalProducer.java | 29 +++++++++++
.../flink/connector/kafka/sink/KafkaWriter.java | 11 +++-
.../sink/FlinkKafkaInternalProducerITCase.java | 59 +++++++++++++++++++---
.../connector/kafka/sink/KafkaWriterITCase.java | 26 ++++++++++
4 files changed, 118 insertions(+), 7 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
index a023cdd..246fca6 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
@@ -17,8 +17,11 @@
package org.apache.flink.connector.kafka.sink;
+import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.internals.TransactionManager;
import org.apache.kafka.clients.producer.internals.TransactionalRequestResult;
import org.apache.kafka.common.errors.ProducerFencedException;
@@ -33,6 +36,7 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Properties;
+import java.util.concurrent.Future;
import static org.apache.flink.util.Preconditions.checkState;
@@ -49,6 +53,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
@Nullable private String transactionalId;
private volatile boolean inTransaction;
+ private volatile boolean hasRecordsInTransaction;
private volatile boolean closed;
public FlinkKafkaInternalProducer(Properties properties, @Nullable String transactionalId) {
@@ -67,6 +72,14 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
return props;
}
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ if (inTransaction) {
+ hasRecordsInTransaction = true;
+ }
+ return super.send(record, callback);
+ }
+
@Override
public void flush() {
super.flush();
@@ -86,6 +99,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
LOG.debug("abortTransaction {}", transactionalId);
checkState(inTransaction, "Transaction was not started");
inTransaction = false;
+ hasRecordsInTransaction = false;
super.abortTransaction();
}
@@ -94,6 +108,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
LOG.debug("commitTransaction {}", transactionalId);
checkState(inTransaction, "Transaction was not started");
inTransaction = false;
+ hasRecordsInTransaction = false;
super.commitTransaction();
}
@@ -101,6 +116,10 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
return inTransaction;
}
+ public boolean hasRecordsInTransaction() {
+ return hasRecordsInTransaction;
+ }
+
@Override
public void close() {
closed = true;
@@ -302,8 +321,18 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
transitionTransactionManagerStateTo(transactionManager, "READY");
transitionTransactionManagerStateTo(transactionManager, "IN_TRANSACTION");
+
+ // the transactionStarted flag in the KafkaProducer controls whether
+ // an EndTxnRequest will actually be sent to Kafka for a commit
+ // or abort API call. This flag is set only after the first send (i.e.
+ // only if data is actually written to some partition).
+ // In checkpoints, we only ever store metadata of pre-committed
+ // transactions that actually have records; therefore, on restore
+ // when we create recovery producers to resume transactions and commit
+ // them, we should always set this flag.
setField(transactionManager, "transactionStarted", true);
this.inTransaction = true;
+ this.hasRecordsInTransaction = true;
}
}
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 0df3bcf..48c5238 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
@@ -214,13 +214,22 @@ class KafkaWriter<IN>
@Override
public Collection<KafkaCommittable> prepareCommit() {
- if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
+ if (deliveryGuarantee != DeliveryGuarantee.EXACTLY_ONCE) {
+ return Collections.emptyList();
+ }
+
+ // only return a KafkaCommittable if the current transaction has been written some data
+ if (currentProducer.hasRecordsInTransaction()) {
final List<KafkaCommittable> committables =
Collections.singletonList(
KafkaCommittable.of(currentProducer, producerPool::add));
LOG.debug("Committing {} committables.", committables);
return committables;
}
+
+ // otherwise, we commit the empty transaction as is (no-op) and just recycle the producer
+ currentProducer.commitTransaction();
+ producerPool.add(currentProducer);
return Collections.emptyList();
}
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 51770f0..dd15ec2 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
@@ -28,6 +28,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
+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;
@@ -63,16 +64,15 @@ class FlinkKafkaInternalProducerITCase {
private static final KafkaContainer KAFKA_CONTAINER =
createKafkaContainer(KAFKA, LOG).withEmbeddedZookeeper();
- private static final String TRANSACTION_PREFIX = "test-transaction-";
-
@Test
void testInitTransactionId() {
final String topic = "test-init-transactions";
+ final String transactionIdPrefix = "testInitTransactionId-";
try (FlinkKafkaInternalProducer<String, String> reuse =
new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
int numTransactions = 20;
for (int i = 1; i <= numTransactions; i++) {
- reuse.initTransactionId(TRANSACTION_PREFIX + i);
+ reuse.initTransactionId(transactionIdPrefix + i);
reuse.beginTransaction();
reuse.send(new ProducerRecord<>(topic, "test-value-" + i));
if (i % 2 == 0) {
@@ -81,12 +81,59 @@ class FlinkKafkaInternalProducerITCase {
reuse.flush();
reuse.abortTransaction();
}
- assertNumTransactions(i);
+ assertNumTransactions(i, transactionIdPrefix);
assertThat(readRecords(topic).count()).isEqualTo(i / 2);
}
}
}
+ @Test
+ void testCommitResumedTransaction() {
+ final String topic = "test-commit-resumed-transaction";
+ final String transactionIdPrefix = "testCommitResumedTransaction-";
+ final String transactionalId = transactionIdPrefix + "id";
+
+ KafkaCommittable snapshottedCommittable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(topic, "test-value"));
+ producer.flush();
+ snapshottedCommittable = KafkaCommittable.of(producer, ignored -> {});
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ resumedProducer.resumeTransaction(
+ snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch());
+ resumedProducer.commitTransaction();
+ }
+
+ assertNumTransactions(1, transactionIdPrefix);
+ assertThat(readRecords(topic).count()).isEqualTo(1);
+ }
+
+ @Test
+ void testCommitResumedEmptyTransactionShouldFail() {
+ KafkaCommittable snapshottedCommittable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ snapshottedCommittable = KafkaCommittable.of(producer, ignored -> {});
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ resumedProducer.resumeTransaction(
+ snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch());
+
+ assertThatThrownBy(resumedProducer::commitTransaction)
+ .isInstanceOf(InvalidTxnStateException.class);
+ }
+ }
+
@ParameterizedTest
@MethodSource("provideTransactionsFinalizer")
void testResetInnerTransactionIfFinalizingTransactionFailed(
@@ -131,10 +178,10 @@ class FlinkKafkaInternalProducerITCase {
FlinkKafkaInternalProducer::abortTransaction);
}
- private void assertNumTransactions(int numTransactions) {
+ private void assertNumTransactions(int numTransactions, String transactionIdPrefix) {
List<KafkaTransactionLog.TransactionRecord> transactions =
new KafkaTransactionLog(getProperties())
- .getTransactions(id -> id.startsWith(TRANSACTION_PREFIX));
+ .getTransactions(id -> id.startsWith(transactionIdPrefix));
assertThat(
transactions.stream()
.map(KafkaTransactionLog.TransactionRecord::getTransactionId)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index c9d226d..c1b022d 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -416,6 +416,7 @@ public class KafkaWriterITCase {
getKafkaClientConfiguration(), DeliveryGuarantee.EXACTLY_ONCE)) {
assertThat(writer.getProducerPool()).hasSize(0);
+ writer.write(1, SINK_WRITER_CONTEXT);
writer.flush(false);
Collection<KafkaCommittable> committables0 = writer.prepareCommit();
writer.snapshotState(1);
@@ -435,6 +436,7 @@ public class KafkaWriterITCase {
committable.getProducer().get().close();
assertThat(writer.getProducerPool()).hasSize(1);
+ writer.write(1, SINK_WRITER_CONTEXT);
writer.flush(false);
Collection<KafkaCommittable> committables1 = writer.prepareCommit();
writer.snapshotState(2);
@@ -448,6 +450,30 @@ public class KafkaWriterITCase {
}
}
+ /**
+ * Tests that if a pre-commit attempt occurs on an empty transaction, the writer should not emit
+ * a KafkaCommittable, and instead immediately commit the empty transaction and recycle the
+ * producer.
+ */
+ @Test
+ void prepareCommitForEmptyTransaction() throws Exception {
+ try (final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(
+ getKafkaClientConfiguration(), DeliveryGuarantee.EXACTLY_ONCE)) {
+ assertThat(writer.getProducerPool()).hasSize(0);
+
+ // no data written to current transaction
+ writer.flush(false);
+ Collection<KafkaCommittable> emptyCommittables = writer.prepareCommit();
+
+ assertThat(emptyCommittables).hasSize(0);
+ assertThat(writer.getProducerPool()).hasSize(1);
+ final FlinkKafkaInternalProducer<?, ?> recycledProducer =
+ writer.getProducerPool().pop();
+ assertThat(recycledProducer.isInTransaction()).isFalse();
+ }
+ }
+
/**
* Tests that open transactions are automatically aborted on close such that successive writes
* succeed.
[flink-connector-kafka] 04/05: [FLINK-31305] fix error propagation bug in WriterCallback and use TestSinkInitContext general purpose sink testing tool
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit dd09fa9a3d05c7096c85cb14f2a792a66a915547
Author: mas-chen <ma...@berkeley.edu>
AuthorDate: Wed Mar 29 16:40:22 2023 -0700
[FLINK-31305] fix error propagation bug in WriterCallback and use TestSinkInitContext general purpose sink testing tool
This closes #22303.
---
flink-connector-kafka/pom.xml | 8 +
.../flink/connector/kafka/sink/KafkaWriter.java | 23 +--
.../connector/kafka/sink/KafkaWriterITCase.java | 190 +++++++++++++++------
3 files changed, 149 insertions(+), 72 deletions(-)
diff --git a/flink-connector-kafka/pom.xml b/flink-connector-kafka/pom.xml
index 0355775..23598f4 100644
--- a/flink-connector-kafka/pom.xml
+++ b/flink-connector-kafka/pom.xml
@@ -180,6 +180,14 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-metrics-jmx</artifactId>
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 adc05b8..0df3bcf 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
@@ -271,16 +271,6 @@ class KafkaWriter<IN>
return currentProducer;
}
- @VisibleForTesting
- Exception getAsyncProducerException() {
- return asyncProducerException;
- }
-
- @VisibleForTesting
- void setAsyncProducerException(Exception asyncProducerException) {
- this.asyncProducerException = asyncProducerException;
- }
-
void abortLingeringTransactions(
Collection<KafkaWriterState> recoveredStates, long startCheckpointId) {
List<String> prefixesToAbort = Lists.newArrayList(transactionalIdPrefix);
@@ -414,13 +404,17 @@ class KafkaWriter<IN>
});
}
- /** This logic needs to be invoked by write AND flush since we support various semantics. */
+ /**
+ * This method should only be invoked in the mailbox thread since the counter is not volatile.
+ * Logic needs to be invoked by write AND flush since we support various semantics.
+ */
private void checkAsyncException() throws IOException {
// reset this exception since we could close the writer later on
Exception e = asyncProducerException;
if (e != null) {
asyncProducerException = null;
+ numRecordsOutErrorsCounter.inc();
throw new IOException(
"One or more Kafka Producer send requests have encountered exception", e);
}
@@ -448,17 +442,12 @@ class KafkaWriter<IN>
// complete. The same guarantee does not hold for tasks executed in separate
// executor e.g. mailbox executor. flush() needs to have the exception immediately
// available to fail the checkpoint.
- if (asyncProducerException != null) {
+ if (asyncProducerException == null) {
asyncProducerException = decorateException(metadata, exception, producer);
}
mailboxExecutor.submit(
() -> {
- // Need to send metrics through mailbox thread since we are in the
- // producer io
- // thread
- numRecordsOutErrorsCounter.inc();
-
// Checking for exceptions from previous writes
checkAsyncException();
},
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index e7627b2..c9d226d 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -17,20 +17,18 @@
package org.apache.flink.connector.kafka.sink;
-import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.base.sink.writer.TestSinkInitContext;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
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.TestLoggerExtension;
@@ -41,6 +39,7 @@ import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -198,39 +197,135 @@ public class KafkaWriterITCase {
}
@Test
- void testNumRecordsOutErrorsCounterMetric() throws Exception {
+ void testFlushAsyncErrorPropagationAndErrorCounter() throws Exception {
Properties properties = getKafkaClientConfiguration();
- final InternalSinkWriterMetricGroup metricGroup =
- InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup());
- try (final KafkaWriter<Integer> writer =
+ SinkInitContext sinkInitContext =
+ new SinkInitContext(
+ InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup()),
+ timeService,
+ null);
+ final KafkaWriter<Integer> writer =
createWriterWithConfiguration(
- properties, DeliveryGuarantee.EXACTLY_ONCE, metricGroup)) {
- final Counter numRecordsOutErrors = metricGroup.getNumRecordsOutErrorsCounter();
- assertThat(numRecordsOutErrors.getCount()).isEqualTo(0L);
+ properties, DeliveryGuarantee.EXACTLY_ONCE, sinkInitContext);
+ final Counter numRecordsOutErrors =
+ sinkInitContext.metricGroup.getNumRecordsOutErrorsCounter();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(0L);
+
+ triggerProducerException(writer, properties);
+
+ // test flush
+ assertThatCode(() -> writer.flush(false))
+ .hasRootCauseExactlyInstanceOf(ProducerFencedException.class);
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+
+ assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
+ .as("the exception is not thrown again")
+ .doesNotThrowAnyException();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+ }
- writer.write(1, SINK_WRITER_CONTEXT);
- assertThat(numRecordsOutErrors.getCount()).isEqualTo(0L);
+ @Test
+ void testWriteAsyncErrorPropagationAndErrorCounter() throws Exception {
+ Properties properties = getKafkaClientConfiguration();
- final String transactionalId = writer.getCurrentProducer().getTransactionalId();
+ SinkInitContext sinkInitContext =
+ new SinkInitContext(
+ InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup()),
+ timeService,
+ null);
+ final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(
+ properties, DeliveryGuarantee.EXACTLY_ONCE, sinkInitContext);
+ final Counter numRecordsOutErrors =
+ sinkInitContext.metricGroup.getNumRecordsOutErrorsCounter();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(0L);
+
+ triggerProducerException(writer, properties);
+ // to ensure that the exceptional send request has completed
+ writer.getCurrentProducer().flush();
+
+ assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
+ .hasRootCauseExactlyInstanceOf(ProducerFencedException.class);
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+
+ assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
+ .as("the exception is not thrown again")
+ .doesNotThrowAnyException();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+ }
- try (FlinkKafkaInternalProducer<byte[], byte[]> producer =
- new FlinkKafkaInternalProducer<>(properties, transactionalId)) {
+ @Test
+ void testMailboxAsyncErrorPropagationAndErrorCounter() throws Exception {
+ Properties properties = getKafkaClientConfiguration();
- producer.initTransactions();
- producer.beginTransaction();
- producer.send(new ProducerRecord<byte[], byte[]>(topic, "2".getBytes()));
- producer.commitTransaction();
- }
+ SinkInitContext sinkInitContext =
+ new SinkInitContext(
+ InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup()),
+ timeService,
+ null);
+ final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(
+ properties, DeliveryGuarantee.EXACTLY_ONCE, sinkInitContext);
+ final Counter numRecordsOutErrors =
+ sinkInitContext.metricGroup.getNumRecordsOutErrorsCounter();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(0L);
- writer.write(3, SINK_WRITER_CONTEXT);
- // this doesn't throw exception because the exception is thrown in the Producer IO
- // thread in unit tests due to the mock mailbox executor, while it would be thrown in
- // flush() when the real mailbox executor is configured
- writer.flush(false);
- writer.prepareCommit();
- assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+ triggerProducerException(writer, properties);
+ // to ensure that the exceptional send request has completed
+ writer.getCurrentProducer().flush();
+
+ while (sinkInitContext.getMailboxExecutor().tryYield()) {
+ // execute all mails
+ }
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+
+ assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
+ .as("the exception is not thrown again")
+ .doesNotThrowAnyException();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+ }
+
+ @Test
+ void testCloseAsyncErrorPropagationAndErrorCounter() throws Exception {
+ Properties properties = getKafkaClientConfiguration();
+
+ SinkInitContext sinkInitContext =
+ new SinkInitContext(
+ InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup()),
+ timeService,
+ null);
+ final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(
+ properties, DeliveryGuarantee.EXACTLY_ONCE, sinkInitContext);
+ final Counter numRecordsOutErrors =
+ sinkInitContext.metricGroup.getNumRecordsOutErrorsCounter();
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(0L);
+
+ triggerProducerException(writer, properties);
+ // to ensure that the exceptional send request has completed
+ writer.getCurrentProducer().flush();
+
+ // test flush
+ assertThatCode(writer::close)
+ .as("flush should throw the exception from the WriterCallback")
+ .hasRootCauseExactlyInstanceOf(ProducerFencedException.class);
+ assertThat(numRecordsOutErrors.getCount()).isEqualTo(1L);
+ }
+
+ private void triggerProducerException(KafkaWriter<Integer> writer, Properties properties)
+ throws IOException {
+ final String transactionalId = writer.getCurrentProducer().getTransactionalId();
+
+ try (FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ new FlinkKafkaInternalProducer<>(properties, transactionalId)) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<byte[], byte[]>(topic, "1".getBytes()));
+ producer.commitTransaction();
}
+
+ writer.write(1, SINK_WRITER_CONTEXT);
}
@Test
@@ -387,28 +482,6 @@ public class KafkaWriterITCase {
}
}
- @Test
- public void testErrorPropagation() {
- Properties properties = getKafkaClientConfiguration();
- final KafkaWriter<Integer> writer =
- createWriterWithConfiguration(properties, DeliveryGuarantee.AT_LEAST_ONCE);
- try {
- writer.setAsyncProducerException(
- new IOException("previous send request encountered error."));
- assertThatCode(() -> writer.write(1, SINK_WRITER_CONTEXT))
- .hasRootCauseExactlyInstanceOf(IOException.class);
-
- writer.setAsyncProducerException(
- new IOException("previous send request encountered error."));
- assertThatCode(() -> writer.flush(false))
- .hasRootCauseExactlyInstanceOf(IOException.class);
- } finally {
- writer.setAsyncProducerException(
- new IOException("previous send request encountered error."));
- assertThatCode(writer::close).hasRootCauseExactlyInstanceOf(IOException.class);
- }
- }
-
private void assertKafkaMetricNotPresent(
DeliveryGuarantee guarantee, String configKey, String configValue) throws Exception {
final Properties config = getKafkaClientConfiguration();
@@ -449,6 +522,18 @@ public class KafkaWriterITCase {
ImmutableList.of());
}
+ private KafkaWriter<Integer> createWriterWithConfiguration(
+ Properties config, DeliveryGuarantee guarantee, SinkInitContext sinkInitContext) {
+ return new KafkaWriter<>(
+ guarantee,
+ config,
+ "test-prefix",
+ sinkInitContext,
+ new DummyRecordSerializer(),
+ new DummySchemaContext(),
+ ImmutableList.of());
+ }
+
private static Properties getKafkaClientConfiguration() {
final Properties standardProps = new Properties();
standardProps.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
@@ -460,7 +545,7 @@ public class KafkaWriterITCase {
return standardProps;
}
- private static class SinkInitContext implements Sink.InitContext {
+ private static class SinkInitContext extends TestSinkInitContext {
private final SinkWriterMetricGroup metricGroup;
private final ProcessingTimeService timeService;
@@ -480,11 +565,6 @@ public class KafkaWriterITCase {
throw new UnsupportedOperationException("Not implemented.");
}
- @Override
- public MailboxExecutor getMailboxExecutor() {
- return new SyncMailboxExecutor();
- }
-
@Override
public ProcessingTimeService getProcessingTimeService() {
return timeService;
[flink-connector-kafka] 02/05: [FLINK-31620] [upsert-kafka] Flush wrapped writer
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 15b51f17a1daa2edb4cd17b4100eb407d83c8eb2
Author: Alex Sorokoumov <al...@gmail.com>
AuthorDate: Sun Mar 26 15:36:30 2023 -0700
[FLINK-31620] [upsert-kafka] Flush wrapped writer
- with the solution, the actual flushing only still happens on checkpoints
- when buffer size is full or periodically triggered, we don't actually do flushing on the wrapped writer.
This closes #17.
---
.../streaming/connectors/kafka/table/ReducingUpsertWriter.java | 9 +++++----
.../connectors/kafka/table/ReducingUpsertWriterTest.java | 7 ++++++-
2 files changed, 11 insertions(+), 5 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java
index 5a87add..67df4a6 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriter.java
@@ -87,7 +87,8 @@ class ReducingUpsertWriter<WriterState>
@Override
public void flush(boolean endOfInput) throws IOException, InterruptedException {
- flush();
+ sinkBuffer();
+ wrappedWriter.flush(endOfInput);
}
@Override
@@ -109,7 +110,7 @@ class ReducingUpsertWriter<WriterState>
reduceBuffer.put(key, new Tuple2<>(changeFlag(value), timestamp));
if (reduceBuffer.size() >= batchMaxRowNums) {
- flush();
+ sinkBuffer();
}
}
@@ -121,7 +122,7 @@ class ReducingUpsertWriter<WriterState>
lastFlush + batchIntervalMs,
(t) -> {
if (t >= lastFlush + batchIntervalMs) {
- flush();
+ sinkBuffer();
}
registerFlush();
});
@@ -140,7 +141,7 @@ class ReducingUpsertWriter<WriterState>
return value;
}
- private void flush() throws IOException, InterruptedException {
+ private void sinkBuffer() throws IOException, InterruptedException {
for (Tuple2<RowData, Long> value : reduceBuffer.values()) {
wrappedContext.setTimestamp(value.f1);
wrappedWriter.write(value.f0, wrappedContext);
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriterTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriterTest.java
index b6f2788..1ad9d09 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriterTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertWriterTest.java
@@ -222,6 +222,7 @@ public class ReducingUpsertWriterTest {
writeData(bufferedWriter, new ReusableIterator(0, 4));
// snapshot should flush the buffer
bufferedWriter.flush(true);
+ assertThat(writer.flushed).isTrue();
HashMap<Integer, List<RowData>> expected = new HashMap<>();
expected.put(
@@ -328,6 +329,8 @@ public class ReducingUpsertWriterTest {
private static class MockedSinkWriter
implements StatefulSink.StatefulSinkWriter<RowData, Void> {
+ boolean flushed = false;
+
transient List<RowData> rowDataCollectors;
MockedSinkWriter() {
@@ -343,7 +346,9 @@ public class ReducingUpsertWriterTest {
}
@Override
- public void flush(boolean endOfInput) throws IOException, InterruptedException {}
+ public void flush(boolean endOfInput) throws IOException, InterruptedException {
+ flushed = true;
+ }
@Override
public void close() throws Exception {}