You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2025/02/27 13:05:00 UTC
(flink-connector-kafka) branch main updated: [FLINK-37380][connector/kafka] Change TransactionalIdPrefix as required option when Exactly-Once enabled
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 7c112abe [FLINK-37380][connector/kafka] Change TransactionalIdPrefix as required option when Exactly-Once enabled
7c112abe is described below
commit 7c112abe8bf78e0cd8a310aaa65b57f6a70ad30a
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Thu Feb 27 21:04:54 2025 +0800
[FLINK-37380][connector/kafka] Change TransactionalIdPrefix as required option when Exactly-Once enabled
This closes #156.
---
.../flink/connector/kafka/sink/KafkaSinkBuilder.java | 2 +-
.../connector/kafka/sink/KafkaSinkBuilderTest.java | 18 ++++++++++++++++--
.../kafka/table/KafkaChangelogTableITCase.java | 1 +
.../pyflink/datastream/connectors/tests/test_kafka.py | 1 +
4 files changed, 19 insertions(+), 3 deletions(-)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java
index f0c20cfc..ffd4177b 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java
@@ -70,7 +70,7 @@ public class KafkaSinkBuilder<IN> {
private static final int MAXIMUM_PREFIX_BYTES = 64000;
private DeliveryGuarantee deliveryGuarantee = DeliveryGuarantee.NONE;
- private String transactionalIdPrefix = "kafka-sink";
+ private String transactionalIdPrefix;
private final Properties kafkaProducerConfig;
private KafkaRecordSerializationSchema<IN> recordSerializer;
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilderTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilderTest.java
index 3e2beb9a..9e26cd4e 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilderTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilderTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.kafka.sink;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.util.TestLogger;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -28,6 +29,7 @@ import java.util.Properties;
import java.util.function.Consumer;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link KafkaSinkBuilder}. */
public class KafkaSinkBuilderTest extends TestLogger {
@@ -41,7 +43,7 @@ public class KafkaSinkBuilderTest extends TestLogger {
};
@Test
- public void testPropertyHandling() {
+ void testPropertyHandling() {
validateProducerConfig(
getBasicBuilder(),
p -> {
@@ -77,7 +79,7 @@ public class KafkaSinkBuilderTest extends TestLogger {
}
@Test
- public void testBootstrapServerSetting() {
+ void testBootstrapServerSetting() {
Properties testConf1 = new Properties();
testConf1.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "testServer");
@@ -86,6 +88,18 @@ public class KafkaSinkBuilderTest extends TestLogger {
p -> assertThat(p).containsKeys(DEFAULT_KEYS));
}
+ @Test
+ void testTransactionIdSanityCheck() {
+ assertThatThrownBy(
+ () ->
+ getBasicBuilder()
+ .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
+ .build())
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessageContaining(
+ "EXACTLY_ONCE delivery guarantee requires a transactionIdPrefix to be set to provide unique transaction names across multiple KafkaSinks writing to the same Kafka cluster.");
+ }
+
private void validateProducerConfig(
KafkaSinkBuilder<?> builder, Consumer<Properties> validator) {
validator.accept(builder.build().getKafkaProducerConfig());
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
index 632b74ac..cd8e58a7 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
@@ -474,6 +474,7 @@ public class KafkaChangelogTableITCase extends KafkaTableTestBase {
.setPartitioner(partitioner)
.build())
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
+ .setTransactionalIdPrefix("kafka-sink")
.build());
env.execute("Write sequence");
}
diff --git a/flink-python/pyflink/datastream/connectors/tests/test_kafka.py b/flink-python/pyflink/datastream/connectors/tests/test_kafka.py
index dea06b3e..931f296e 100644
--- a/flink-python/pyflink/datastream/connectors/tests/test_kafka.py
+++ b/flink-python/pyflink/datastream/connectors/tests/test_kafka.py
@@ -472,6 +472,7 @@ class KafkaSinkTests(PyFlinkStreamingTestCase):
sink = KafkaSink.builder() \
.set_bootstrap_servers('localhost:9092') \
.set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE) \
+ .set_transactional_id_prefix("kafka-sink") \
.set_record_serializer(self._build_serialization_schema()) \
.build()
guarantee = get_field_value(sink.get_java_function(), 'deliveryGuarantee')