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 16bb0fdc [improve][test] Add Debezium MongoDB source integration test
(#58)
16bb0fdc is described below
commit 16bb0fdc332086e5c9437a6cf3c0c3f3fd88f6bb
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Thu Jul 9 14:29:28 2026 -0700
[improve][test] Add Debezium MongoDB source integration test (#58)
* [improve][test] Add Debezium MongoDB source integration test
The debezium/mongodb module had no tests. Add a Testcontainers-based
integration test following the existing DebeziumMysqlSourceTest
pattern: a single-node MongoDB replica set container plus a Pulsar
container for Debezium offset storage, seeded with two documents, and
an assertion that the initial snapshot produces CDC records.
* [improve][test] Bound record reads so an under-delivering connector
cannot hang CI
AbstractKafkaConnectSource.read() loops until a record is available and
never returns null. Calling it on the test thread inside an Awaitility
untilAsserted block meant a connector that produced fewer records than
expected blocked forever: Awaitility cannot interrupt a blocked
assertion, so its atMost(60s) never fired and the CI job ran until its
own 45-minute timeout.
Read each record on a separate thread with a per-record deadline, and
add a test-level timeOut as a backstop. Under-delivery now fails in
seconds with a message pointing at the connector, instead of hanging.
---
debezium/mongodb/build.gradle.kts | 5 +
.../mongodb/DebeziumMongoDbSourceTest.java | 168 +++++++++++++++++++++
2 files changed, 173 insertions(+)
diff --git a/debezium/mongodb/build.gradle.kts
b/debezium/mongodb/build.gradle.kts
index d4292a98..975fdb60 100644
--- a/debezium/mongodb/build.gradle.kts
+++ b/debezium/mongodb/build.gradle.kts
@@ -25,4 +25,9 @@ dependencies {
implementation(libs.pulsar.io.core)
implementation(project(":debezium:pulsar-io-debezium-core"))
implementation(libs.debezium.connector.mongodb)
+
+ testImplementation(libs.testcontainers.mongodb)
+ testImplementation(libs.testcontainers.pulsar)
+ testImplementation(libs.pulsar.client)
+ testImplementation(libs.mongodb.driver.sync)
}
diff --git
a/debezium/mongodb/src/test/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSourceTest.java
b/debezium/mongodb/src/test/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSourceTest.java
new file mode 100644
index 00000000..fb6e0b6e
--- /dev/null
+++
b/debezium/mongodb/src/test/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSourceTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.mongodb;
+
+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 com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoCollection;
+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.bson.Document;
+import org.testcontainers.containers.MongoDBContainer;
+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 DebeziumMongoDbSourceTest {
+
+ private static final String PULSAR_IMAGE =
+ System.getenv().getOrDefault("PULSAR_TEST_IMAGE",
"apachepulsar/pulsar:4.1.3");
+
+ /** Documents 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 MongoDBContainer mongoContainer;
+ private PulsarContainer pulsarContainer;
+ private PulsarClient pulsarClient;
+ private DebeziumMongoDbSource source;
+ private ExecutorService readerExecutor;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ readerExecutor = Executors.newSingleThreadExecutor();
+ // MongoDBContainer starts a single-node replica set, which Debezium
requires
+ // for change streams.
+ mongoContainer = new
MongoDBContainer(DockerImageName.parse("mongo:7.0"));
+ mongoContainer.start();
+
+ pulsarContainer = new
PulsarContainer(DockerImageName.parse(PULSAR_IMAGE));
+ pulsarContainer.start();
+
+ pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarContainer.getPulsarBrokerUrl())
+ .build();
+
+ // Create test collection and insert initial data
+ try (MongoClient client =
MongoClients.create(mongoContainer.getConnectionString())) {
+ MongoCollection<Document> products =
client.getDatabase("testdb").getCollection("products");
+ products.insertOne(new Document("name",
"widget").append("description", "A small widget"));
+ products.insertOne(new Document("name",
"gadget").append("description", "A fancy gadget"));
+ }
+
+ source = new DebeziumMongoDbSource();
+ }
+
+ @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();
+ } catch (Exception e) {
+ log.warn("Failed to close source", e);
+ }
+ }
+ if (pulsarClient != null) {
+ pulsarClient.close();
+ }
+ if (pulsarContainer != null) {
+ pulsarContainer.stop();
+ }
+ if (mongoContainer != null) {
+ mongoContainer.stop();
+ }
+ }
+
+ @Test(timeOut = 600_000)
+ public void testMongoDbCdcEvents() 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-mongodb-test");
+ when(sourceContext.getSecret(anyString())).thenReturn(null);
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("mongodb.connection.string",
mongoContainer.getConnectionString());
+ config.put("topic.prefix", "mongodb1");
+ config.put("collection.include.list", "testdb.products");
+
+ source.open(config, sourceContext);
+
+ // Debezium performs an initial snapshot of existing data.
+ // We should receive CDC records for the 2 documents we inserted.
+ 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);
+ }
+ }
+}
