This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b3db27900bd [Test][E2E] Cover a kafka connect json tombstone in the
Kafka format IT (#13010)
b3db27900bd is described below
commit b3db27900bd4da855f0247e791e672f7fe301a4e
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Sun Nov 1 10:52:18 2026 +0000
[Test][E2E] Cover a kafka connect json tombstone in the Kafka format IT
(#13010)
---
.../e2e/connector/kafka/KafkaFormatIT.java | 208 +++++++++++++++++++--
...kafkasource_jdbc_record_tombstone_to_pgsql.conf | 58 ++++++
2 files changed, 254 insertions(+), 12 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
index 9faa9e90773..68e97d453cb 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
@@ -41,14 +41,17 @@ import org.apache.seatunnel.e2e.common.util.DependencyJar;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -90,6 +93,7 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -123,6 +127,8 @@ public class KafkaFormatIT extends TestSuiteBase implements
TestResource {
// ---------------------------Compatible Format
Parameter---------------------------------------
private static final String COMPATIBLE_DATA_PATH =
"/compatible/compatible_data.txt";
private static final String COMPATIBLE_KAFKA_SOURCE_TOPIC =
"jdbc_source_record";
+ private static final String COMPATIBLE_TOMBSTONE_CONFIG =
+
"/compatibleFormatIT/kafkasource_jdbc_record_tombstone_to_pgsql.conf";
// ---------------------------Debezium Format Parameter
---------------------------------------
private static final String DEBEZIUM_KAFKA_SINK_TOPIC =
"test-debezium-sink";
@@ -443,6 +449,84 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
checkCompatibleFormat();
}
+ /**
+ * A tombstone, a record carrying a key and a {@code null} value, has to
produce no row and let
+ * the source carry on past it. The guard that makes that happen sits at
the top of {@code
+ * CompatibleKafkaConnectDeserializationSchema.deserialize}, before {@code
convertToSinkRecord},
+ * which is the same branch the format module unit tests cover.
+ *
+ * <p>Restricted to Zeta because the committed offset assertion needs the
offsets to be
+ * committed when a batch job finishes, which the other engines do not do.
+ */
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "flink and spark won't commit offset when batch
job finished")
+ @TestTemplate
+ public void testFormatCompatibleTombstoneCheck(TestContainer container)
+ throws IOException, InterruptedException, ExecutionException {
+
+ LOG.info("====================== Check Compatible Tombstone
====================== ");
+ // The broker is shared by every test in this class, so this one
publishes to its own
+ // topic under its own consumer group. Putting a tombstone on
jdbc_source_record would
+ // change the fixture testFormatCompatibleCheck reads, and a shared
group would carry
+ // committed offsets in from the other lanes.
+ String topic =
+ COMPATIBLE_KAFKA_SOURCE_TOPIC + "_tombstone_" +
container.identifier().name();
+ createKafkaTopic(topic);
+
+ // The envelopes the compatible fixture already carries, with a
tombstone inserted before
+ // the last one. Everything after the tombstone still has to arrive,
so the expectation
+ // stays exactly the one testFormatCompatibleCheck asserts.
+ List<String> envelopes = readCompatibleEnvelopes();
+ try (KafkaProducer<String, String> producer =
+ new KafkaProducer<>(kafkaProducerProperties())) {
+ for (int i = 0; i < envelopes.size() - 1; i++) {
+ producer.send(new ProducerRecord<>(topic, null,
envelopes.get(i))).get();
+ }
+ producer.send(new ProducerRecord<>(topic, "tombstone-key",
null)).get();
+ producer.send(new ProducerRecord<>(topic, null,
envelopes.get(envelopes.size() - 1)))
+ .get();
+ producer.flush();
+ }
+ // One record per envelope, plus the tombstone.
+ long endOffset = envelopes.size() + 1L;
+
+ // Both values of format_error_handle_way. The guard returns before
any format specific
+ // parsing runs, so nothing is thrown, the emitter's catch block is
never entered, and
+ // the mode stops deciding what a tombstone does.
+ for (String errorHandleWay : Arrays.asList("FAIL", "SKIP")) {
+ // A group per run, so the second run has to commit an offset of
its own instead of
+ // passing on the one the first run already left behind.
+ String consumerGroup = topic + "_group_" + errorHandleWay;
+ awaitGroupCoordinator(consumerGroup);
+ // Reading the sink table also truncates it, so the check below
sees only the rows
+ // this run writes.
+ getPostgreSinkTableList(PG_SINK_TABLE1);
+ Container.ExecResult execCompatibleTombstoneResultToPgSql =
+ container.executeJob(
+ COMPATIBLE_TOMBSTONE_CONFIG,
+ Arrays.asList(
+ "topic=" + topic,
+ "consumerGroup=" + consumerGroup,
+ "errorHandleWay=" + errorHandleWay));
+ Assertions.assertEquals(
+ 0,
+ execCompatibleTombstoneResultToPgSql.getExitCode(),
+ "format_error_handle_way="
+ + errorHandleWay
+ + ": "
+ +
execCompatibleTombstoneResultToPgSql.getStderr());
+
+ // The tombstone produced no row and every real record still
arrived.
+ checkCompatibleFormat();
+
+ // And the offset moved past the tombstone, so a restart does not
replay it. A row
+ // count on its own would not catch that, which is why this
assertion is here.
+ assertCommittedOffsetReached(topic, consumerGroup, endOffset,
errorHandleWay);
+ }
+ }
+
@TestTemplate
public void testFormatMaxWellCheck(TestContainer container)
throws IOException, InterruptedException {
@@ -888,6 +972,117 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
Assertions.assertIterableEquals(expected, actual);
}
+ /**
+ * Reads the envelopes the compatible format fixture already carries, so
the tombstone test
+ * publishes the same payload shape and can reuse {@link
#checkCompatibleFormat()} as its
+ * expectation.
+ */
+ private List<String> readCompatibleEnvelopes() throws IOException {
+ InputStream inputStream =
KafkaFormatIT.class.getResourceAsStream(COMPATIBLE_DATA_PATH);
+ Assertions.assertNotNull(inputStream, "missing fixture " +
COMPATIBLE_DATA_PATH);
+ List<String> envelopes = new ArrayList<>();
+ try (BufferedReader reader = new BufferedReader(new
InputStreamReader(inputStream))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ envelopes.add(line);
+ }
+ }
+ Assertions.assertTrue(
+ envelopes.size() > 1,
+ "the compatible fixture needs at least two records to place a
tombstone between");
+ return envelopes;
+ }
+
+ private void createKafkaTopic(String topicName) {
+ try (AdminClient adminClient = createKafkaAdmin()) {
+ Set<String> existingTopics =
adminClient.listTopics().names().get(30, TimeUnit.SECONDS);
+ if (existingTopics.contains(topicName)) {
+ return;
+ }
+ adminClient
+ .createTopics(Arrays.asList(new NewTopic(topicName, 1,
(short) 1)))
+ .all()
+ .get(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while creating Kafka topic
" + topicName, e);
+ } catch (ExecutionException | TimeoutException e) {
+ throw new RuntimeException("Failed to create Kafka topic " +
topicName, e);
+ }
+ }
+
+ /**
+ * Waits until the broker can coordinate the consumer group. A new group
that commits first in a
+ * fresh broker races the creation of {@code __consumer_offsets}: the
commit fails with {@code
+ * CoordinatorNotAvailableException} and the source only logs it. The
admin client retries the
+ * coordinator lookup itself, and this commits nothing for the group.
+ */
+ private void awaitGroupCoordinator(String consumerGroup) {
+ try (AdminClient adminClient = createKafkaAdmin()) {
+ adminClient
+ .describeConsumerGroups(Arrays.asList(consumerGroup))
+ .all()
+ .get(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "Interrupted while waiting for the coordinator of " +
consumerGroup, e);
+ } catch (ExecutionException | TimeoutException e) {
+ throw new RuntimeException("No coordinator for consumer group " +
consumerGroup, e);
+ }
+ }
+
+ /**
+ * Asserts the consumer group committed an offset past the tombstone. The
source commits from
+ * {@code notifyCheckpointComplete}, which runs once the job has already
reported success, so
+ * this polls instead of reading the offset once.
+ */
+ private void assertCommittedOffsetReached(
+ String topic, String consumerGroup, long expectedOffset, String
errorHandleWay) {
+ TopicPartition topicPartition = new TopicPartition(topic, 0);
+ ListConsumerGroupOffsetsOptions options =
+ new ListConsumerGroupOffsetsOptions()
+ .topicPartitions(Arrays.asList(topicPartition));
+ try (AdminClient adminClient = createKafkaAdmin()) {
+ await().atMost(60, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Map<TopicPartition, OffsetAndMetadata>
committed =
+ adminClient
+
.listConsumerGroupOffsets(consumerGroup, options)
+
.partitionsToOffsetAndMetadata()
+ .get(30, TimeUnit.SECONDS);
+ OffsetAndMetadata offsetAndMetadata =
committed.get(topicPartition);
+ Assertions.assertNotNull(
+ offsetAndMetadata,
+ "format_error_handle_way="
+ + errorHandleWay
+ + ": no committed offset for "
+ + topicPartition);
+ Assertions.assertEquals(
+ expectedOffset,
+ offsetAndMetadata.offset(),
+ "format_error_handle_way="
+ + errorHandleWay
+ + ": the committed offset must
sit past the"
+ + " tombstone");
+ });
+ }
+ }
+
+ private Properties kafkaProducerProperties() {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
+ props.put("acks", "all");
+ props.put("retries", 0);
+ props.put("linger.ms", 1);
+ props.put("buffer.memory", 33554432);
+ props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
+ props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
+ return props;
+ }
+
// Initialize the kafka Consumer
private void initKafkaConsumer() {
Properties prop = new Properties();
@@ -1025,18 +1220,7 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
// Initialize ogg data to kafka
private void initLocalDataToKafka() {
- String bootstrapServers = KAFKA_CONTAINER.getBootstrapServers();
- Properties props = new Properties();
-
- props.put("bootstrap.servers", bootstrapServers);
- props.put("acks", "all");
- props.put("retries", 0);
- props.put("linger.ms", 1);
- props.put("buffer.memory", 33554432);
- props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
- props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
-
- KafkaProducer<String, String> producer = new KafkaProducer<>(props);
+ KafkaProducer<String, String> producer = new
KafkaProducer<>(kafkaProducerProperties());
for (String localPath : LOCAL_DATA_TO_KAFKA_MAPPING.keySet()) {
String kafkaTopic = LOCAL_DATA_TO_KAFKA_MAPPING.get(localPath);
InputStream inputStream =
KafkaFormatIT.class.getResourceAsStream(localPath);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/compatibleFormatIT/kafkasource_jdbc_record_tombstone_to_pgsql.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/compatibleFormatIT/kafkasource_jdbc_record_tombstone_to_pgsql.conf
new file mode 100644
index 00000000000..f4163d28ca1
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/compatibleFormatIT/kafkasource_jdbc_record_tombstone_to_pgsql.conf
@@ -0,0 +1,58 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafka_e2e:9092"
+ topic = "${topic}"
+ consumer.group = "${consumerGroup}"
+ plugin_output = "kafka_table"
+ start_mode = earliest
+ # Driven from the test with both FAIL and SKIP. A tombstone is a deletion
marker rather
+ # than a format error, so reading one must not depend on which value is
configured.
+ format_error_handle_way = "${errorHandleWay}"
+ schema = {
+ fields {
+ id = "int"
+ name = "string"
+ description = "string"
+ weight = "string"
+ }
+ },
+ format = COMPATIBLE_KAFKA_CONNECT_JSON
+ }
+}
+
+
+sink {
+ Jdbc {
+
+ driver = org.postgresql.Driver
+ url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
+ user = test
+ password = test
+ generate_sink_sql = true
+ database = test
+ table = public.sink
+ primary_keys = ["id"]
+ }
+}
