[fix][test] Bound record reads in Debezium MySQL and Postgres tests

[david-streamlio]

Fixes #70

Motivation

DebeziumMysqlSourceTest and DebeziumPostgresSourceTest can hang CI for 45 minutes instead of failing, if their connector delivers fewer CDC records than expected.

AbstractKafkaConnectSource.read() is a while (true) loop whose only return yields a record — an empty poll hits continue. It never returns null. Both tests drive it with while ((record = source.read()) != null), escaping only via break after 2 records, all inside an Awaitility untilAsserted block. If fewer than 2 records arrive, read() spins forever, and Awaitility's atMost(60s) never fires because untilAsserted cannot interrupt a blocked assertion. Neither test declares a timeOut, so nothing bounds them.

They pass today only because MySQL and Postgres reliably emit their snapshot records. This was observed for real on #58, where the Tests - Connectors job ran 45m15s and was cancelled — no diagnostics, 45 minutes of runner time gone.

Modifications

For both tests:

• Read each record on a single-threaded executor with a per-record deadline (READ_TIMEOUT_SECONDS = 120), cancelling the blocked read and failing with a message that names the connector and points at the Debezium logs.
• Add @Test(timeOut = 600_000) as a backstop.
• shutdownNow() the executor in cleanup, since a reader may still be blocked in read().
• Replace the while (read() != null) idiom with an explicit EXPECTED_RECORDS loop, so the expectation is stated rather than implied by a break.

This is the same harness applied in #58.

Verifying this change

Both tests pass locally against real containers:

DebeziumMysqlSourceTest > testMysqlCdcEvents PASSED
DebeziumPostgresSourceTest > testPostgresCdcEvents PASSED
BUILD SUCCESSFUL in 26s

The guard itself was verified by forcing under-delivery (expecting 3 records against 2 seeded rows): the fixed harness fails in ~37s with a TimeoutException, where the old one hangs indefinitely.

Related

• [improve][test] Add Debezium MongoDB source integration test #58 (MongoDB) carries the same fix
• [improve][test] Add Debezium SQL Server source integration test #60 (SQL Server) still has the unbounded pattern and no timeOut
• [improve][test] Add Debezium Oracle source integration test #62 (Oracle) and [feat][io] Add Debezium MariaDB source connector #69 (MariaDB) have a timeOut backstop but still read on the test thread

Once those land, a follow-up can lift readOne() into a shared test base class rather than repeating it per module.

[Copilot]

Pull request overview

This PR hardens the Debezium MySQL and Postgres integration tests so they fail promptly (instead of hanging the entire CI job) when the connector under-delivers CDC records during the initial snapshot.

Changes:

• Replace the unbounded while (source.read() != null) pattern with a fixed EXPECTED_RECORDS loop.
• Add a per-record read deadline by running source.read() on a separate executor thread and failing on timeout.
• Add a TestNG @Test(timeOut = 600_000) backstop and ensure the executor is shut down during cleanup.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java	Adds bounded, per-record reads + test timeout to prevent indefinite CI hangs on under-delivery.
debezium/postgres/src/test/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSourceTest.java	Applies the same bounded read harness + test timeout to Postgres CDC snapshot assertions.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.