This is an automated email from the ASF dual-hosted git repository.
yzeng1618 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 0b8407a498 Split unrelated client summary and Pulsar E2E follow-ups
from PR 10491 (#10922)
0b8407a498 is described below
commit 0b8407a4988c1c0594a4ae6cafd38150e38c4e3c
Author: David Zollo <[email protected]>
AuthorDate: Wed May 27 17:04:12 2026 +0800
Split unrelated client summary and Pulsar E2E follow-ups from PR 10491
(#10922)
Co-authored-by: Daniel <[email protected]>
---
.../seatunnel/command/ClientExecuteCommand.java | 6 ++--
.../e2e/connector/pulsar/CanalToPulsarIT.java | 39 ++++++++++++++++++++--
2 files changed, 41 insertions(+), 4 deletions(-)
diff --git
a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java
b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java
index b5f5fb4327..c212a6e876 100644
---
a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java
+++
b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java
@@ -288,8 +288,10 @@ public class ClientExecuteCommand implements
Command<ClientCommandArgs> {
"Total Write Count",
jobMetricsSummary.getSinkWriteCount(),
"Total Failed Count",
- jobMetricsSummary.getSourceReadCount()
- -
jobMetricsSummary.getSinkWriteCount()));
+ Math.max(
+ 0,
+ jobMetricsSummary.getSourceReadCount()
+ -
jobMetricsSummary.getSinkWriteCount())));
}
closeClient();
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
index 8c2c132bc9..9e4552a7f3 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
@@ -34,7 +34,12 @@ import
org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -251,6 +256,31 @@ public class CanalToPulsarIT extends TestSuiteBase
implements TestResource {
}
}
+ private void waitForCanalDataInTopic() throws IOException,
PulsarClientException {
+ String topicFullName = "persistent://public/default/" + TOPIC;
+ try (PulsarClient pulsarClient =
+ PulsarClient.builder()
+ .serviceUrl(
+ String.format(
+ "pulsar://%s:%s",
+ PULSAR_CONTAINER.getHost(),
+
PULSAR_CONTAINER.getMappedPort(PULSAR_BROKER_PORT)))
+ .build();
+ Reader<byte[]> reader =
+ pulsarClient
+ .newReader(Schema.BYTES)
+ .topic(topicFullName)
+ .startMessageId(MessageId.earliest)
+ .create()) {
+ Message<byte[]> message = reader.readNext(5, TimeUnit.SECONDS);
+ Assertions.assertNotNull(message, "Canal has not yet forwarded
data to topic " + TOPIC);
+ LOG.info(
+ "Detected Canal data in topic {} with message id {}",
+ TOPIC,
+ message.getMessageId());
+ }
+ }
+
@BeforeAll
@Override
public void startUp() throws ClassNotFoundException, InterruptedException {
@@ -275,8 +305,13 @@ public class CanalToPulsarIT extends TestSuiteBase
implements TestResource {
.untilAsserted(this::waitForTopicCreated);
// before ddl, the pulsar_canal connector should be started
inventoryDatabase.createAndInitialize();
- // wait pulsar get data from canal server
- Thread.sleep(10 * 1000);
+ // wait for Canal server to forward data to Pulsar topic
+ given().ignoreExceptions()
+ .await()
+ .atLeast(100, TimeUnit.MILLISECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(5, TimeUnit.MINUTES)
+ .untilAsserted(this::waitForCanalDataInTopic);
LOG.info("The fourth stage: Starting PostgresSQL container...");
createPostgreSQLContainer();
Startables.deepStart(Stream.of(POSTGRESQL_CONTAINER)).join();
