[Hotfix][Connector-V2][kafka] code review

apache · Oct 21, 2024 · e63cee9 · e63cee9
1 parent 1f1c6a3
commit e63cee9
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 16 deletions.
diff --git a/...ain/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java b/...ain/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java
@@ -56,6 +56,7 @@ public KafkaTransactionSender(String transactionPrefix, Properties kafkaProperti
     @Override
     public void send(ProducerRecord<K, V> producerRecord) {
         kafkaProducer.send(producerRecord);
+        recordNumInTransaction++;
     }
 
     @Override

diff --git a/...e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java b/...e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -770,11 +770,22 @@ public void testKafkaProtobufToAssert(TestContainer container)
     }
 
     @TestTemplate
-    public void testKafkaExactlyOnce(TestContainer container) throws Exception {
+    public void testKafkaToKafkaExactlyOnce(TestContainer container) throws Exception {
+        TextSerializationSchema serializer =
+                TextSerializationSchema.builder()
+                        .seaTunnelRowType(SEATUNNEL_ROW_TYPE)
+                        .delimiter(",")
+                        .build();
+        generateTestData(
+                row ->
+                        new ProducerRecord<>(
+                                "kafka_topic_exactly_once", null, serializer.serialize(row)),
+                0,
+                10);
         container.executeJob("/kafka/fake_to_kafka_exactly_once.conf");
         String topicName = "kafka_topic_exactly_once";
         Map<String, String> data = getKafkaConsumerData(topicName);
-        Assertions.assertEquals(4, data.size());
+        Assertions.assertEquals(10, data.size());
     }
 
     public static String getTestConfigFile(String configFile)

diff --git a/...ces/kafka/fake_to_kafka_exactly_once.conf → ...es/kafka/kafka_to_kafka_exactly_once.conf b/...ces/kafka/fake_to_kafka_exactly_once.conf → ...es/kafka/kafka_to_kafka_exactly_once.conf
@@ -16,25 +16,20 @@
 #
 env {
     execution.parallelism = 1
-    job.mode = "BATCH"
+    job.mode = "STREAMING"
   }
 
 source {
-    FakeSource {
-      parallelism = 1
-      result_table_name = "fake"
-      split.read-interval = 30000
-      split.num = 2
-      row.num = 4
-      schema = {
-        fields {
-          name = "string"
-          age = "int"
-        }
-      }
-    }
+  Kafka {
+    bootstrap.servers = "kafkaCluster:9092"
+    topic = "kafka_topic_exactly_once"
+    result_table_name = "kafka_topic_exactly_once"
+    # The default format is json, which is optional
+    format = json
+    start_mode = earliest
   }
 
+}
 transform {}