[FLINK-37380][connector/kafka] Change TransactionalIdPrefix as required option when Exactly-Once enabled

This closes #156.

Author: loserwang1024

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;

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.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 void testPropertyHandling() {
     }
 
     @Test
-    public void testBootstrapServerSetting() {
+    void testBootstrapServerSetting() {
         Properties testConf1 = new Properties();
         testConf1.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "testServer");
 
@@ -86,6 +88,18 @@ public void testBootstrapServerSetting() {
                 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());

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java

@@ -474,6 +474,7 @@ private void writeRecordsToKafka(String topic, List<String> lines) throws Except
                                         .setPartitioner(partitioner)
                                         .build())
                         .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
+                        .setTransactionalIdPrefix("kafka-sink")
                         .build());
         env.execute("Write sequence");
     }

flink-python/pyflink/datastream/connectors/tests/test_kafka.py

@@ -472,6 +472,7 @@ def test_set_delivery_guarantee(self):
         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')