Skip to content

[improve][test] Add Debezium SQL Server source integration test#60

Open
david-streamlio wants to merge 1 commit into
apache:masterfrom
david-streamlio:test/debezium-mssql-integration
Open

[improve][test] Add Debezium SQL Server source integration test#60
david-streamlio wants to merge 1 commit into
apache:masterfrom
david-streamlio:test/debezium-mssql-integration

Conversation

@david-streamlio

Copy link
Copy Markdown
Contributor

Fixes #43

Motivation

The debezium/mssql module has zero tests — DebeziumMsSqlSource is not exercised at all in CI. This adds a Testcontainers-based integration test following the existing pattern in debezium/mysql (DebeziumMysqlSourceTest).

Modifications

  • gradle/libs.versions.toml: add a testcontainers-mssqlserver catalog entry (no version needed — it resolves via the Testcontainers BOM imported by the dependencies platform).
  • debezium/mssql/build.gradle.kts: add testImplementation dependencies (testcontainers-mssqlserver, testcontainers-pulsar, pulsar-client).
  • debezium/mssql/src/test/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSourceTest.java: new integration test that
    • starts an MSSQLServerContainer (mcr.microsoft.com/mssql/server:2022-latest) with MSSQL_AGENT_ENABLED=true — the SQL Server Agent is required to run the CDC capture jobs,
    • creates testdb, enables CDC on the database (sys.sp_cdc_enable_db) and on a products table (sys.sp_cdc_enable_table), and inserts two rows over plain JDBC as sa,
    • starts a PulsarContainer for the Debezium schema history topic,
    • opens DebeziumMsSqlSource and asserts (via Awaitility) that the initial snapshot yields at least 2 CDC records from source.read().

One SQL Server-specific detail: the test sets task.id=0 explicitly. The SQL Server connector runs in multi-partition mode and its metrics require a task id, but DebeziumSource starts the connector task directly (it does not set the adapter's kafkaConnectorSourceClass config), so SqlServerConnector.taskConfigs() — which would normally assign the task id — never runs. Without it, source.open() fails with an NPE in Debezium's JMX metric-name construction.

Verifying this change

Ran locally with Docker (macOS, Docker Desktop):

  • ./gradlew :debezium:pulsar-io-debezium-mssql:compileTestJava — BUILD SUCCESSFUL
  • ./gradlew :debezium:pulsar-io-debezium-mssql:testDebeziumMsSqlSourceTest > testMsSqlCdcEvents PASSED, BUILD SUCCESSFUL in 48s (with images already pulled)
  • ./gradlew :debezium:pulsar-io-debezium-mssql:check -x test — BUILD SUCCESSFUL

Documentation

  • doc-not-needed

Test-only change; no user-facing behavior is modified.

Fixes apache#43

The debezium/mssql module had no tests. Add a Testcontainers-based
integration test following the existing DebeziumMysqlSourceTest pattern:
start an MSSQLServerContainer (SQL Server 2022) with the SQL Server
Agent enabled (required for CDC capture jobs), enable CDC on the test
database and table, start a PulsarContainer for schema history, and
assert the initial snapshot produces CDC records via source.read().

The SQL Server connector runs in multi-partition mode and requires a
task.id; since DebeziumSource starts the connector task directly
(bypassing SqlServerConnector.taskConfigs which would normally assign
it), the test sets task.id explicitly.

Adds a testcontainers-mssqlserver entry to the version catalog
(version resolved via the existing Testcontainers BOM).
@david-streamlio

Copy link
Copy Markdown
Contributor Author

Note for reviewers: the explicit task.id=0 in this test is a workaround for a real product bug, not a test artifact — filed as #61. AbstractKafkaConnectSource only calls connector.taskConfigs() when the adaptor's kafkaConnectorSourceClass key is present, but Debezium sources set Debezium's own connector.class/task.class keys instead, so taskConfigs() never runs and the multi-partition SQL Server connector never gets its task.id. Once #61 is fixed, the task.id line here should be removed and this test will guard the fix.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds CI coverage for the Debezium SQL Server source connector by introducing a Testcontainers-based integration test, aligning debezium/mssql with existing Debezium connector modules that already have similar coverage.

Changes:

  • Added a new MSSQL CDC integration test for DebeziumMsSqlSource using Testcontainers (SQL Server + Pulsar) and Awaitility-style assertions.
  • Added testcontainers-mssqlserver to the version catalog and wired new test dependencies into the debezium/mssql module.
  • Ensured connector startup succeeds in the test by explicitly setting task.id.

Reviewed changes

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

File Description
gradle/libs.versions.toml Adds a version-catalog alias for org.testcontainers:mssqlserver to support MSSQL Testcontainers usage.
debezium/mssql/build.gradle.kts Adds the necessary testImplementation dependencies for the new MSSQL + Pulsar integration test.
debezium/mssql/src/test/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSourceTest.java New integration test that provisions SQL Server CDC and asserts snapshot/CDC records are emitted by the source.

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

Comment on lines +48 to +62
private static final String PULSAR_IMAGE =
System.getenv().getOrDefault("PULSAR_TEST_IMAGE", "apachepulsar/pulsar:4.1.3");

private static final int MSSQL_PORT = 1433;

private MSSQLServerContainer<?> mssqlContainer;
private PulsarContainer pulsarContainer;
private PulsarClient pulsarClient;
private DebeziumMsSqlSource source;

@BeforeMethod
public void setup() throws Exception {
mssqlContainer = new MSSQLServerContainer<>(
DockerImageName.parse("mcr.microsoft.com/mssql/server:2022-latest"))
.acceptLicense()
Comment on lines +150 to +167
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);
});
@david-streamlio

Copy link
Copy Markdown
Contributor Author

@lhotari flagging this for your review because of the task.id=0 line in the test config, which is a workaround for a real product bug rather than a test artifact — filed as #61.

AbstractKafkaConnectSource.open() only calls connector.taskConfigs() when the adaptor's own kafkaConnectorSourceClass key is present. Debezium sources set Debezium's connector.class/task.class keys instead, so they all take the backward-compat branch and taskConfigs() never runs. Harmless for single-partition connectors (MySQL, Postgres — which is why existing tests pass), but SQL Server is multi-partition and needs the task.id that taskConfigs() would assign; without it a null reaches Debezium's JMX metrics layer and open() throws an NPE.

Two things worth your judgement: whether the right fix in #61 is to have DebeziumSource populate kafkaConnectorSourceClass, or to invoke taskConfigs() in the backward-compat branch when a connector class is resolvable — and whether this test should land now with the workaround (the task.id line then gets deleted as part of #61, and this test becomes its regression guard) or wait for the fix.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[improve][test] Add integration tests for the Debezium SQL Server source connector

2 participants