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..7da9965bc8 --- /dev/null +++ b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceContainerTest.java @@ -0,0 +1,183 @@ +/* + * 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.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 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. + * + *

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 { + + 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; + + @BeforeMethod + public void setUp() { + mongoContainer = new MongoDBContainer("mongo:6.0") + .withStartupTimeout(Duration.ofMinutes(3)); + mongoContainer.start(); + + // Sync driver used only to write the test documents. + 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 { + 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 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++) { + String name = "seed-" + i; + insertDoc(name); + expectedNames.add(name); + } + + source.open(buildConfig(), mock(SourceContext.class)); + + Set receivedNames = new HashSet<>(); + 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"); + + String name = fullDocument.getString("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); + receivedNames.add(name); + } + + // Every seeded document was replayed exactly once through the change stream. + assertEquals(receivedNames, expectedNames, + "FULL_SYNC did not replay all seeded documents; got " + receivedNames); + } + + /** + * 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 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; + } +}