This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new e0f878ce [fix][test] Bound record reads in Debezium MySQL and Postgres
tests (#71)
e0f878ce is described below
commit e0f878ce84e4b89d116bd4f58182f72bd7c3f595
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Thu Jul 9 14:44:18 2026 -0700
[fix][test] Bound record reads in Debezium MySQL and Postgres tests (#71)
* [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.
* Potential fix for pull request finding
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../io/debezium/mysql/DebeziumMysqlSourceTest.java | 72 +++++++++++++++-------
.../postgres/DebeziumPostgresSourceTest.java | 68 +++++++++++++-------
2 files changed, 98 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 b6ec3ed6..fa8f4f9a 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 @@ package org.apache.pulsar.io.debezium.mysql;
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,26 @@ 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(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"))
@@ -101,6 +115,10 @@ public class DebeziumMysqlSourceTest {
@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 +137,7 @@ public class DebeziumMysqlSourceTest {
}
}
- @Test
+ @Test(timeOut = 600_000)
public void testMysqlCdcEvents() throws Exception {
String pulsarServiceUrl = pulsarContainer.getPulsarBrokerUrl();
@@ -147,23 +165,35 @@ public class DebeziumMysqlSourceTest {
// 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<KeyValue<byte[], byte[]>> 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<KeyValue<byte[], byte[]>> 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.
+ *
+ * <p>{@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<KeyValue<byte[], byte[]>> readOne() throws Exception {
+ Future<Record<KeyValue<byte[], byte[]>>> 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 836a7161..f550d8e9 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 @@ package org.apache.pulsar.io.debezium.postgres;
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 class DebeziumPostgresSourceTest {
@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 class DebeziumPostgresSourceTest {
}
}
- @Test
+ @Test(timeOut = 600_000)
public void testPostgresCdcEvents() throws Exception {
String pulsarServiceUrl = pulsarContainer.getPulsarBrokerUrl();
@@ -134,23 +148,35 @@ public class DebeziumPostgresSourceTest {
// 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<KeyValue<byte[], byte[]>> 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<KeyValue<byte[], byte[]>> 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.
+ *
+ * <p>{@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<KeyValue<byte[], byte[]>> readOne() throws Exception {
+ Future<Record<KeyValue<byte[], byte[]>>> 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);
+ }
}
}
