This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 7be1b0e464 [Test][E2E] Stabilize Kafka exactly-once record reads
(#11205)
7be1b0e464 is described below
commit 7be1b0e4643451c448ca0a909c1ada8a118824d7
Author: Daniel <[email protected]>
AuthorDate: Mon Jun 29 22:35:37 2026 +0800
[Test][E2E] Stabilize Kafka exactly-once record reads (#11205)
---
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 81 +++++++++++++++++-----
1 file changed, 63 insertions(+), 18 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index f91c1827a1..72ac59dc22 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -1689,6 +1689,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
createKafkaTopic(producerTopic);
createKafkaTopic(consumerTopic);
String sourceData = "Seatunnel Exactly Once Example";
+ String keepAliveData = sourceData + "-keepalive-" + resourceSuffix;
long sinkStartOffset = endOffsetOnP0(consumerTopic);
for (int i = 0; i < 10; i++) {
ProducerRecord<byte[], byte[]> record =
@@ -1716,9 +1717,24 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
.await()
.atMost(5, MINUTES)
.untilAsserted(
- () ->
- Assertions.assertTrue(
- checkData(consumerTopic,
sinkStartOffset, 10, sourceData)));
+ () -> {
+ // Keep the streaming source active so the last
exactly-once transaction
+ // is forced through a later checkpoint on slow
Flink CI axes.
+ ProducerRecord<byte[], byte[]> keepAliveRecord =
+ new ProducerRecord<>(
+ producerTopic,
+ null,
+
keepAliveData.getBytes(StandardCharsets.UTF_8));
+ producer.send(keepAliveRecord);
+ producer.flush();
+ Assertions.assertTrue(
+ checkData(
+ consumerTopic,
+ sinkStartOffset,
+ 10,
+ sourceData,
+
Collections.singletonList(keepAliveData)));
+ });
}
@TestTemplate
@@ -1754,21 +1770,40 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
// Compare the values of data fields obtained from consumers
private boolean checkData(String topicName, long startOffset, long
expectedCount, String data) {
- List<String> listData = getKafkaConsumerListData(topicName,
startOffset, expectedCount);
- if (listData.isEmpty() || listData.size() != expectedCount) {
+ return checkData(topicName, startOffset, expectedCount, data,
Collections.emptyList());
+ }
+
+ private boolean checkData(
+ String topicName,
+ long startOffset,
+ long expectedCount,
+ String data,
+ List<String> ignoredValues) {
+ List<String> listData = getKafkaConsumerListData(topicName,
startOffset);
+ List<String> matchedData = new ArrayList<>();
+ for (String value : listData) {
+ if (data.equals(value)) {
+ matchedData.add(value);
+ continue;
+ }
+ if (ignoredValues.contains(value)) {
+ continue;
+ }
log.error(
- "testKafkaToKafkaExactlyOnce get data size is not
expect,get consumer data size {},start offset {},expected count {}",
+ "testKafkaToKafkaExactlyOnce get unexpected data value {},
start offset {}",
+ value,
+ startOffset);
+ return false;
+ }
+ if (matchedData.isEmpty() || matchedData.size() != expectedCount) {
+ log.error(
+ "testKafkaToKafkaExactlyOnce get data size is not
expect,get matched data size {},visible data size {},start offset {},expected
count {}",
+ matchedData.size(),
listData.size(),
startOffset,
expectedCount);
return false;
}
- for (String value : listData) {
- if (!data.equals(value)) {
- log.error("testKafkaToKafkaExactlyOnce get data value is not
expect");
- return false;
- }
- }
return true;
}
@@ -2305,8 +2340,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
}
- private List<String> getKafkaConsumerListData(
- String topicName, long startOffset, long expectedCount) {
+ private List<String> getKafkaConsumerListData(String topicName, long
startOffset) {
KafkaConsumer<String, String> consumer = null;
try {
List<String> data = new ArrayList<>();
@@ -2314,17 +2348,28 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
TopicPartition topicPartition = new TopicPartition(topicName, 0);
consumer.assign(Collections.singletonList(topicPartition));
consumer.seek(topicPartition, startOffset);
- long targetOffsetExclusive = startOffset + expectedCount;
+ // READ_COMMITTED consumers may skip aborted transactional
offsets, so N committed
+ // records are not guaranteed to occupy N contiguous offsets after
startOffset.
+ long visibleEndOffsetExclusive =
+
consumer.endOffsets(Collections.singletonList(topicPartition))
+ .get(topicPartition);
Long lastProcessedOffset = startOffset - 1;
+ int consecutiveEmptyPolls = 0;
do {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
- for (ConsumerRecord<String, String> record : records) {
- if (record.offset() >= startOffset && record.offset() <
targetOffsetExclusive) {
+ if (records.isEmpty()) {
+ consecutiveEmptyPolls++;
+ continue;
+ }
+ consecutiveEmptyPolls = 0;
+ for (ConsumerRecord<String, String> record :
records.records(topicPartition)) {
+ if (record.offset() >= startOffset && record.offset() >
lastProcessedOffset) {
data.add(record.value());
}
lastProcessedOffset = record.offset();
}
- } while (lastProcessedOffset < targetOffsetExclusive - 1);
+ } while (lastProcessedOffset < visibleEndOffsetExclusive - 1
+ && consecutiveEmptyPolls < 20);
return data;
} finally {
closeKafkaConsumer(consumer);
