This is an automated email from the ASF dual-hosted git repository. rgoers pushed a commit to branch handle-duplicates in repository https://gitbox.apache.org/repos/asf/logging-flume-mongodb.git
commit 5e12f3ab3adcdde4e077d347c4c1f2780f659d05 Author: Ralph Goers <[email protected]> AuthorDate: Tue Sep 1 21:14:10 2026 -0700 handle duplicate events --- .flattened-pom.xml | 216 +++++++++++++++++++++ flume-mongodb-sink/.flattened-pom.xml | 6 + flume-mongodb-sink/pom.xml | 7 + .../flume/sink/mongodb/DefaultMongoDbWriter.java | 38 +++- .../org/apache/flume/sink/mongodb/MongoDbSink.java | 23 ++- .../flume/sink/mongodb/MongoDbSinkCounter.java | 50 +++++ ...{MongoDbWriter.java => MongoDbWriteResult.java} | 38 ++-- .../apache/flume/sink/mongodb/MongoDbWriter.java | 10 +- .../apache/flume/sink/mongodb/TestMongoDbSink.java | 26 ++- .../sink/mongodb/TestMongoDbSinkEmbedded.java | 164 ++++++++++++++++ 10 files changed, 547 insertions(+), 31 deletions(-) diff --git a/.flattened-pom.xml b/.flattened-pom.xml new file mode 100644 index 0000000..815ad8e --- /dev/null +++ b/.flattened-pom.xml @@ -0,0 +1,216 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ 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. + --> +<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"; xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.flume</groupId> + <artifactId>flume-parent</artifactId> + <version>2.0.0-SNAPSHOT</version> + <relativePath></relativePath> + </parent> + <groupId>org.apache.flume</groupId> + <artifactId>flume-mongodb-parent</artifactId> + <version>2.0.0-SNAPSHOT</version> + <packaging>pom</packaging> + <name>Flume MongoDB Parent</name> + <url>https://logging.apache.org/flume/2.x/index.html/flume-mongodb-parent</url> + <inceptionYear>2022</inceptionYear> + <organization> + <name>Apache Software Foundation</name> + <url>http://www.apache.org</url> + </organization> + <licenses> + <license> + <name>The Apache Software License, Version 2.0</name> + <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url> + </license> + </licenses> + <developers> + <developer> + <id>rgoers</id> + <name>Ralph Goers</name> + <email>[email protected]</email> + <organization>Intuit</organization> + </developer> + </developers> + <mailingLists> + <mailingList> + <name>Flume User List</name> + <subscribe>[email protected]</subscribe> + <unsubscribe>[email protected]</unsubscribe> + <post>[email protected]</post> + <archive>http://mail-archives.apache.org/mod_mbox/flume-user/</archive> + </mailingList> + <mailingList> + <name>Flume Developer List</name> + <subscribe>[email protected]</subscribe> + <unsubscribe>[email protected]</unsubscribe> + <post>[email protected]</post> + <archive>http://mail-archives.apache.org/mod_mbox/flume-dev/</archive> + </mailingList> + <mailingList> + <name>Flume Commits</name> + <subscribe>[email protected]</subscribe> + <unsubscribe>[email protected]</unsubscribe> + <post>[email protected]</post> + <archive>http://mail-archives.apache.org/mod_mbox/flume-commits/</archive> + </mailingList> + </mailingLists> + <modules> + <module>flume-mongodb-sink</module> + </modules> + <scm> + <connection>https://gitbox.apache.org/repos/asf/flume-spring-boot.git</connection> + <developerConnection>https://gitbox.apache.org/repos/asf/flume-spring-boot.git</developerConnection> + <url>https://gitbox.apache.org/repos/asf/flume-spring-boot.git</url> + </scm> + <issueManagement> + <system>JIRA</system> + <url>https://issues.apache.org/jira/browse/FLUME</url> + </issueManagement> + <properties> + <mongodb.version>5.10.0</mongodb.version> + <jackson.version>2.17.0</jackson.version> + <findsecbugs-plugin.version>1.12.0</findsecbugs-plugin.version> + <ReleaseVersion>2.0.0</ReleaseVersion> + <ReleaseManager>Ralph Goers</ReleaseManager> + <rat.version>0.12</rat.version> + <log4j.version>2.26.1</log4j.version> + <junit.version>4.13.2</junit.version> + <ReleaseKey>B3D8E1BA</ReleaseKey> + <maven.compiler.source>11</maven.compiler.source> + <spotless-maven-plugin.version>2.27.2</spotless-maven-plugin.version> + <mvn-gpg-plugin.version>1.6</mvn-gpg-plugin.version> + <checksum-maven-plugin.version>1.11</checksum-maven-plugin.version> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + <module.name>org.apache.flume.mongodb</module.name> + <mvn-javadoc-plugin.version>2.9</mvn-javadoc-plugin.version> + <mockito.version>1.9.0</mockito.version> + <dropwizard-metrics.version>4.1.18</dropwizard-metrics.version> + <flume.version>2.0.0-SNAPSHOT</flume.version> + <spotbugs-maven-plugin.version>4.7.2.1</spotbugs-maven-plugin.version> + <maven.compiler.target>11</maven.compiler.target> + <SigningUserName>[email protected]</SigningUserName> + </properties> + <dependencyManagement> + <dependencies> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-core</artifactId> + <version>${flume.version}</version> + </dependency> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-sdk</artifactId> + <version>${flume.version}</version> + </dependency> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-configuration</artifactId> + <version>${flume.version}</version> + </dependency> + <dependency> + <groupId>org.mongodb</groupId> + <artifactId>mongodb-driver-bom</artifactId> + <version>${mongodb.version}</version> + <type>pom</type> + <scope>import</scope> + </dependency> + <dependency> + <groupId>io.dropwizard.metrics</groupId> + <artifactId>metrics-core</artifactId> + <version>${dropwizard-metrics.version}</version> + </dependency> + <dependency> + <groupId>org.apache.logging.log4j</groupId> + <artifactId>log4j-api</artifactId> + <version>${log4j.version}</version> + </dependency> + <dependency> + <groupId>org.apache.logging.log4j</groupId> + <artifactId>log4j-core</artifactId> + <version>${log4j.version}</version> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + <version>${jackson.version}</version> + </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <version>${junit.version}</version> + </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-all</artifactId> + <version>${mockito.version}</version> + <scope>test</scope> + </dependency> + </dependencies> + </dependencyManagement> + <build> + <plugins> + <plugin> + <groupId>org.apache.rat</groupId> + <artifactId>apache-rat-plugin</artifactId> + <version>${rat.version}</version> + <executions> + <execution> + <id>verify.rat</id> + <phase>verify</phase> + <goals> + <goal>check</goal> + </goals> + </execution> + </executions> + <configuration> + <excludes> + <exclude>**/.idea/</exclude> + <exclude>**/*.iml</exclude> + <exclude>src/main/resources/META-INF/services/**/*</exclude> + <exclude>**/nb-configuration.xml</exclude> + <exclude>.git/</exclude> + <exclude>patchprocess/</exclude> + <exclude>.gitignore</exclude> + <exclude>**/*.yml</exclude> + <exclude>**/*.yaml</exclude> + <exclude>**/*.json</exclude> + <exclude>.repository/</exclude> + <exclude>**/*.diff</exclude> + <exclude>**/*.patch</exclude> + <exclude>**/*.avsc</exclude> + <exclude>**/*.avro</exclude> + <exclude>**/docs/**</exclude> + <exclude>**/test/resources/**</exclude> + <exclude>**/.settings/*</exclude> + <exclude>**/.classpath</exclude> + <exclude>**/.project</exclude> + <exclude>**/target/**</exclude> + <exclude>**/derby.log</exclude> + <exclude>**/metastore_db/</exclude> + <exclude>.mvn/**</exclude> + <exclude>**/exclude-pmd.properties</exclude> + </excludes> + <consoleOutput>true</consoleOutput> + </configuration> + </plugin> + </plugins> + </build> +</project> diff --git a/flume-mongodb-sink/.flattened-pom.xml b/flume-mongodb-sink/.flattened-pom.xml index 26355e6..48e0e54 100644 --- a/flume-mongodb-sink/.flattened-pom.xml +++ b/flume-mongodb-sink/.flattened-pom.xml @@ -79,6 +79,12 @@ <artifactId>mockito-all</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>de.flapdoodle.embed</groupId> + <artifactId>de.flapdoodle.embed.mongo</artifactId> + <version>3.5.4</version> + <scope>test</scope> + </dependency> </dependencies> <build> <plugins> diff --git a/flume-mongodb-sink/pom.xml b/flume-mongodb-sink/pom.xml index da3c4b8..2f73796 100644 --- a/flume-mongodb-sink/pom.xml +++ b/flume-mongodb-sink/pom.xml @@ -83,6 +83,13 @@ <scope>test</scope> </dependency> + <dependency> + <groupId>de.flapdoodle.embed</groupId> + <artifactId>de.flapdoodle.embed.mongo</artifactId> + <version>3.5.4</version> + <scope>test</scope> + </dependency> + </dependencies> <build> diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java index d039717..28fb4ab 100644 --- a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java +++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java @@ -16,10 +16,15 @@ */ package org.apache.flume.sink.mongodb; +import com.mongodb.DuplicateKeyException; +import com.mongodb.ErrorCategory; +import com.mongodb.MongoWriteException; import com.mongodb.WriteConcern; import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoDatabase; import java.util.List; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.bson.Document; /** @@ -28,6 +33,8 @@ import org.bson.Document; */ public class DefaultMongoDbWriter implements MongoDbWriter { + private static final Logger logger = LogManager.getLogger(DefaultMongoDbWriter.class); + private final MongoDatabase mongoDatabase; private final WriteConcern writeConcern; @@ -37,12 +44,39 @@ public class DefaultMongoDbWriter implements MongoDbWriter { } @Override - public void write(String collectionName, List<Document> documents) { + public MongoDbWriteResult write(String collectionName, List<Document> documents) { MongoCollection<Document> collection = mongoDatabase.getCollection(collectionName); if (writeConcern != null) { collection = collection.withWriteConcern(writeConcern); } - collection.insertMany(documents); + + long insertedCount = 0; + long duplicateCount = 0; + // Insert one document at a time (rather than insertMany) so that a + // single duplicate key does not abort the rest of the batch. + for (Document document : documents) { + try { + collection.insertOne(document); + insertedCount++; + } catch (DuplicateKeyException ex) { + logger.warn( + "Duplicate key while inserting into collection {}, skipping event: {}", + collectionName, + ex.getMessage()); + duplicateCount++; + } catch (MongoWriteException ex) { + if (ex.getError().getCategory() == ErrorCategory.DUPLICATE_KEY) { + logger.warn( + "Duplicate key while inserting into collection {}, skipping event: {}", + collectionName, + ex.getMessage()); + duplicateCount++; + } else { + throw ex; + } + } + } + return new MongoDbWriteResult(insertedCount, duplicateCount); } @Override diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java index 9c32289..adc545a 100644 --- a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java +++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java @@ -34,7 +34,6 @@ import org.apache.flume.Transaction; import org.apache.flume.conf.BatchSizeSupported; import org.apache.flume.conf.Configurable; import org.apache.flume.conf.ConfigurationException; -import org.apache.flume.instrumentation.SinkCounter; import org.apache.flume.sink.AbstractSink; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -83,7 +82,7 @@ public class MongoDbSink extends AbstractSink implements Configurable, BatchSize private MongoClient mongoClient; private MongoDbWriter writer; - private SinkCounter counter; + private MongoDbSinkCounter counter; // For testing public String getDatabaseName() { @@ -94,6 +93,11 @@ public class MongoDbSink extends AbstractSink implements Configurable, BatchSize return defaultCollection; } + /** For testing: number of events skipped due to duplicate keys. */ + public long getDuplicateEventCount() { + return counter.getDuplicateEventCount(); + } + @Override public long getBatchSize() { return batchSize; @@ -138,12 +142,19 @@ public class MongoDbSink extends AbstractSink implements Configurable, BatchSize .add(document); } + long insertedEvents = 0; + long duplicateEvents = 0; for (Map.Entry<String, List<Document>> entry : documentsByCollection.entrySet()) { - writer.write(entry.getKey(), entry.getValue()); + MongoDbWriteResult writeResult = writer.write(entry.getKey(), entry.getValue()); + insertedEvents += writeResult.getInsertedCount(); + duplicateEvents += writeResult.getDuplicateCount(); } - if (processedEvents > 0) { - counter.addToEventDrainSuccessCount(processedEvents); + if (insertedEvents > 0) { + counter.addToEventDrainSuccessCount(insertedEvents); + } + if (duplicateEvents > 0) { + counter.addToDuplicateEventCount(duplicateEvents); } transaction.commit(); @@ -279,7 +290,7 @@ public class MongoDbSink extends AbstractSink implements Configurable, BatchSize } if (counter == null) { - counter = new SinkCounter(getName()); + counter = new MongoDbSinkCounter(getName()); } } } diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSinkCounter.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSinkCounter.java new file mode 100644 index 0000000..97737d2 --- /dev/null +++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSinkCounter.java @@ -0,0 +1,50 @@ +/* + * 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.flume.sink.mongodb; + +import org.apache.flume.instrumentation.SinkCounter; + +/** + * {@link SinkCounter} extension that additionally tracks the number of + * events skipped because they duplicated a document already present in + * MongoDB (i.e. resulted in a {@code com.mongodb.DuplicateKeyException}). + * These events are not counted towards {@code eventDrainSuccessCount} since + * they were not actually inserted, but they should also not be treated as a + * batch failure. + */ +public class MongoDbSinkCounter extends SinkCounter { + + private static final String COUNTER_DUPLICATE_EVENT = "sink.event.duplicate"; + + private static final String[] ATTRIBUTES = {COUNTER_DUPLICATE_EVENT}; + + public MongoDbSinkCounter(String name) { + super(name, ATTRIBUTES); + } + + public long getDuplicateEventCount() { + return get(COUNTER_DUPLICATE_EVENT); + } + + public long incrementDuplicateEventCount() { + return increment(COUNTER_DUPLICATE_EVENT); + } + + public long addToDuplicateEventCount(long delta) { + return addAndGet(COUNTER_DUPLICATE_EVENT, delta); + } +} diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriteResult.java similarity index 53% copy from flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java copy to flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriteResult.java index 005fb53..d697119 100644 --- a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java +++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriteResult.java @@ -16,28 +16,30 @@ */ package org.apache.flume.sink.mongodb; -import java.util.List; -import org.bson.Document; - /** - * Abstraction over the MongoDB write path used by {@link MongoDbSink}. - * Separating this out of the sink keeps the routing/document-construction - * logic in {@link MongoDbSink} independent of, and easily testable without, - * a live MongoDB connection. + * Outcome of writing a batch of documents to a single MongoDB collection. */ -public interface MongoDbWriter { +public final class MongoDbWriteResult { - /** - * Writes the given documents to the named collection. - * - * @param collectionName the target collection name - * @param documents the documents to insert, in order - */ - void write(String collectionName, List<Document> documents); + private final long insertedCount; + private final long duplicateCount; + + public MongoDbWriteResult(long insertedCount, long duplicateCount) { + this.insertedCount = insertedCount; + this.duplicateCount = duplicateCount; + } + + /** Number of documents that were successfully inserted. */ + public long getInsertedCount() { + return insertedCount; + } /** - * Releases any resources (e.g. the underlying MongoDB client) held by - * this writer. + * Number of documents that were skipped because they duplicated a + * document that already existed in the collection (i.e. triggered a + * {@code com.mongodb.DuplicateKeyException}). */ - void close(); + public long getDuplicateCount() { + return duplicateCount; + } } diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java index 005fb53..d458bc6 100644 --- a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java +++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java @@ -28,12 +28,18 @@ import org.bson.Document; public interface MongoDbWriter { /** - * Writes the given documents to the named collection. + * Writes the given documents to the named collection. Documents that + * fail to insert because they duplicate an existing document (i.e. + * trigger a {@code com.mongodb.DuplicateKeyException}) are skipped + * rather than causing the whole batch to fail; they are reported via + * {@link MongoDbWriteResult#getDuplicateCount()}. * * @param collectionName the target collection name * @param documents the documents to insert, in order + * @return the number of documents inserted and the number skipped as + * duplicates */ - void write(String collectionName, List<Document> documents); + MongoDbWriteResult write(String collectionName, List<Document> documents); /** * Releases any resources (e.g. the underlying MongoDB client) held by diff --git a/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java index a2d2846..38f3bd7 100644 --- a/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java +++ b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java @@ -36,7 +36,6 @@ import org.apache.flume.channel.MemoryChannel; import org.apache.flume.conf.Configurables; import org.apache.flume.conf.ConfigurationException; import org.apache.flume.event.EventBuilder; -import org.apache.flume.instrumentation.SinkCounter; import org.bson.Document; import org.junit.Test; import org.mockito.internal.util.reflection.Whitebox; @@ -50,10 +49,14 @@ public class TestMongoDbSink { private static final class FakeMongoDbWriter implements MongoDbWriter { private final Map<String, List<Document>> written = new LinkedHashMap<>(); private boolean closed = false; + private long duplicateCountToReport = 0; @Override - public void write(String collectionName, List<Document> documents) { + public MongoDbWriteResult write(String collectionName, List<Document> documents) { written.computeIfAbsent(collectionName, k -> new ArrayList<>()).addAll(documents); + long duplicates = Math.min(duplicateCountToReport, documents.size()); + duplicateCountToReport -= duplicates; + return new MongoDbWriteResult(documents.size() - duplicates, duplicates); } @Override @@ -78,7 +81,7 @@ public class TestMongoDbSink { channel.start(); Configurables.configure(sink, context); Whitebox.setInternalState(sink, "writer", writer); - Whitebox.setInternalState(sink, "counter", new SinkCounter("test")); + Whitebox.setInternalState(sink, "counter", new MongoDbSinkCounter("test")); return sink; } @@ -114,6 +117,23 @@ public class TestMongoDbSink { new MongoDbSink().configure(context); } + @Test + public void testDuplicateEventsDoNotFailBatchAndAreCountedSeparately() throws EventDeliveryException { + FakeMongoDbWriter writer = new FakeMongoDbWriter(); + writer.duplicateCountToReport = 1; + Context context = baseContext(); + MongoDbSink sink = createSink(context, writer); + Channel channel = sink.getChannel(); + + putEvent(channel, "{\"foo\":\"1\"}".getBytes(StandardCharsets.UTF_8), new HashMap<String, String>()); + putEvent(channel, "{\"foo\":\"2\"}".getBytes(StandardCharsets.UTF_8), new HashMap<String, String>()); + + Sink.Status status = sink.process(); + + assertEquals(Sink.Status.READY, status); + assertEquals(1, sink.getDuplicateEventCount()); + } + @Test public void testWritesToDefaultCollectionWhenNoHeaderConfigured() throws EventDeliveryException { FakeMongoDbWriter writer = new FakeMongoDbWriter(); diff --git a/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSinkEmbedded.java b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSinkEmbedded.java new file mode 100644 index 0000000..cc4ffbf --- /dev/null +++ b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSinkEmbedded.java @@ -0,0 +1,164 @@ +/* + * 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.flume.sink.mongodb; + +import static org.junit.Assert.assertEquals; + +import com.mongodb.client.MongoClient; +import com.mongodb.client.MongoClients; +import com.mongodb.client.MongoCollection; +import com.mongodb.client.model.IndexOptions; +import com.mongodb.client.model.Indexes; +import de.flapdoodle.embed.mongo.MongodExecutable; +import de.flapdoodle.embed.mongo.MongodProcess; +import de.flapdoodle.embed.mongo.MongodStarter; +import de.flapdoodle.embed.mongo.config.MongodConfig; +import de.flapdoodle.embed.mongo.config.Net; +import de.flapdoodle.embed.mongo.distribution.Version; +import de.flapdoodle.embed.process.runtime.Network; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import org.apache.flume.Channel; +import org.apache.flume.Context; +import org.apache.flume.EventDeliveryException; +import org.apache.flume.Sink; +import org.apache.flume.Transaction; +import org.apache.flume.channel.MemoryChannel; +import org.apache.flume.conf.Configurables; +import org.apache.flume.event.EventBuilder; +import org.bson.Document; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Integration tests that exercise {@link MongoDbSink} against a real (but + * embedded/in-memory) MongoDB instance provided by Flapdoodle, verifying + * that duplicate key errors are tolerated instead of failing the whole + * batch and are tracked via a dedicated counter. + */ +public class TestMongoDbSinkEmbedded { + + private static MongodExecutable mongodExecutable; + private static MongoClient mongoClient; + private static int port; + + @BeforeClass + public static void startMongo() throws Exception { + port = Network.getFreeServerPort(); + MongodStarter starter = MongodStarter.getDefaultInstance(); + MongodConfig mongodConfig = MongodConfig.builder() + .version(Version.Main.V4_4) + .net(new Net(port, Network.localhostIsIPv6())) + .build(); + MongodExecutable executable = starter.prepare(mongodConfig); + MongodProcess process = executable.start(); + mongodExecutable = executable; + mongoClient = MongoClients.create("mongodb://localhost:" + port); + // Keep a reference so the process isn't garbage collected/stopped early. + assert process != null; + } + + @AfterClass + public static void stopMongo() { + if (mongoClient != null) { + mongoClient.close(); + } + if (mongodExecutable != null) { + mongodExecutable.stop(); + } + } + + private static Context baseContext(String database, String collection) { + Context context = new Context(); + context.put(MongoDbSinkConstants.CONNECTION_URI, "mongodb://localhost:" + port); + context.put(MongoDbSinkConstants.DATABASE_NAME, database); + context.put(MongoDbSinkConstants.COLLECTION, collection); + return context; + } + + private static MongoDbSink createAndStartSink(Context context) { + MongoDbSink sink = new MongoDbSink(); + Channel channel = new MemoryChannel(); + Configurables.configure(channel, new Context()); + sink.setChannel(channel); + channel.start(); + Configurables.configure(sink, context); + sink.start(); + return sink; + } + + private static void putEvent(Channel channel, String json) { + Transaction tx = channel.getTransaction(); + tx.begin(); + channel.put(EventBuilder.withBody(json.getBytes(StandardCharsets.UTF_8), new HashMap<>())); + tx.commit(); + tx.close(); + } + + @Test + public void testDuplicateKeyDoesNotFailBatchAndIsCountedSeparately() throws EventDeliveryException { + String database = "testDb1"; + String collectionName = "events"; + Context context = baseContext(database, collectionName); + MongoDbSink sink = createAndStartSink(context); + try { + MongoCollection<Document> collection = + mongoClient.getDatabase(database).getCollection(collectionName); + collection.createIndex(Indexes.ascending("uid"), new IndexOptions().unique(true)); + + Channel channel = sink.getChannel(); + // Two distinct events plus one that duplicates the first's unique key. + putEvent(channel, "{\"uid\":1,\"value\":\"a\"}"); + putEvent(channel, "{\"uid\":2,\"value\":\"b\"}"); + putEvent(channel, "{\"uid\":1,\"value\":\"c\"}"); + + Sink.Status status = sink.process(); + + assertEquals(Sink.Status.READY, status); + assertEquals(2, collection.countDocuments()); + assertEquals(1, sink.getDuplicateEventCount()); + } finally { + sink.stop(); + } + } + + @Test + public void testNoDuplicatesLeavesDuplicateCountAtZero() throws EventDeliveryException { + String database = "testDb2"; + String collectionName = "events"; + Context context = baseContext(database, collectionName); + MongoDbSink sink = createAndStartSink(context); + try { + MongoCollection<Document> collection = + mongoClient.getDatabase(database).getCollection(collectionName); + collection.createIndex(Indexes.ascending("uid"), new IndexOptions().unique(true)); + + Channel channel = sink.getChannel(); + putEvent(channel, "{\"uid\":1,\"value\":\"a\"}"); + putEvent(channel, "{\"uid\":2,\"value\":\"b\"}"); + + Sink.Status status = sink.process(); + + assertEquals(Sink.Status.READY, status); + assertEquals(2, collection.countDocuments()); + assertEquals(0, sink.getDuplicateEventCount()); + } finally { + sink.stop(); + } + } +}
