This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13010-b434d2e4f46807a82b0cf4b1a6b52c1e88fd0532 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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"] + } +}
