This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 c875a87bd6 [Test][Connector-V2] Synchronize Python source close test
(#12987)
c875a87bd6 is described below
commit c875a87bd6fb238585dd6856c2c4233b6a8ece46
Author: zhiweiniu <[email protected]>
AuthorDate: Tue Oct 27 04:15:23 2026 +0000
[Test][Connector-V2] Synchronize Python source close test (#12987)
---
.../seatunnel/python/source/PythonSourceTest.java | 50 ++++++++++++++--------
.../seatunnel/e2e/connector/python/PythonIT.java | 7 +--
2 files changed, 37 insertions(+), 20 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-python/src/test/java/org/apache/seatunnel/connectors/seatunnel/python/source/PythonSourceTest.java
b/seatunnel-connectors-v2/connector-python/src/test/java/org/apache/seatunnel/connectors/seatunnel/python/source/PythonSourceTest.java
index 455d3410e2..346dc175ae 100644
---
a/seatunnel-connectors-v2/connector-python/src/test/java/org/apache/seatunnel/connectors/seatunnel/python/source/PythonSourceTest.java
+++
b/seatunnel-connectors-v2/connector-python/src/test/java/org/apache/seatunnel/connectors/seatunnel/python/source/PythonSourceTest.java
@@ -465,7 +465,6 @@ class PythonSourceTest {
BlockingCollector collector = new BlockingCollector();
AtomicReference<Throwable> pollFailure = new AtomicReference<>();
AtomicReference<Throwable> closeFailure = new AtomicReference<>();
- CountDownLatch closeStarted = new CountDownLatch(1);
Thread pollThread =
new Thread(
@@ -477,13 +476,9 @@ class PythonSourceTest {
}
},
"python-source-buffered-poll-test");
- pollThread.start();
- Assertions.assertTrue(collector.collectEntered.await(5,
TimeUnit.SECONDS));
-
Thread closeThread =
new Thread(
() -> {
- closeStarted.countDown();
try {
reader.close();
} catch (Throwable e) {
@@ -491,18 +486,39 @@ class PythonSourceTest {
}
},
"python-source-buffered-close-test");
- closeThread.start();
- Assertions.assertTrue(closeStarted.await(5, TimeUnit.SECONDS));
- Thread.sleep(50L);
- collector.releaseCollect.countDown();
-
- pollThread.join(TimeUnit.SECONDS.toMillis(5));
- closeThread.join(TimeUnit.SECONDS.toMillis(5));
- Assertions.assertFalse(pollThread.isAlive());
- Assertions.assertFalse(closeThread.isAlive());
- Assertions.assertNull(pollFailure.get());
- Assertions.assertNull(closeFailure.get());
- Assertions.assertEquals(1, collector.rows.size());
+ pollThread.start();
+ try {
+ Assertions.assertTrue(collector.collectEntered.await(5,
TimeUnit.SECONDS));
+ closeThread.start();
+ // close() publishes its cancellation flag before waiting for this
blocked poll.
+ // Starting the thread alone does not guarantee it has reached
that point.
+ long deadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(5);
+ while (closeThread.getState() != Thread.State.WAITING
+ && closeThread.isAlive()
+ && System.nanoTime() < deadlineNanos) {
+ Thread.sleep(10L);
+ }
+ Assertions.assertEquals(
+ Thread.State.WAITING,
+ closeThread.getState(),
+ "close must be waiting for the blocked collector before it
is released");
+ collector.releaseCollect.countDown();
+
+ pollThread.join(TimeUnit.SECONDS.toMillis(5));
+ closeThread.join(TimeUnit.SECONDS.toMillis(5));
+ Assertions.assertFalse(pollThread.isAlive());
+ Assertions.assertFalse(closeThread.isAlive());
+ Assertions.assertNull(pollFailure.get());
+ Assertions.assertNull(closeFailure.get());
+ Assertions.assertEquals(1, collector.rows.size());
+ } finally {
+ collector.releaseCollect.countDown();
+ pollThread.interrupt();
+ closeThread.interrupt();
+ pollThread.join(TimeUnit.SECONDS.toMillis(5));
+ closeThread.join(TimeUnit.SECONDS.toMillis(5));
+ reader.close();
+ }
}
@Test
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-python-e2e/src/test/java/org/apache/seatunnel/e2e/connector/python/PythonIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-python-e2e/src/test/java/org/apache/seatunnel/e2e/connector/python/PythonIT.java
index 6b2d8892a1..99c673d5af 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-python-e2e/src/test/java/org/apache/seatunnel/e2e/connector/python/PythonIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-python-e2e/src/test/java/org/apache/seatunnel/e2e/connector/python/PythonIT.java
@@ -22,7 +22,7 @@ import
org.apache.seatunnel.e2e.common.container.seatunnel.SeaTunnelContainer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.TestInstance;
import org.testcontainers.containers.Container;
@@ -68,9 +68,10 @@ public class PythonIT extends SeaTunnelContainer {
}
/**
- * Verifies the SeaTunnel Zeta engine can read rows from the documented
Python script contract.
+ * Verifies exact output on successive jobs in the same engine after the
previous Python reader
+ * has closed.
*/
- @Test
+ @RepeatedTest(2)
public void testPythonSourceToAssert() throws IOException,
InterruptedException {
Container.ExecResult result = executeJob("/python_to_assert.conf");
Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
