diff --git a/flume-mongodb-sink/pom.xml b/flume-mongodb-sink/pom.xml index cfaeaa8..5d0106f 100644 --- a/flume-mongodb-sink/pom.xml +++ b/flume-mongodb-sink/pom.xml @@ -67,6 +67,12 @@ test + + org.testcontainers + mongodb + test + + org.apache.logging.log4j log4j-api @@ -78,5 +84,34 @@ test + + org.apache.logging.log4j + log4j-slf4j2-impl + test + + + + + + + + org.apache.maven.plugins + maven-failsafe-plugin + + + + integration-test + verify + + + + + + + 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..73fb4db 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,20 @@ */ package org.apache.flume.sink.mongodb; +import com.mongodb.ErrorCategory; +import com.mongodb.MongoBulkWriteException; import com.mongodb.WriteConcern; +import com.mongodb.bulk.BulkWriteError; import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoDatabase; +import com.mongodb.client.model.InsertManyOptions; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.bson.Document; /** @@ -28,6 +38,20 @@ */ public class DefaultMongoDbWriter implements MongoDbWriter { + private static final Logger logger = LogManager.getLogger(DefaultMongoDbWriter.class); + + /** + * Unordered inserts let the server attempt every document of the batch. + */ + private static final InsertManyOptions UNORDERED = new InsertManyOptions().ordered(false); + + /** + * Matches the index name and key value in a server duplicate key message. + */ + private static final Pattern DUPLICATE_KEY_MESSAGE = Pattern.compile("index:\\s+(\\S+)\\s+dup key:\\s+(\\{.*?\\})"); + + private static final String UNKNOWN_INDEX_NAME = ""; + private final MongoDatabase mongoDatabase; private final WriteConcern writeConcern; @@ -37,12 +61,72 @@ public DefaultMongoDbWriter(MongoDatabase mongoDatabase, WriteConcern writeConce } @Override - public void write(String collectionName, List documents) { + public MongoDbWriteResult write(String collectionName, List documents) { MongoCollection collection = mongoDatabase.getCollection(collectionName); if (writeConcern != null) { collection = collection.withWriteConcern(writeConcern); } - collection.insertMany(documents); + + // Try to insert the whole batch in a single operation, + // which is far more efficient than one insert per document. + try { + collection.insertMany(documents, UNORDERED); + return new MongoDbWriteResult(documents.size(), 0); + } catch (MongoBulkWriteException ex) { + if (isDuplicateKeyOnly(ex)) { + List errors = ex.getWriteErrors(); + logDuplicates(collectionName, documents, errors); + return new MongoDbWriteResult(documents.size() - errors.size(), errors.size()); + } + throw ex; + } + } + + /** + * Returns {@code true} if every error reported by the bulk write failure + * is a duplicate key error. + */ + private boolean isDuplicateKeyOnly(MongoBulkWriteException ex) { + if (ex.getWriteConcernError() != null) { + return false; + } + List errors = ex.getWriteErrors(); + if (errors.isEmpty()) { + return false; + } + for (BulkWriteError error : errors) { + if (ErrorCategory.fromErrorCode(error.getCode()) != ErrorCategory.DUPLICATE_KEY) { + return false; + } + } + return true; + } + + /** + * Reports how many documents of the batch were rejected, broken down by + * the unique index that rejected them, and logs the offending documents + * themselves at debug level. + */ + private void logDuplicates(String collectionName, List documents, List errors) { + Map duplicatesByIndexName = new LinkedHashMap<>(); + for (BulkWriteError error : errors) { + Matcher matcher = DUPLICATE_KEY_MESSAGE.matcher(error.getMessage()); + String indexName = matcher.find() ? matcher.group(1) : UNKNOWN_INDEX_NAME; + duplicatesByIndexName.merge(indexName, 1, Integer::sum); + if (logger.isDebugEnabled()) { + logger.debug( + "Duplicate key in collection {} for the event at position {} of the batch: {}", + collectionName, + error.getIndex(), + documents.get(error.getIndex()).toJson()); + } + } + logger.warn( + "Skipped {} of {} event(s) written to collection {} as duplicates, per unique index: {}", + errors.size(), + documents.size(), + collectionName, + duplicatesByIndexName); } @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 518916e..439bc2d 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 @@ -36,7 +36,6 @@ 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; @@ -86,7 +85,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() { @@ -97,6 +96,11 @@ public String getDefaultCollection() { return defaultCollection; } + /** For testing: number of events skipped due to duplicate keys. */ + public long getDuplicateEventCount() { + return counter.getDuplicateEventCount(); + } + @Override public long getBatchSize() { return batchSize; @@ -141,12 +145,19 @@ public Status process() throws EventDeliveryException { .add(document); } + long insertedEvents = 0; + long duplicateEvents = 0; for (Map.Entry> 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(); @@ -284,7 +295,7 @@ public void configure(Context context) { } 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/MongoDbWriteResult.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriteResult.java new file mode 100644 index 0000000..d697119 --- /dev/null +++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriteResult.java @@ -0,0 +1,45 @@ +/* + * 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; + +/** + * Outcome of writing a batch of documents to a single MongoDB collection. + */ +public final class MongoDbWriteResult { + + 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; + } + + /** + * 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}). + */ + 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..106eee6 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,19 @@ 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. + * violate a unique index) are skipped rather than causing the whole + * batch to fail; they are reported via + * {@link MongoDbWriteResult#getDuplicateCount()}. Implementations may + * insert the documents in any order. * * @param collectionName the target collection name - * @param documents the documents to insert, in order + * @param documents the documents to insert + * @return the number of documents inserted and the number skipped as + * duplicates */ - void write(String collectionName, List documents); + MongoDbWriteResult write(String collectionName, List 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/MongoDbSinkIT.java b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/MongoDbSinkIT.java new file mode 100644 index 0000000..a99f89b --- /dev/null +++ b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/MongoDbSinkIT.java @@ -0,0 +1,214 @@ +/* + * 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 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.Assume; +import org.junit.BeforeClass; +import org.junit.Test; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.MongoDBContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * Integration tests that exercise {@link MongoDbSink} against MongoDB running + * in a Testcontainers-managed Docker container. The tests are skipped + * (rather than failed) when Docker is not available in the environment. + */ +public class MongoDbSinkIT { + + private static final DockerImageName MONGO_IMAGE = DockerImageName.parse("mongo:latest"); + + private static MongoDBContainer mongoDBContainer; + private static MongoClient mongoClient; + + @BeforeClass + public static void startMongoContainer() { + Assume.assumeTrue( + "Docker is not available, skipping MongoDbSinkIT", + DockerClientFactory.instance().isDockerAvailable()); + + mongoDBContainer = new MongoDBContainer(MONGO_IMAGE); + mongoDBContainer.start(); + mongoClient = MongoClients.create(mongoDBContainer.getConnectionString()); + } + + @AfterClass + public static void stopMongoContainer() { + if (mongoClient != null) { + mongoClient.close(); + } + if (mongoDBContainer != null) { + mongoDBContainer.stop(); + } + } + + private static Context baseContext(String database, String collection) { + Context context = new Context(); + context.put(MongoDbSinkConstants.CONNECTION_URI, mongoDBContainer.getConnectionString()); + 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 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 testDuplicatesAnywhereInTheBatchAreSkippedIndividually() throws EventDeliveryException { + String database = "testDb3"; + String collectionName = "events"; + Context context = baseContext(database, collectionName); + MongoDbSink sink = createAndStartSink(context); + try { + MongoCollection collection = + mongoClient.getDatabase(database).getCollection(collectionName); + collection.createIndex(Indexes.ascending("uid"), new IndexOptions().unique(true)); + collection.insertOne(new Document("uid", 1).append("value", "pre-existing")); + + Channel channel = sink.getChannel(); + // The very first event of the batch is a duplicate, as is the last + // one; the events in between must still be inserted. + putEvent(channel, "{\"uid\":1,\"value\":\"a\"}"); + putEvent(channel, "{\"uid\":2,\"value\":\"b\"}"); + putEvent(channel, "{\"uid\":3,\"value\":\"c\"}"); + putEvent(channel, "{\"uid\":1,\"value\":\"d\"}"); + + Sink.Status status = sink.process(); + + assertEquals(Sink.Status.READY, status); + assertEquals(3, collection.countDocuments()); + assertEquals(2, sink.getDuplicateEventCount()); + } finally { + sink.stop(); + } + } + + @Test + public void testDuplicatesWithinTheSameBatchAreSkipped() throws EventDeliveryException { + String database = "testDb4"; + String collectionName = "events"; + Context context = baseContext(database, collectionName); + MongoDbSink sink = createAndStartSink(context); + try { + MongoCollection collection = + mongoClient.getDatabase(database).getCollection(collectionName); + collection.createIndex(Indexes.ascending("uid"), new IndexOptions().unique(true)); + + Channel channel = sink.getChannel(); + // Nothing is pre-existing: the duplicates are between events of the + // batch itself, so only the first occurrence of each key survives. + putEvent(channel, "{\"uid\":1,\"value\":\"a\"}"); + putEvent(channel, "{\"uid\":1,\"value\":\"b\"}"); + putEvent(channel, "{\"uid\":1,\"value\":\"c\"}"); + + Sink.Status status = sink.process(); + + assertEquals(Sink.Status.READY, status); + assertEquals(1, collection.countDocuments()); + assertEquals(2, 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 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(); + } + } +} 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 c1fa57e..bab3cfc 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 @@ -38,7 +38,6 @@ 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; @@ -50,10 +49,14 @@ public class TestMongoDbSink { */ private static final class FakeMongoDbWriter implements MongoDbWriter { private final Map> written = new LinkedHashMap<>(); + private long duplicateCountToReport = 0; @Override - public void write(String collectionName, List documents) { + public MongoDbWriteResult write(String collectionName, List 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 @@ -96,7 +99,7 @@ private static MongoDbSink createSink(Context context, MongoDbWriter writer) { channel.start(); Configurables.configure(sink, context); setInternalState(sink, "writer", writer); - setInternalState(sink, "counter", new SinkCounter("test")); + setInternalState(sink, "counter", new MongoDbSinkCounter("test")); return sink; } @@ -188,6 +191,23 @@ public void testConfigureEmptyMappedCollection() { assertConfigurationFailure(context, "Invalid MongoDB collection name in `mongodb.collectionMap.typeA`: ``"); } + @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()); + putEvent(channel, "{\"foo\":\"2\"}".getBytes(StandardCharsets.UTF_8), new HashMap()); + + 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/pom.xml b/pom.xml index 97f5cb3..a242437 100644 --- a/pom.xml +++ b/pom.xml @@ -55,6 +55,7 @@ 4.2.39 5.10.0 + 1.21.4 @@ -74,6 +75,14 @@ pom import + + + org.testcontainers + testcontainers-bom + ${testcontainers.version} + pom + import +