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 fc8e2204b5 [Test][E2E] Bound Pulsar sink IT receive wait (#11159)
fc8e2204b5 is described below
commit fc8e2204b5e53548be889b0b4b2b45d93af3721f
Author: Daniel <[email protected]>
AuthorDate: Thu Jul 2 23:10:56 2026 +0800
[Test][E2E] Bound Pulsar sink IT receive wait (#11159)
Co-authored-by: DanielLeens <[email protected]>
---
.../e2e/connector/pulsar/PulsarSinkIT.java | 117 +++++++++++++--------
1 file changed, 75 insertions(+), 42 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarSinkIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarSinkIT.java
index 9a9201f341..49a3550025 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarSinkIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarSinkIT.java
@@ -39,7 +39,7 @@ import org.testcontainers.shaded.org.awaitility.Awaitility;
import lombok.extern.slf4j.Slf4j;
-import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
@@ -53,7 +53,17 @@ public class PulsarSinkIT extends TestSuiteBase implements
TestResource {
private static final String PULSAR_IMAGE_NAME =
"apachepulsar/pulsar:2.3.1";
public static final String PULSAR_HOST = "pulsar.e2e.sink";
- public static final String TOPIC = "topic_test02";
+ private static final String TOPIC = "topic_test02";
+
+ /** Expected record count produced by fake_to_pulsar.conf. */
+ private static final int EXPECTED_RECORD_COUNT = 10;
+
+ /** Total time budget for draining sink output before failing the test. */
+ private static final Duration RECEIVE_TIMEOUT = Duration.ofSeconds(60);
+
+ /** Single receive poll timeout to avoid blocking the CI job forever. */
+ private static final int RECEIVE_POLL_TIMEOUT_SECONDS = 5;
+
private PulsarContainer pulsarContainer;
@Override
@@ -83,51 +93,74 @@ public class PulsarSinkIT extends TestSuiteBase implements
TestResource {
pulsarContainer.close();
}
- private List<String> getPulsarConsumerData() {
- List<String> data = new ArrayList<>();
- try {
- PulsarClient client =
-
PulsarClient.builder().serviceUrl(pulsarContainer.getPulsarBrokerUrl()).build();
-
- Random random = new Random();
- Consumer consumer =
- client.newConsumer()
- .topic(TOPIC)
- .subscriptionName("PulsarSubTest" +
random.nextInt())
- .subscriptionType(SubscriptionType.Exclusive)
-
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
- .subscribe();
- for (int i = 0; i < 10; i++) {
- Message msg = consumer.receive(30, TimeUnit.SECONDS);
- if (msg != null) {
- data.add(new String(msg.getData()));
- consumer.acknowledge(msg.getMessageId());
- log.info("value:{}", new String(msg.getData()));
- } else {
- log.warn("No message received within timeout, received {}
so far", data.size());
- break;
- }
+ /** Create a dedicated client for the broker exposed by the test
container. */
+ private PulsarClient createPulsarClient() throws Exception {
+ return
PulsarClient.builder().serviceUrl(pulsarContainer.getPulsarBrokerUrl()).build();
+ }
+
+ /**
+ * Subscribe to the sink topic before the batch job starts so the
assertion cannot miss records
+ * because the producer finished before the consumer began polling.
+ */
+ private Consumer<byte[]> createPulsarConsumer(PulsarClient client) throws
Exception {
+ return client.newConsumer()
+ .topic(TOPIC)
+ .subscriptionName("PulsarSubTest" + new Random().nextInt())
+ .subscriptionType(SubscriptionType.Exclusive)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ }
+
+ /**
+ * Consume the records written by the sink job without leaving the test
stuck in a blocking
+ * receive call when CI delivery is delayed.
+ */
+ private List<String> getPulsarConsumerData(Consumer<byte[]> consumer)
throws Exception {
+ List<String> data = new ArrayList<>(EXPECTED_RECORD_COUNT);
+ long deadlineNanos = System.nanoTime() + RECEIVE_TIMEOUT.toNanos();
+ while (data.size() < EXPECTED_RECORD_COUNT && System.nanoTime() <
deadlineNanos) {
+ Message<byte[]> msg =
consumer.receive(RECEIVE_POLL_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ if (msg == null) {
+ continue;
}
- } catch (Exception e) {
- throw new RuntimeException("Failed to get pulsar consumer data",
e);
+
+ String value = new String(msg.getData(), StandardCharsets.UTF_8);
+ data.add(value);
+ consumer.acknowledge(msg);
+ log.info("value:{}", value);
+ }
+
+ if (data.size() < EXPECTED_RECORD_COUNT) {
+ log.warn(
+ "Timed out waiting for {} Pulsar sink records, only
consumed {}",
+ EXPECTED_RECORD_COUNT,
+ data.size());
}
return data;
}
@TestTemplate
- public void testSinkPulsar(TestContainer container) throws IOException,
InterruptedException {
- Container.ExecResult execResult =
container.executeJob("/fake_to_pulsar.conf");
- Assertions.assertEquals(execResult.getExitCode(), 0);
-
- List<String> data = getPulsarConsumerData();
- log.info("data size:{}", data.size());
- ObjectMapper objectMapper = new ObjectMapper();
- ObjectNode objectNode = objectMapper.readValue(data.get(0),
ObjectNode.class);
- Assertions.assertTrue(objectNode.has("c_map"));
- Assertions.assertTrue(objectNode.has("c_array"));
- Assertions.assertTrue(objectNode.has("c_string"));
- Assertions.assertTrue(objectNode.has("c_boolean"));
- Assertions.assertTrue(objectNode.has("c_double"));
- Assertions.assertEquals(10, data.size());
+ public void testSinkPulsar(TestContainer container) throws Exception {
+ try (PulsarClient client = createPulsarClient();
+ Consumer<byte[]> consumer = createPulsarConsumer(client)) {
+ Container.ExecResult execResult =
container.executeJob("/fake_to_pulsar.conf");
+ Assertions.assertEquals(execResult.getExitCode(), 0);
+
+ List<String> data = getPulsarConsumerData(consumer);
+ log.info("data size:{}", data.size());
+ Assertions.assertEquals(
+ EXPECTED_RECORD_COUNT,
+ data.size(),
+ String.format(
+ "Expected %d Pulsar records within %d seconds but
received %d",
+ EXPECTED_RECORD_COUNT,
RECEIVE_TIMEOUT.getSeconds(), data.size()));
+ ObjectMapper objectMapper = new ObjectMapper();
+ ObjectNode objectNode = objectMapper.readValue(data.get(0),
ObjectNode.class);
+ Assertions.assertTrue(objectNode.has("c_map"));
+ Assertions.assertTrue(objectNode.has("c_array"));
+ Assertions.assertTrue(objectNode.has("c_string"));
+ Assertions.assertTrue(objectNode.has("c_boolean"));
+ Assertions.assertTrue(objectNode.has("c_double"));
+ }
}
}
