You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "david-streamlio (via GitHub)" <gi...@apache.org> on 2026/07/09 19:25:46 UTC
[PR] [improve][test] Add Debezium SQL Server source integration test [pulsar-connectors]
david-streamlio opened a new pull request, #60:
URL: https://github.com/apache/pulsar-connectors/pull/60
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:test` — `DebeziumMsSqlSourceTest > testMsSqlCdcEvents PASSED`, BUILD SUCCESSFUL in 48s (with images already pulled)
- `./gradlew :debezium:pulsar-io-debezium-mssql:check -x test` — BUILD SUCCESSFUL
### Documentation
- [x] `doc-not-needed`
Test-only change; no user-facing behavior is modified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][test] Add Debezium SQL Server source integration test [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio commented on PR #60:
URL: https://github.com/apache/pulsar-connectors/pull/60#issuecomment-4929183787
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][test] Add Debezium SQL Server source integration test [pulsar-connectors]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #60:
URL: https://github.com/apache/pulsar-connectors/pull/60
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][test] Add Debezium SQL Server source integration test [pulsar-connectors]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60:
URL: https://github.com/apache/pulsar-connectors/pull/60#discussion_r3554286445
##########
debezium/mssql/src/test/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSourceTest.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.debezium.mssql;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+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.TimeUnit;
+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.MSSQLServerContainer;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.utility.DockerImageName;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class DebeziumMsSqlSourceTest {
+
+ 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()
Review Comment:
Using the `mcr.microsoft.com/mssql/server:2022-latest` tag makes the test non-deterministic (the image contents can change over time), which can cause CI flakiness. Consider making the SQL Server image configurable (and ideally pinning to a specific tag in CI) the same way `PULSAR_TEST_IMAGE` is handled.
##########
debezium/mssql/src/test/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSourceTest.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.debezium.mssql;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+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.TimeUnit;
+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.MSSQLServerContainer;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.utility.DockerImageName;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class DebeziumMsSqlSourceTest {
+
+ 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()
+ // The SQL Server Agent is required to run the CDC capture jobs
+ .withEnv("MSSQL_AGENT_ENABLED", "true");
+ mssqlContainer.start();
+
+ pulsarContainer = new PulsarContainer(DockerImageName.parse(PULSAR_IMAGE));
+ pulsarContainer.start();
+
+ pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarContainer.getPulsarBrokerUrl())
+ .build();
+
+ String jdbcUrl = mssqlContainer.getJdbcUrl() + ";encrypt=false";
+
+ // Create the test database, enable CDC on it and on the test table,
+ // then insert initial data (using sa which has all required privileges)
+ try (Connection conn = DriverManager.getConnection(
+ jdbcUrl, mssqlContainer.getUsername(), mssqlContainer.getPassword());
+ Statement stmt = conn.createStatement()) {
+ stmt.execute("CREATE DATABASE testdb");
+ stmt.execute("USE testdb");
+ stmt.execute("EXEC sys.sp_cdc_enable_db");
+ stmt.execute("CREATE TABLE products ("
+ + "id INT IDENTITY PRIMARY KEY, "
+ + "name VARCHAR(255) NOT NULL, "
+ + "description VARCHAR(512))");
+ stmt.execute("EXEC sys.sp_cdc_enable_table "
+ + "@source_schema='dbo', @source_name='products', @role_name=NULL");
+ stmt.execute("INSERT INTO products (name, description) VALUES ('widget', 'A small widget')");
+ stmt.execute("INSERT INTO products (name, description) VALUES ('gadget', 'A fancy gadget')");
+ }
+
+ source = new DebeziumMsSqlSource();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() throws Exception {
+ if (source != null) {
+ try {
+ source.close();
+ } catch (Exception e) {
+ log.warn("Failed to close source", e);
+ }
+ }
+ if (pulsarClient != null) {
+ pulsarClient.close();
+ }
+ if (pulsarContainer != null) {
+ pulsarContainer.stop();
+ }
+ if (mssqlContainer != null) {
+ mssqlContainer.stop();
+ }
+ }
+
+ @Test
+ public void testMsSqlCdcEvents() throws Exception {
+ String pulsarServiceUrl = pulsarContainer.getPulsarBrokerUrl();
+
+ SourceContext sourceContext = mock(SourceContext.class);
+ when(sourceContext.getPulsarClient()).thenReturn(pulsarClient);
+ when(sourceContext.getPulsarClientBuilder()).thenReturn(
+ PulsarClient.builder().serviceUrl(pulsarServiceUrl));
+ when(sourceContext.getTenant()).thenReturn("public");
+ when(sourceContext.getNamespace()).thenReturn("default");
+ when(sourceContext.getSourceName()).thenReturn("debezium-mssql-test");
+ when(sourceContext.getSecret(anyString())).thenReturn(null);
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("database.hostname", mssqlContainer.getHost());
+ config.put("database.port", String.valueOf(mssqlContainer.getMappedPort(MSSQL_PORT)));
+ // Use sa which has all required CDC privileges
+ config.put("database.user", mssqlContainer.getUsername());
+ config.put("database.password", mssqlContainer.getPassword());
+ config.put("database.names", "testdb");
+ // The SQL Server connector runs in multi-partition mode and requires a task id.
+ // DebeziumSource starts the connector task directly (without going through
+ // SqlServerConnector.taskConfigs which would normally assign it), so set it here.
+ config.put("task.id", "0");
+ config.put("database.encrypt", "false");
+ config.put("topic.prefix", "sqlserver1");
+ config.put("include.schema.changes", "false");
+ config.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
+
+ source.open(config, sourceContext);
+
+ // 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);
+ });
Review Comment:
`AbstractKafkaConnectSource.read()` blocks until a record is available and never returns null (it loops on `sourceTask.poll()`), so calling `source.read()` inside `Awaitility.untilAsserted` can cause the condition evaluation to block and bypass Awaitility's timeout/polling. Running the blocking reads on a background thread and having Awaitility wait on a counter keeps the 60s timeout effective and avoids potential hangs if the connector never produces records.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][test] Add Debezium SQL Server source integration test [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio commented on PR #60:
URL: https://github.com/apache/pulsar-connectors/pull/60#issuecomment-4928776619
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org