From 7072743dff917644bd1d440dcae43af0b65b62ec Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Thu, 9 Jul 2026 13:29:33 -0700 Subject: [PATCH 1/2] [fix][test] Bound record reads in Debezium MySQL and Postgres tests AbstractKafkaConnectSource.read() loops until a record is available and never returns null. Calling it on the test thread inside an Awaitility untilAsserted block means a connector that produces fewer records than expected blocks forever: Awaitility cannot interrupt a blocked assertion, so its atMost(60s) never fires. Neither test declared a timeOut, so the CI job ran to its own 45-minute limit and reported a cancellation rather than a failure. Read each record on a bounded worker thread and add a test-level timeOut as a backstop, so under-delivery fails promptly with a message naming the connector. --- .../mysql/DebeziumMysqlSourceTest.java | 68 +++++++++++++------ .../postgres/DebeziumPostgresSourceTest.java | 68 +++++++++++++------ 2 files changed, 94 insertions(+), 42 deletions(-) diff --git a/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java b/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java index b6ec3ed6dd..a6a73d710e 100644 --- a/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java +++ b/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java @@ -21,20 +21,23 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; -import static org.testng.Assert.assertTrue; import java.sql.Connection; import java.sql.DriverManager; import java.sql.Statement; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.io.core.SourceContext; -import org.awaitility.Awaitility; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.PulsarContainer; import org.testcontainers.containers.wait.strategy.Wait; @@ -49,15 +52,22 @@ public class DebeziumMysqlSourceTest { private static final String PULSAR_IMAGE = System.getenv().getOrDefault("PULSAR_TEST_IMAGE", "apachepulsar/pulsar:4.1.3"); + /** Rows seeded before open(), each of which the initial snapshot must emit. */ + private static final int EXPECTED_RECORDS = 2; + + private static final int READ_TIMEOUT_SECONDS = 120; + private static final int MYSQL_PORT = 3306; private GenericContainer mysqlContainer; private PulsarContainer pulsarContainer; private PulsarClient pulsarClient; private DebeziumMysqlSource source; + private ExecutorService readerExecutor; @BeforeMethod public void setup() throws Exception { + readerExecutor = Executors.newSingleThreadExecutor(); // Use GenericContainer instead of MySQLContainer to get full root access // which is required for Debezium's REPLICATION SLAVE/CLIENT privileges mysqlContainer = new GenericContainer<>(DockerImageName.parse("mysql:8.0")) @@ -101,6 +111,10 @@ public void setup() throws Exception { @AfterMethod(alwaysRun = true) public void cleanup() throws Exception { + if (readerExecutor != null) { + // shutdownNow: a reader may still be blocked in read(), which never returns null + readerExecutor.shutdownNow(); + } if (source != null) { try { source.close(); @@ -119,7 +133,7 @@ public void cleanup() throws Exception { } } - @Test + @Test(timeOut = 600_000) public void testMysqlCdcEvents() throws Exception { String pulsarServiceUrl = pulsarContainer.getPulsarBrokerUrl(); @@ -147,23 +161,35 @@ public void testMysqlCdcEvents() throws Exception { // Debezium performs an initial snapshot of existing data. // We should receive CDC records for the 2 rows we inserted. - Awaitility.await() - .atMost(60, TimeUnit.SECONDS) - .pollInterval(500, TimeUnit.MILLISECONDS) - .untilAsserted(() -> { - int recordCount = 0; - Record> record; - while ((record = source.read()) != null) { - assertNotNull(record.getValue()); - log.info("Received CDC record: key={}", record.getKey().orElse(null)); - recordCount++; - record.ack(); - if (recordCount >= 2) { - break; - } - } - assertTrue(recordCount >= 2, - "Expected at least 2 CDC records from initial snapshot, got " + recordCount); - }); + int received = 0; + for (int i = 0; i < EXPECTED_RECORDS; i++) { + Record> record = readOne(); + assertNotNull(record.getValue()); + log.info("Received CDC record: key={}", record.getKey().orElse(null)); + record.ack(); + received++; + } + assertEquals(received, EXPECTED_RECORDS); + } + + /** + * Reads a single record, failing if none arrives in time. + * + *

{@code AbstractKafkaConnectSource.read()} loops until a record is available and never + * returns null, so calling it on the test thread means an under-delivering connector hangs + * the test — and, since Awaitility cannot interrupt a blocked assertion, the whole CI job + * until its own timeout. Run it on a separate thread so a missing record surfaces as a + * prompt, diagnosable failure instead. + */ + private Record> readOne() throws Exception { + Future>> future = readerExecutor.submit(() -> source.read()); + try { + return future.get(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (TimeoutException e) { + future.cancel(true); + throw new AssertionError("Timed out after " + READ_TIMEOUT_SECONDS + + "s waiting for a CDC record from the initial snapshot. " + + "The connector produced no record; see the Debezium logs above.", e); + } } } diff --git a/debezium/postgres/src/test/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSourceTest.java b/debezium/postgres/src/test/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSourceTest.java index 836a7161ad..f550d8e9d6 100644 --- a/debezium/postgres/src/test/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSourceTest.java +++ b/debezium/postgres/src/test/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSourceTest.java @@ -21,20 +21,23 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; -import static org.testng.Assert.assertTrue; import java.sql.Connection; import java.sql.DriverManager; import java.sql.Statement; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.io.core.SourceContext; -import org.awaitility.Awaitility; import org.testcontainers.containers.PostgreSQLContainer; import org.testcontainers.containers.PulsarContainer; import org.testcontainers.utility.DockerImageName; @@ -48,13 +51,20 @@ public class DebeziumPostgresSourceTest { private static final String PULSAR_IMAGE = System.getenv().getOrDefault("PULSAR_TEST_IMAGE", "apachepulsar/pulsar:4.1.3"); + /** Rows seeded before open(), each of which the initial snapshot must emit. */ + private static final int EXPECTED_RECORDS = 2; + + private static final int READ_TIMEOUT_SECONDS = 120; + private PostgreSQLContainer postgresContainer; private PulsarContainer pulsarContainer; private PulsarClient pulsarClient; private DebeziumPostgresSource source; + private ExecutorService readerExecutor; @BeforeMethod public void setup() throws Exception { + readerExecutor = Executors.newSingleThreadExecutor(); postgresContainer = new PostgreSQLContainer<>(DockerImageName.parse("postgres:16")) .withDatabaseName("testdb") .withUsername("debezium") @@ -89,6 +99,10 @@ public void setup() throws Exception { @AfterMethod(alwaysRun = true) public void cleanup() throws Exception { + if (readerExecutor != null) { + // shutdownNow: a reader may still be blocked in read(), which never returns null + readerExecutor.shutdownNow(); + } if (source != null) { try { source.close(); @@ -107,7 +121,7 @@ public void cleanup() throws Exception { } } - @Test + @Test(timeOut = 600_000) public void testPostgresCdcEvents() throws Exception { String pulsarServiceUrl = pulsarContainer.getPulsarBrokerUrl(); @@ -134,23 +148,35 @@ public void testPostgresCdcEvents() throws Exception { // Debezium performs an initial snapshot of existing data. // We should receive CDC records for the 2 rows we inserted. - Awaitility.await() - .atMost(60, TimeUnit.SECONDS) - .pollInterval(500, TimeUnit.MILLISECONDS) - .untilAsserted(() -> { - int recordCount = 0; - Record> record; - while ((record = source.read()) != null) { - assertNotNull(record.getValue()); - log.info("Received CDC record: key={}", record.getKey().orElse(null)); - recordCount++; - record.ack(); - if (recordCount >= 2) { - break; - } - } - assertTrue(recordCount >= 2, - "Expected at least 2 CDC records from initial snapshot, got " + recordCount); - }); + int received = 0; + for (int i = 0; i < EXPECTED_RECORDS; i++) { + Record> record = readOne(); + assertNotNull(record.getValue()); + log.info("Received CDC record: key={}", record.getKey().orElse(null)); + record.ack(); + received++; + } + assertEquals(received, EXPECTED_RECORDS); + } + + /** + * Reads a single record, failing if none arrives in time. + * + *

{@code AbstractKafkaConnectSource.read()} loops until a record is available and never + * returns null, so calling it on the test thread means an under-delivering connector hangs + * the test — and, since Awaitility cannot interrupt a blocked assertion, the whole CI job + * until its own timeout. Run it on a separate thread so a missing record surfaces as a + * prompt, diagnosable failure instead. + */ + private Record> readOne() throws Exception { + Future>> future = readerExecutor.submit(() -> source.read()); + try { + return future.get(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (TimeoutException e) { + future.cancel(true); + throw new AssertionError("Timed out after " + READ_TIMEOUT_SECONDS + + "s waiting for a CDC record from the initial snapshot. " + + "The connector produced no record; see the Debezium logs above.", e); + } } } From 628d608e6a1e6bb04761723aa6dacb7dc965a335 Mon Sep 17 00:00:00 2001 From: David Kjerrumgaard <35466513+david-streamlio@users.noreply.github.com> Date: Thu, 9 Jul 2026 14:30:11 -0700 Subject: [PATCH 2/2] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java b/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java index a6a73d710e..fa8f4f9af2 100644 --- a/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java +++ b/debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java @@ -67,7 +67,11 @@ public class DebeziumMysqlSourceTest { @BeforeMethod public void setup() throws Exception { - readerExecutor = Executors.newSingleThreadExecutor(); +readerExecutor = Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "debezium-mysql-test-reader"); + t.setDaemon(true); + return t; +}); // Use GenericContainer instead of MySQLContainer to get full root access // which is required for Debezium's REPLICATION SLAVE/CLIENT privileges mysqlContainer = new GenericContainer<>(DockerImageName.parse("mysql:8.0"))