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 b810f07479 [Test][E2E] Stabilize Kafka restore start-mode tests
(#11426)
b810f07479 is described below
commit b810f07479d8a8c46b68c1b4dedb6c39124a0eeb
Author: Daniel <[email protected]>
AuthorDate: Thu Jul 30 23:48:53 2026 +0800
[Test][E2E] Stabilize Kafka restore start-mode tests (#11426)
Co-authored-by: DanielLeens <[email protected]>
---
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 459 +++++++++++++++------
.../kafkasource_restore_with_earliest_mode.conf | 6 +-
.../kafkasource_restore_with_latest_mode.conf | 6 +-
...asource_restore_with_specific_offsets_mode.conf | 6 +-
.../kafkasource_restore_with_timestamp_mode.conf | 6 +-
5 files changed, 339 insertions(+), 144 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 f906bab4c6..ea1fc779d9 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
@@ -105,14 +105,17 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -808,10 +811,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
public void testSourceKafkaRestoreWithEarliestMode(TestContainer container)
throws IOException, InterruptedException {
- final String sourceTopic = "test_topic_restore_earliest";
- final String sinkTopic = "test_topic_restore_earliest_output";
+ String resourceSuffix = startModeTopic(container, "restore-earliest");
+ final String sourceTopic = resourceSuffix + "_source";
+ final String sinkTopic = resourceSuffix + "_sink";
+ final String consumerGroup = resourceSuffix + "_group";
final String payload = "Seatunnel Restore Test Data";
final String jobId = "18696753645408";
+ List<String> restoreVariables =
+ buildKafkaRestoreVariables(sourceTopic, sinkTopic,
consumerGroup);
+
+ // The Kafka broker is shared across all engine variants in this
class, so restore tests
+ // need dedicated topics and groups to avoid inheriting offsets from
earlier template runs.
+ createKafkaTopic(sourceTopic);
+ createKafkaTopic(sinkTopic);
// Write 20 initial records with unique keys (avoid any potential
dedup logic
// elsewhere).
@@ -825,18 +837,22 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
long srcEndBeforeStart = endOffsetOnP0(sourceTopic);
// Start the first streaming job asynchronously.
- CompletableFuture.runAsync(
- () -> {
- try {
- container.executeJob(
-
"/kafka/kafkasource_restore_with_earliest_mode.conf", jobId);
- } catch (Exception e) {
- log.error("First job execution exception", e);
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> firstJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/kafka/kafkasource_restore_with_earliest_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ log.error("First job execution exception", e);
+ throw new RuntimeException(e);
+ }
+ });
awaitJobRunning(container, jobId);
+ waitForStreamingJobStartup(firstJobFuture);
// Produce 10 additional records after the job starts.
for (int i = 0; i < 10; i++) {
@@ -857,7 +873,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun);
// Savepoint the running job (so restore should continue from this
position).
- container.savepointJob(jobId);
+ assertCommandSucceeded("Savepoint", container.savepointJob(jobId));
+ assertCommandSucceeded("Initial job",
waitForJobResult(firstJobFuture));
// Append 15 records after savepoint, used to validate restore
progress.
for (int i = 0; i < 15; i++) {
@@ -879,18 +896,24 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
// Deliberately move the consumer-group offset past the savepoint
position.
// Restore must still use the checkpointed split offsets instead of
the external group
// offset, otherwise the 15 post-savepoint records will be skipped.
- commitOffset(sourceTopic, "test_restore_group", srcEndAfterAll);
+ commitOffset(sourceTopic, consumerGroup, srcEndAfterAll);
// Restore the job from the savepoint asynchronously.
- CompletableFuture.runAsync(
- () -> {
- try {
- container.restoreJob(
-
"/kafka/kafkasource_restore_with_earliest_mode.conf", jobId);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> restoredJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.restoreJob(
+
"/kafka/kafkasource_restore_with_earliest_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ awaitJobRunning(container, jobId);
+ waitForStreamingJobStartup(restoredJobFuture);
// After restore, sink should advance by the 15 newly produced records
at
// minimum.
@@ -898,6 +921,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15);
+ assertCommandSucceeded("Restored job cancellation",
container.cancelJob(jobId));
+ assertCommandSucceeded("Restored job",
waitForJobResult(restoredJobFuture));
}
// ------------------------------ LATEST MODE
------------------------------
@@ -909,12 +934,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
public void testSourceKafkaRestoreWithLatestMode(TestContainer container)
throws IOException, InterruptedException {
- final String sourceTopic = "test_topic_restore_latest";
- final String sinkTopic = "test_topic_restore_latest_output";
+ String resourceSuffix = startModeTopic(container, "restore-latest");
+ final String sourceTopic = resourceSuffix + "_source";
+ final String sinkTopic = resourceSuffix + "_sink";
+ final String consumerGroup = resourceSuffix + "_group";
final String payload = "Seatunnel Restore Test Data Latest";
final String firstRunPayload = payload + "_additional";
final String restorePayload = payload + "_restore";
final String jobId = "18696753645410";
+ List<String> restoreVariables =
+ buildKafkaRestoreVariables(sourceTopic, sinkTopic,
consumerGroup);
+
+ createKafkaTopic(sourceTopic);
+ createKafkaTopic(sinkTopic);
// Write 20 initial records before starting the job.
for (int i = 0; i < 20; i++) {
@@ -925,18 +957,22 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
long srcEndBeforeStart = endOffsetOnP0(sourceTopic);
- CompletableFuture.runAsync(
- () -> {
- try {
- container.executeJob(
-
"/kafka/kafkasource_restore_with_latest_mode.conf", jobId);
- } catch (Exception e) {
- log.error("First job execution exception", e);
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> firstJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/kafka/kafkasource_restore_with_latest_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ log.error("First job execution exception", e);
+ throw new RuntimeException(e);
+ }
+ });
awaitJobRunning(container, jobId);
+ waitForStreamingJobStartup(firstJobFuture);
AtomicInteger latestModeProbeId = new AtomicInteger();
Awaitility.await()
@@ -978,7 +1014,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(
10, countKafkaMessages(sinkTopic,
firstRunPayload)));
- container.savepointJob(jobId);
+ assertCommandSucceeded("Savepoint", container.savepointJob(jobId));
+ assertCommandSucceeded("Initial job",
waitForJobResult(firstJobFuture));
// Append 15 more records after savepoint.
for (int i = 0; i < 15; i++) {
@@ -995,15 +1032,20 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
srcEndAfterAll >= srcEndBeforeStart + 25,
"Final end offset should advance by at least 25");
- CompletableFuture.runAsync(
- () -> {
- try {
- container.restoreJob(
-
"/kafka/kafkasource_restore_with_latest_mode.conf", jobId);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> restoredJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.restoreJob(
+
"/kafka/kafkasource_restore_with_latest_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ awaitJobRunning(container, jobId);
Awaitility.await()
.pollInterval(2, SECONDS)
@@ -1012,6 +1054,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
() ->
Assertions.assertEquals(
15, countKafkaMessages(sinkTopic,
restorePayload)));
+ assertCommandSucceeded("Restored job cancellation",
container.cancelJob(jobId));
+ assertCommandSucceeded("Restored job",
waitForJobResult(restoredJobFuture));
}
// ---------------------------- TIMESTAMP MODE
-----------------------------
@@ -1023,10 +1067,17 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
public void testSourceKafkaRestoreWithTimestampMode(TestContainer
container)
throws IOException, InterruptedException {
- final String sourceTopic = "test_topic_restore_timestamp";
- final String sinkTopic = "test_topic_restore_timestamp_output";
+ String resourceSuffix = startModeTopic(container, "restore-timestamp");
+ final String sourceTopic = resourceSuffix + "_source";
+ final String sinkTopic = resourceSuffix + "_sink";
+ final String consumerGroup = resourceSuffix + "_group";
final String payload = "Seatunnel Restore Test Data Timestamp";
final String jobId = "18696753645411";
+ List<String> restoreVariables =
+ buildKafkaRestoreVariables(sourceTopic, sinkTopic,
consumerGroup);
+
+ createKafkaTopic(sourceTopic);
+ createKafkaTopic(sinkTopic);
for (int i = 0; i < 20; i++) {
producer.send(
@@ -1036,16 +1087,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
long srcEndBeforeStart = endOffsetOnP0(sourceTopic);
- CompletableFuture.runAsync(
- () -> {
- try {
- container.executeJob(
-
"/kafka/kafkasource_restore_with_timestamp_mode.conf", jobId);
- } catch (Exception e) {
- log.error("First job execution exception", e);
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> firstJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/kafka/kafkasource_restore_with_timestamp_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ log.error("First job execution exception", e);
+ throw new RuntimeException(e);
+ }
+ });
awaitJobRunning(container, jobId);
@@ -1067,7 +1121,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
.atMost(2, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun);
- container.savepointJob(jobId);
+ assertCommandSucceeded("Savepoint", container.savepointJob(jobId));
+ assertCommandSucceeded("Initial job",
waitForJobResult(firstJobFuture));
// Append 15 more records after savepoint.
for (int i = 0; i < 15; i++) {
@@ -1084,20 +1139,27 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
srcEndAfterAll == srcEndBeforeStart + 25,
"Final end offset should advance by at least 25");
- CompletableFuture.runAsync(
- () -> {
- try {
- container.restoreJob(
-
"/kafka/kafkasource_restore_with_timestamp_mode.conf", jobId);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> restoredJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.restoreJob(
+
"/kafka/kafkasource_restore_with_timestamp_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ awaitJobRunning(container, jobId);
Awaitility.await()
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15);
+ assertCommandSucceeded("Restored job cancellation",
container.cancelJob(jobId));
+ assertCommandSucceeded("Restored job",
waitForJobResult(restoredJobFuture));
}
// ------------------------- SPECIFIC OFFSETS MODE
-------------------------
@@ -1110,9 +1172,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
throws IOException, InterruptedException {
final String sourceTopic = "test_topic_restore_specific_offsets";
- final String sinkTopic = "test_topic_restore_specific_offsets_output";
+ String resourceSuffix = startModeTopic(container,
"restore-specific-offsets");
+ final String sinkTopic = resourceSuffix + "_sink";
+ final String consumerGroup = resourceSuffix + "_group";
final String payload = "Seatunnel Restore Test Data Specific Offsets";
final String jobId = "18696753645412";
+ List<String> restoreVariables =
+ buildKafkaRestoreVariables(sourceTopic, sinkTopic,
consumerGroup);
+
+ // The specific_offsets config stores the source topic inside a map
key, and current
+ // variable replacement does not rewrite config keys. Recreate the
fixed source topic
+ // before each run so the test still gets isolated offsets without
changing parser rules.
+ recreateKafkaTopic(sourceTopic);
+ createKafkaTopic(sinkTopic);
for (int i = 0; i < 20; i++) {
producer.send(
@@ -1122,17 +1194,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
long srcEndBeforeStart = endOffsetOnP0(sourceTopic);
- CompletableFuture.runAsync(
- () -> {
- try {
- container.executeJob(
-
"/kafka/kafkasource_restore_with_specific_offsets_mode.conf",
- jobId);
- } catch (Exception e) {
- log.error("First job execution exception", e);
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> firstJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/kafka/kafkasource_restore_with_specific_offsets_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ log.error("First job execution exception", e);
+ throw new RuntimeException(e);
+ }
+ });
awaitJobRunning(container, jobId);
@@ -1154,7 +1228,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
.atMost(2, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun - 11);
- container.savepointJob(jobId);
+ assertCommandSucceeded("Savepoint", container.savepointJob(jobId));
+ assertCommandSucceeded("Initial job",
waitForJobResult(firstJobFuture));
// Append 15 more records after savepoint.
for (int i = 0; i < 15; i++) {
@@ -1171,21 +1246,27 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
srcEndAfterAll == srcEndBeforeStart + 25,
"Final end offset should advance by at least 25");
- CompletableFuture.runAsync(
- () -> {
- try {
- container.restoreJob(
-
"/kafka/kafkasource_restore_with_specific_offsets_mode.conf",
- jobId);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ CompletableFuture<Container.ExecResult> restoredJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.restoreJob(
+
"/kafka/kafkasource_restore_with_specific_offsets_mode.conf",
+ jobId,
+ restoreVariables.toArray(new
String[0]));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ awaitJobRunning(container, jobId);
Awaitility.await()
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
.until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15 - 11);
+ assertCommandSucceeded("Restored job cancellation",
container.cancelJob(jobId));
+ assertCommandSucceeded("Restored job",
waitForJobResult(restoredJobFuture));
}
private void awaitJobRunning(TestContainer container, String jobId) {
@@ -1196,6 +1277,69 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
() -> Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId)));
}
+ /**
+ * Wait for a streaming submit command to exit after savepoint or
cancellation.
+ *
+ * <p>A bounded wait prevents a failed lifecycle transition from leaving
the test thread blocked
+ * indefinitely.
+ */
+ private Container.ExecResult waitForJobResult(
+ CompletableFuture<Container.ExecResult> jobFuture) {
+ try {
+ return jobFuture.get(2, MINUTES);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("Interrupted while waiting for streaming
job completion", e);
+ } catch (ExecutionException | TimeoutException e) {
+ throw new AssertionError("Streaming job command did not complete
successfully", e);
+ }
+ }
+
+ /**
+ * Give a streaming submit command a short startup window before the test
publishes data that
+ * must only be observed after initialization.
+ */
+ private void
waitForStreamingJobStartup(CompletableFuture<Container.ExecResult> jobFuture) {
+ try {
+ jobFuture.get(15, SECONDS);
+ throw new AssertionError("Streaming job finished before startup
grace window elapsed");
+ } catch (TimeoutException expected) {
+ // Expected path: the job is still running and had time to
initialize.
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("Interrupted while waiting for streaming
job startup", e);
+ } catch (ExecutionException e) {
+ throw new AssertionError("Streaming job failed before startup
grace window elapsed", e);
+ }
+ }
+
+ /**
+ * Publish a probe record and wait until it becomes visible on the sink
topic so later
+ * assertions only count business records emitted after the Kafka source
is actively consuming.
+ */
+ private long awaitStreamingPipelineReady(
+ String producerTopic, String consumerTopic, long sinkStartOffset,
String probeData) {
+ sendTextRecordAndWait(producerTopic, probeData);
+ given().pollDelay(5, SECONDS)
+ .pollInterval(5, SECONDS)
+ .await()
+ .atMost(2, MINUTES)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+ checkData(consumerTopic,
sinkStartOffset, 1, probeData)));
+ return endOffsetOnP0(consumerTopic);
+ }
+
+ /**
+ * Assert that a lifecycle command completed successfully.
+ *
+ * <p>The command stderr is included so CI failures retain the
container-side error context.
+ */
+ private void assertCommandSucceeded(String action, Container.ExecResult
result) {
+ Assertions.assertEquals(0, result.getExitCode(), action + " failed: "
+ result.getStderr());
+ }
+
/**
* Get visible record count on partition-0: endOffset - beginningOffset
(exclusive upper bound).
*/
@@ -1926,51 +2070,38 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
createKafkaTopic(producerTopic);
createKafkaTopic(consumerTopic);
String sourceData = "Seatunnel Exactly Once Example";
- String keepAliveData = sourceData + "-keepalive-" + resourceSuffix;
+ String readinessProbeData = sourceData + "-probe-" + resourceSuffix;
long sinkStartOffset = endOffsetOnP0(consumerTopic);
- for (int i = 0; i < 10; i++) {
- ProducerRecord<byte[], byte[]> record =
- new ProducerRecord<>(producerTopic, null,
sourceData.getBytes());
- producer.send(record);
- producer.flush();
- }
// async execute
- CompletableFuture.supplyAsync(
- () -> {
- try {
- container.executeJob(
-
"/kafka/kafka_to_kafka_exactly_once_streaming.conf",
- exactlyOnceVariables);
- } catch (Exception e) {
- log.error("Commit task exception :" + e.getMessage());
- throw new RuntimeException(e);
- }
- return null;
- });
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/kafka/kafka_to_kafka_exactly_once_streaming.conf",
+ exactlyOnceVariables);
+ } catch (Exception e) {
+ log.error("Commit task exception :" +
e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+ waitForStreamingJobStartup(jobFuture);
+ long sinkDataStartOffset =
+ awaitStreamingPipelineReady(
+ producerTopic, consumerTopic, sinkStartOffset,
readinessProbeData);
+ for (int i = 0; i < 10; i++) {
+ sendTextRecordAndWait(producerTopic, sourceData);
+ }
// wait for data written to kafka
- given().pollDelay(120, SECONDS)
+ given().pollDelay(10, SECONDS)
.pollInterval(5, SECONDS)
.await()
.atMost(5, MINUTES)
.untilAsserted(
() -> {
- // 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)));
+ checkData(consumerTopic,
sinkDataStartOffset, 10, sourceData));
});
}
@@ -2010,6 +2141,23 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
return checkData(topicName, startOffset, expectedCount, data,
Collections.emptyList());
}
+ /**
+ * Sends one text record and waits for the broker acknowledgement so
producer-side delivery
+ * issues surface immediately instead of appearing later as assertion
flakes.
+ */
+ private void sendTextRecordAndWait(String topicName, String data) {
+ ProducerRecord<byte[], byte[]> record =
+ new ProducerRecord<>(topicName, null,
data.getBytes(StandardCharsets.UTF_8));
+ try {
+ producer.send(record).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while sending Kafka test
record", e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException("Failed to send Kafka test record to
topic " + topicName, e);
+ }
+ }
+
private boolean checkData(
String topicName,
long startOffset,
@@ -2318,15 +2466,18 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
/**
- * Create a dedicated Kafka topic for the exactly-once tests so each
method reads its own data
- * and never reuses offsets from earlier runs in the same class.
+ * Create a dedicated Kafka topic for streaming tests that require
isolated records and offsets.
+ *
+ * <p>Each registered topic is removed during class cleanup.
*/
private void createKafkaTopic(String topicName) {
NewTopic topic = new NewTopic(topicName, 1, (short) 1);
topic.configs(Collections.singletonMap("retention.ms", "-1"));
try (AdminClient adminClient = createKafkaAdmin()) {
adminClient.createTopics(Collections.singletonList(topic)).all().get();
- dynamicTopics.add(topicName);
+ if (!dynamicTopics.contains(topicName)) {
+ dynamicTopics.add(topicName);
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException(
@@ -2336,6 +2487,36 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
}
+ /**
+ * Recreate a fixed topic so restore tests can start from a clean
partition state without
+ * inheriting offsets or retained records from earlier runs.
+ */
+ private void recreateKafkaTopic(String topicName) {
+ try (AdminClient adminClient = createKafkaAdmin()) {
+
adminClient.deleteTopics(Collections.singletonList(topicName)).all().get();
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(30))
+ .untilAsserted(
+ () ->
+ Assertions.assertFalse(
+ adminClient
+ .listTopics()
+ .names()
+ .get()
+ .contains(topicName)));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "Interrupted while recreating Kafka topic " + topicName,
e);
+ } catch (ExecutionException e) {
+ log.info(
+ "Kafka topic {} did not exist before recreation or could
not be deleted cleanly: {}",
+ topicName,
+ e.getMessage());
+ }
+ createKafkaTopic(topicName);
+ }
+
/** Build the dynamic `-i key=value` variables for the exactly-once
streaming template. */
private List<String> buildExactlyOnceStreamingVariables(
String sourceTopic, String sinkTopic, String consumerGroup) {
@@ -2345,6 +2526,19 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
EXACTLY_ONCE_CONSUMER_GROUP_VARIABLE + "=" + consumerGroup);
}
+ /**
+ * Build the dynamic {@code -i key=value} variables for Kafka restore-mode
templates.
+ *
+ * <p>Each test invocation uses isolated Kafka topics and a consumer group.
+ */
+ private List<String> buildKafkaRestoreVariables(
+ String sourceTopic, String sinkTopic, String consumerGroup) {
+ return Arrays.asList(
+ "sourceTopic=" + sourceTopic,
+ "sinkTopic=" + sinkTopic,
+ "consumerGroup=" + consumerGroup);
+ }
+
private void initKafkaProducer() {
Properties props = new Properties();
String bootstrapServers = kafkaContainer.getBootstrapServers();
@@ -2581,6 +2775,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
KafkaConsumer<String, String> consumer = null;
try {
List<String> data = new ArrayList<>();
+ Set<Long> seenOffsets = new HashSet<>();
consumer = new KafkaConsumer<>(kafkaManualConsumerConfig());
TopicPartition topicPartition = new TopicPartition(topicName, 0);
consumer.assign(Collections.singletonList(topicPartition));
@@ -2590,7 +2785,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
long visibleEndOffsetExclusive =
consumer.endOffsets(Collections.singletonList(topicPartition))
.get(topicPartition);
- Long lastProcessedOffset = startOffset - 1;
+ long lastObservedOffset = startOffset - 1;
int consecutiveEmptyPolls = 0;
do {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
@@ -2600,12 +2795,12 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
consecutiveEmptyPolls = 0;
for (ConsumerRecord<String, String> record :
records.records(topicPartition)) {
- if (record.offset() >= startOffset && record.offset() >
lastProcessedOffset) {
+ if (record.offset() >= startOffset &&
seenOffsets.add(record.offset())) {
data.add(record.value());
}
- lastProcessedOffset = record.offset();
+ lastObservedOffset = Math.max(lastObservedOffset,
record.offset());
}
- } while (lastProcessedOffset < visibleEndOffsetExclusive - 1
+ } while (lastObservedOffset < visibleEndOffsetExclusive - 1
&& consecutiveEmptyPolls < 20);
return data;
} finally {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_earliest_mode.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_earliest_mode.conf
index ac4c7d9117..ed34c090d8 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_earliest_mode.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_earliest_mode.conf
@@ -25,10 +25,10 @@ env {
source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_earliest"
+ topic = "${sourceTopic}"
format = text
start_mode = earliest
- consumer.group = "test_restore_group"
+ consumer.group = "${consumerGroup}"
discovery.interval.millis = 10000
}
}
@@ -39,7 +39,7 @@ transform {
sink {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_earliest_output"
+ topic = "${sinkTopic}"
format = text
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_latest_mode.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_latest_mode.conf
index 00b7b38c0a..f2c295dd8d 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_latest_mode.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_latest_mode.conf
@@ -25,10 +25,10 @@ env {
source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_latest"
+ topic = "${sourceTopic}"
format = text
start_mode = latest
- consumer.group = "test_restore_latest"
+ consumer.group = "${consumerGroup}"
discovery.interval.millis = 10000
}
}
@@ -39,7 +39,7 @@ transform {
sink {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_latest_output"
+ topic = "${sinkTopic}"
format = text
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_specific_offsets_mode.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_specific_offsets_mode.conf
index bc929bb24b..b285240349 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_specific_offsets_mode.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_specific_offsets_mode.conf
@@ -25,13 +25,13 @@ env {
source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_specific_offsets"
+ topic = "${sourceTopic}"
format = text
start_mode = specific_offsets
start_mode.offsets = {
test_topic_restore_specific_offsets-0 = 11
}
- consumer.group = "test_restore_specific_offsets"
+ consumer.group = "${consumerGroup}"
discovery.interval.millis = 10000
}
}
@@ -42,7 +42,7 @@ transform {
sink {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_specific_offsets_output"
+ topic = "${sinkTopic}"
format = text
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_timestamp_mode.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_timestamp_mode.conf
index f42dae81f5..5fd230317f 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_timestamp_mode.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_restore_with_timestamp_mode.conf
@@ -25,11 +25,11 @@ env {
source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_timestamp"
+ topic = "${sourceTopic}"
format = text
start_mode = timestamp
start_mode.timestamp = 1738395840000
- consumer.group = "test_restore_timestamp"
+ consumer.group = "${consumerGroup}"
discovery.interval.millis = 10000
}
}
@@ -40,7 +40,7 @@ transform {
sink {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "test_topic_restore_timestamp_output"
+ topic = "${sinkTopic}"
format = text
}
}
