This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12987-76e11cb80e36b4844ba6820d43e0522dbd912d3c in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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());
