From 27e39e5ca25b50c40a9d84e293d4d3d402efe498 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 17 Jul 2026 18:24:09 -0400 Subject: [PATCH 1/3] [improve][test] Add integration tests for the MongoDB source connector Add MongoSourceContainerTest, the first integration test for the MongoDB source. It starts a MongoDBContainer (mongo:6.0, which auto-initializes the single-node replica set that change streams require), seeds documents, and runs MongoSource end-to-end, asserting that inserts flow through a real change stream into the source's queue with the expected operation, namespace, fullDocument and key. Mirrors the sibling MongoSinkContainerTest container harness and the DynamoDBSourceIntegrationTest blocking-read pattern: MongoSource is a PushSource whose read() blocks forever on an empty queue, so reads run on a bounded daemon executor with a timeout. syncType=FULL_SYNC replays the pre-seeded documents from the start of the stream, making delivery deterministic rather than racing the subscription becoming active; a background writer keeps a steady supply as a safety net. Fixes #121 --- .../io/mongodb/MongoSourceContainerTest.java | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java diff --git a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java new file mode 100644 index 0000000000..a63e7563cb --- /dev/null +++ b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java @@ -0,0 +1,201 @@ +/* + * 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.mongodb; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +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 java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import lombok.extern.slf4j.Slf4j; +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.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * Integration test for {@link MongoSource} that exercises the full path from a real MongoDB change + * stream into the source's record queue. A {@link MongoDBContainer} provides a single-node replica + * set (which change streams require) and the source is driven end-to-end against it. + */ +@Slf4j +public class MongoSourceContainerTest { + + private static final String DB = "testdb"; + private static final String COLLECTION = "messages"; + + // Change streams surface within seconds, but give each read a generous deadline so a + // non-delivering source fails promptly rather than hanging the suite. + private static final int READ_TIMEOUT_SECONDS = 60; + private static final int EXPECTED_RECORDS = 3; + + private MongoDBContainer mongoContainer; + private com.mongodb.client.MongoClient verifyClient; + private MongoSource source; + private ExecutorService readerExecutor; + private Thread writerThread; + private final AtomicBoolean keepWriting = new AtomicBoolean(false); + + @BeforeMethod + public void setUp() { + mongoContainer = new MongoDBContainer("mongo:6.0") + .withStartupTimeout(Duration.ofMinutes(3)); + mongoContainer.start(); + + // Sync driver used only to write test documents and drive the change stream. + verifyClient = com.mongodb.client.MongoClients.create(mongoContainer.getConnectionString()); + + readerExecutor = Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "mongo-it-reader"); + t.setDaemon(true); + return t; + }); + + source = new MongoSource(); + } + + @AfterMethod(alwaysRun = true) + public void tearDown() throws Exception { + keepWriting.set(false); + if (writerThread != null) { + writerThread.interrupt(); + } + if (readerExecutor != null) { + // 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 (verifyClient != null) { + verifyClient.close(); + } + if (mongoContainer != null) { + mongoContainer.stop(); + } + } + + @Test(timeOut = 300_000) + public void testReadFromChangeStream() throws Exception { + // Seed some documents before the source starts; FULL_SYNC replays them from the start. + for (int i = 0; i < EXPECTED_RECORDS; i++) { + insertDoc("seed-" + i); + } + + // Keep writing after open as well so the change stream has a steady supply regardless of + // exactly when the subscription becomes active. + startBackgroundWriter(); + + source.open(buildConfig(), mock(SourceContext.class)); + + int received = 0; + for (int i = 0; i < EXPECTED_RECORDS; i++) { + Record record = readOne(); + assertNotNull(record, "read() returned null"); + assertTrue(record.getKey().isPresent(), "record had no key (document key)"); + + Document value = Document.parse(new String(record.getValue(), StandardCharsets.UTF_8)); + assertEquals(value.getString("operation"), "INSERT", + "unexpected operation; got " + value.getString("operation")); + assertNotNull(value.get("ns"), "record missing ns"); + Document fullDocument = value.get("fullDocument", Document.class); + assertNotNull(fullDocument, "record missing fullDocument"); + assertTrue(fullDocument.getString("name").startsWith("seed-") + || fullDocument.getString("name").startsWith("live-"), + "unexpected fullDocument name: " + fullDocument.getString("name")); + log.info("Received change-stream record: key={} name={}", + record.getKey().orElse(null), fullDocument.getString("name")); + received++; + } + assertTrue(received >= EXPECTED_RECORDS, + "Expected at least " + EXPECTED_RECORDS + " records, got " + received); + } + + /** + * Reads a single record on a bounded worker thread, failing (rather than hanging) if none + * arrives in time. {@link MongoSource} is a {@code PushSource}; its {@code read()} blocks + * forever on an empty queue, so it must never be called on the test thread. + */ + private Record readOne() throws Exception { + Future> 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 record from the MongoDB change stream. The source produced " + + "no record; see the logs above.", e); + } + } + + private void startBackgroundWriter() { + keepWriting.set(true); + final AtomicInteger counter = new AtomicInteger(); + writerThread = new Thread(() -> { + while (keepWriting.get()) { + try { + insertDoc("live-" + counter.incrementAndGet()); + Thread.sleep(500); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (Exception e) { + log.warn("Background write failed", e); + } + } + }, "mongo-it-writer"); + writerThread.setDaemon(true); + writerThread.start(); + } + + private void insertDoc(String name) { + verifyClient.getDatabase(DB).getCollection(COLLECTION) + .insertOne(new Document("name", name)); + } + + private Map buildConfig() { + Map config = new HashMap<>(); + config.put("mongoUri", mongoContainer.getConnectionString()); + config.put("database", DB); + config.put("collection", COLLECTION); + config.put("batchSize", 2); + config.put("batchTimeMs", 500); + // FULL_SYNC replays existing documents from the start of the stream, making delivery of the + // pre-seeded documents deterministic rather than racing the subscription becoming active. + config.put("syncType", "FULL_SYNC"); + return config; + } +} From 3425d11e1c668e7cb8c3aa08d96a3a9b0314af4b Mon Sep 17 00:00:00 2001 From: David Kjerrumgaard <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 17 Jul 2026 19:01:09 -0400 Subject: [PATCH 2/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../io/mongodb/MongoSourceContainerTest.java | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java index a63e7563cb..bf683d1f5b 100644 --- a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java +++ b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java @@ -115,12 +115,12 @@ public void testReadFromChangeStream() throws Exception { insertDoc("seed-" + i); } - // Keep writing after open as well so the change stream has a steady supply regardless of + // Keep writing after open so the change stream has a steady supply regardless of // exactly when the subscription becomes active. - startBackgroundWriter(); - source.open(buildConfig(), mock(SourceContext.class)); + startBackgroundWriter(); + int received = 0; for (int i = 0; i < EXPECTED_RECORDS; i++) { Record record = readOne(); @@ -133,11 +133,13 @@ public void testReadFromChangeStream() throws Exception { assertNotNull(value.get("ns"), "record missing ns"); Document fullDocument = value.get("fullDocument", Document.class); assertNotNull(fullDocument, "record missing fullDocument"); - assertTrue(fullDocument.getString("name").startsWith("seed-") - || fullDocument.getString("name").startsWith("live-"), - "unexpected fullDocument name: " + fullDocument.getString("name")); + + String name = fullDocument.getString("name"); + assertNotNull(name, "fullDocument missing name"); + assertTrue(name.startsWith("seed-"), "unexpected fullDocument name: " + name); + log.info("Received change-stream record: key={} name={}", - record.getKey().orElse(null), fullDocument.getString("name")); + record.getKey().orElse(null), name); received++; } assertTrue(received >= EXPECTED_RECORDS, From 62bddbe20fa8923f6841a7927dfe0a1bcbb7f9ca Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 17 Jul 2026 19:05:05 -0400 Subject: [PATCH 3/3] [improve][test] Mongo source IT: drop writer, assert exact FULL_SYNC replay set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds on the autofix (writer moved after open, seed-only assertion). The background writer is now unnecessary for this test: FULL_SYNC replays the documents seeded before open() from the start of the stream, so delivery does not depend on post-open writes. Remove it, and assert the received document set equals the seeded set exactly — proving every seeded doc is replayed once and nothing extraneous is delivered, rather than only checking each record starts with 'seed-'. --- .../io/mongodb/MongoSourceContainerTest.java | 64 +++++++------------ 1 file changed, 22 insertions(+), 42 deletions(-) diff --git a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java index bf683d1f5b..7da9965bc8 100644 --- a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java +++ b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java @@ -25,14 +25,14 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; 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 java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.io.core.SourceContext; @@ -46,6 +46,10 @@ * Integration test for {@link MongoSource} that exercises the full path from a real MongoDB change * stream into the source's record queue. A {@link MongoDBContainer} provides a single-node replica * set (which change streams require) and the source is driven end-to-end against it. + * + *

The test seeds documents before opening the source and configures {@code + * syncType=FULL_SYNC}, then asserts that exactly those seeded documents are replayed through the + * change stream — so a pass proves the FULL_SYNC replay path, not merely that some record arrived. */ @Slf4j public class MongoSourceContainerTest { @@ -62,8 +66,6 @@ public class MongoSourceContainerTest { private com.mongodb.client.MongoClient verifyClient; private MongoSource source; private ExecutorService readerExecutor; - private Thread writerThread; - private final AtomicBoolean keepWriting = new AtomicBoolean(false); @BeforeMethod public void setUp() { @@ -71,7 +73,7 @@ public void setUp() { .withStartupTimeout(Duration.ofMinutes(3)); mongoContainer.start(); - // Sync driver used only to write test documents and drive the change stream. + // Sync driver used only to write the test documents. verifyClient = com.mongodb.client.MongoClients.create(mongoContainer.getConnectionString()); readerExecutor = Executors.newSingleThreadExecutor(r -> { @@ -85,10 +87,6 @@ public void setUp() { @AfterMethod(alwaysRun = true) public void tearDown() throws Exception { - keepWriting.set(false); - if (writerThread != null) { - writerThread.interrupt(); - } if (readerExecutor != null) { // A reader may still be blocked in read(), which never returns null. readerExecutor.shutdownNow(); @@ -110,18 +108,18 @@ public void tearDown() throws Exception { @Test(timeOut = 300_000) public void testReadFromChangeStream() throws Exception { - // Seed some documents before the source starts; FULL_SYNC replays them from the start. + // Seed the documents before the source starts. FULL_SYNC (startAtOperationTime=0) replays + // them from the start of the stream, so these exact documents must be delivered. + Set expectedNames = new HashSet<>(); for (int i = 0; i < EXPECTED_RECORDS; i++) { - insertDoc("seed-" + i); + String name = "seed-" + i; + insertDoc(name); + expectedNames.add(name); } - // Keep writing after open so the change stream has a steady supply regardless of - // exactly when the subscription becomes active. source.open(buildConfig(), mock(SourceContext.class)); - startBackgroundWriter(); - - int received = 0; + Set receivedNames = new HashSet<>(); for (int i = 0; i < EXPECTED_RECORDS; i++) { Record record = readOne(); assertNotNull(record, "read() returned null"); @@ -135,15 +133,17 @@ public void testReadFromChangeStream() throws Exception { assertNotNull(fullDocument, "record missing fullDocument"); String name = fullDocument.getString("name"); - assertNotNull(name, "fullDocument missing name"); - assertTrue(name.startsWith("seed-"), "unexpected fullDocument name: " + name); - + assertTrue(expectedNames.contains(name), + "received an unexpected document '" + name + "'; FULL_SYNC should replay only " + + "the seeded documents " + expectedNames); log.info("Received change-stream record: key={} name={}", record.getKey().orElse(null), name); - received++; + receivedNames.add(name); } - assertTrue(received >= EXPECTED_RECORDS, - "Expected at least " + EXPECTED_RECORDS + " records, got " + received); + + // Every seeded document was replayed exactly once through the change stream. + assertEquals(receivedNames, expectedNames, + "FULL_SYNC did not replay all seeded documents; got " + receivedNames); } /** @@ -163,26 +163,6 @@ private Record readOne() throws Exception { } } - private void startBackgroundWriter() { - keepWriting.set(true); - final AtomicInteger counter = new AtomicInteger(); - writerThread = new Thread(() -> { - while (keepWriting.get()) { - try { - insertDoc("live-" + counter.incrementAndGet()); - Thread.sleep(500); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } catch (Exception e) { - log.warn("Background write failed", e); - } - } - }, "mongo-it-writer"); - writerThread.setDaemon(true); - writerThread.start(); - } - private void insertDoc(String name) { verifyClient.getDatabase(DB).getCollection(COLLECTION) .insertOne(new Document("name", name));