Improved error logging for bulk write exceptions.

KAFKA-237

Author: rozza

CHANGELOG.md

@@ -5,6 +5,7 @@
 ## 1.6.0
 
 ### Improvements
+  - [KAFKA-237](https://jira.mongodb.org/browse/KAFKA-237) Improved error logging for bulk write exceptions.
   - [KAFKA-181](https://jira.mongodb.org/browse/KAFKA-181) Added support for serverApi.
   - [KAFKA-228](https://jira.mongodb.org/browse/KAFKA-228) Added support for automatic timeseries collection creation for 5.0
   - [KAFKA-215](https://jira.mongodb.org/browse/KAFKA-215) Added mongo specific override options for error handling properties

src/integrationTest/java/com/mongodb/kafka/connect/sink/MongoSinkTaskIntegrationTest.java

@@ -40,6 +40,8 @@
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
@@ -49,6 +51,7 @@
 
 import org.bson.Document;
 
+import com.mongodb.kafka.connect.log.LogCapture;
 import com.mongodb.kafka.connect.mongodb.MongoKafkaTestCase;
 
 @RunWith(JUnitPlatform.class)
@@ -116,6 +119,38 @@ void testSinkProcessesTimeseriesData() {
     }
   }
 
+  @Test
+  @DisplayName("Ensure bulk write operation error write models are included in the log")
+  void testBulkWriteOperationErrorWriteModelsIncludedInTheLog() {
+    try (LogCapture logCapture = new LogCapture(Logger.getLogger(MongoSinkTask.class))) {
+      try (AutoCloseableSinkTask task = createSinkTask()) {
+        Map<String, String> cfg = createSettings();
+        cfg.put(
+            MongoSinkTopicConfig.WRITEMODEL_STRATEGY_CONFIG,
+            "com.mongodb.kafka.connect.sink.writemodel.strategy.InsertOneDefaultStrategy");
+        task.start(cfg);
+        List<Document> documents = createDocuments(rangeClosed(1, 10));
+        documents.add(new Document("_id", 4));
+        List<SinkRecord> sinkRecords = createRecords(documents);
+        try {
+          task.put(sinkRecords);
+        } catch (Exception e) {
+          // ignore
+        }
+      }
+
+      assertTrue(
+          logCapture.getEvents().stream()
+              .filter(e -> e.getLevel().equals(Level.ERROR))
+              .anyMatch(
+                  e ->
+                      e.getMessage()
+                          .toString()
+                          .startsWith(
+                              "WriteErrors: [BulkWriteError{writeModel=InsertOneModel{document={\"_id\": 4}}")));
+    }
+  }
+
   @Test
   @DisplayName("Ensure sink can handle Tombstone null events")
   void testSinkCanHandleTombstoneNullEvents() {

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTask.java

@@ -26,6 +26,7 @@
 import static com.mongodb.kafka.connect.util.TimeseriesValidation.validateCollection;
 import static java.util.Collections.emptyList;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
@@ -54,6 +55,7 @@
 import com.mongodb.MongoClientSettings;
 import com.mongodb.MongoException;
 import com.mongodb.MongoNamespace;
+import com.mongodb.bulk.BulkWriteError;
 import com.mongodb.bulk.BulkWriteResult;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
@@ -236,7 +238,7 @@ private void bulkWriteBatch(final List<MongoProcessedSinkRecordData> batch) {
           "Writing {} document(s) into collection [{}] failed.",
           writeModels.size(),
           namespace.getFullName());
-      handleMongoException(config, e);
+      handleMongoException(config, writeModels, e);
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       throw new DataException("Rate limiting was interrupted", e);
@@ -279,14 +281,19 @@ private AtomicInteger getRemainingRetriesForTopic(final String topic) {
     return remainingRetriesTopicMap.get(topic);
   }
 
-  private void handleMongoException(final MongoSinkTopicConfig config, final MongoException e) {
+  private void handleMongoException(
+      final MongoSinkTopicConfig config,
+      final List<WriteModel<BsonDocument>> writeModels,
+      final MongoException e) {
     if (getRemainingRetriesForTopic(config.getTopic()).decrementAndGet() <= 0) {
       if (config.logErrors()) {
         LOGGER.error("Error on mongodb operation", e);
         if (e instanceof MongoBulkWriteException) {
           LOGGER.error("Mongodb bulk write (partially) failed", e);
           LOGGER.error("WriteResult: {}", ((MongoBulkWriteException) e).getWriteResult());
-          LOGGER.error("WriteErrors: {}", ((MongoBulkWriteException) e).getWriteErrors());
+          LOGGER.error(
+              "WriteErrors: {}",
+              generateWriteErrors(((MongoBulkWriteException) e).getWriteErrors(), writeModels));
           LOGGER.error(
               "WriteConcernError: {}", ((MongoBulkWriteException) e).getWriteConcernError());
         }
@@ -301,4 +308,29 @@ private void handleMongoException(final MongoSinkTopicConfig config, final Mongo
       throw new RetriableException(e.getMessage(), e);
     }
   }
+
+  private String generateWriteErrors(
+      final List<BulkWriteError> bulkWriteErrorList,
+      final List<WriteModel<BsonDocument>> writeModels) {
+    List<String> errorString = new ArrayList<>();
+    for (final BulkWriteError bulkWriteError : bulkWriteErrorList) {
+      if (bulkWriteError.getIndex() < writeModels.size()) {
+        errorString.add(
+            "BulkWriteError{"
+                + "writeModel="
+                + writeModels.get(bulkWriteError.getIndex())
+                + ", code="
+                + bulkWriteError.getCode()
+                + ", message='"
+                + bulkWriteError.getMessage()
+                + '\''
+                + ", details="
+                + bulkWriteError.getDetails()
+                + '}');
+      } else {
+        errorString.add(bulkWriteError.toString());
+      }
+    }
+    return "[" + String.join(", ", errorString) + "]";
+  }
 }