Mongo sink connector must tolerate the ErrantRecordReporter being not available (#100)

KAFKA-286

Author: stIncMale

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

@@ -22,18 +22,19 @@
 import java.util.Collection;
 import java.util.List;
 
-import org.apache.kafka.connect.sink.ErrantRecordReporter;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.mongodb.kafka.connect.sink.dlq.ErrorReporter;
+
 final class MongoSinkRecordProcessor {
   private static final Logger LOGGER = LoggerFactory.getLogger(MongoSinkRecordProcessor.class);
 
   static List<List<MongoProcessedSinkRecordData>> orderedGroupByTopicAndNamespace(
       final Collection<SinkRecord> records,
       final MongoSinkConfig sinkConfig,
-      final ErrantRecordReporter errorReporter) {
+      final ErrorReporter errorReporter) {
     LOGGER.debug("Number of sink records to process: {}", records.size());
 
     List<List<MongoProcessedSinkRecordData>> orderedProcessedSinkRecordData = new ArrayList<>();

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

@@ -25,8 +25,6 @@
 
 import java.util.Collection;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
 
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -43,6 +41,7 @@
 import com.mongodb.client.MongoClients;
 
 import com.mongodb.kafka.connect.Versions;
+import com.mongodb.kafka.connect.sink.dlq.ErrorReporter;
 import com.mongodb.kafka.connect.util.VisibleForTesting;
 
 public class MongoSinkTask extends SinkTask {
@@ -119,13 +118,13 @@ public void stop() {
     }
   }
 
-  private ErrantRecordReporter createErrorReporter() {
-    ErrantRecordReporter result = nopErrorReporter();
+  private ErrorReporter createErrorReporter() {
+    ErrorReporter result = nopErrorReporter();
     if (context != null) {
       try {
         ErrantRecordReporter errantRecordReporter = context.errantRecordReporter();
         if (errantRecordReporter != null) {
-          result = errantRecordReporter;
+          result = errantRecordReporter::report;
         } else {
           LOGGER.info("Errant record reporter not configured.");
         }
@@ -138,9 +137,8 @@ private ErrantRecordReporter createErrorReporter() {
   }
 
   @VisibleForTesting(otherwise = PRIVATE)
-  static ErrantRecordReporter nopErrorReporter() {
-    Future<Void> completedFuture = CompletableFuture.completedFuture(null);
-    return (record, e) -> completedFuture;
+  static ErrorReporter nopErrorReporter() {
+    return (record, e) -> {};
   }
 
   private static MongoClient createMongoClient(final MongoSinkConfig sinkConfig) {

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

@@ -27,7 +27,6 @@
 import java.util.stream.Collectors;
 
 import org.apache.kafka.connect.errors.DataException;
-import org.apache.kafka.connect.sink.ErrantRecordReporter;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,19 +41,20 @@
 import com.mongodb.client.model.WriteModel;
 
 import com.mongodb.kafka.connect.sink.dlq.AnalyzedBatchFailedWithBulkWriteException;
+import com.mongodb.kafka.connect.sink.dlq.ErrorReporter;
 
 public final class StartedMongoSinkTask {
   private static final Logger LOGGER = LoggerFactory.getLogger(MongoSinkTask.class);
 
   private final MongoSinkConfig sinkConfig;
   private final MongoClient mongoClient;
-  private final ErrantRecordReporter errorReporter;
+  private final ErrorReporter errorReporter;
   private final Set<MongoNamespace> checkedTimeseriesNamespaces;
 
   StartedMongoSinkTask(
       final MongoSinkConfig sinkConfig,
       final MongoClient mongoClient,
-      final ErrantRecordReporter errorReporter) {
+      final ErrorReporter errorReporter) {
     this.sinkConfig = sinkConfig;
     this.mongoClient = mongoClient;
     this.errorReporter = errorReporter;

src/main/java/com/mongodb/kafka/connect/sink/dlq/AnalyzedBatchFailedWithBulkWriteException.java

@@ -26,7 +26,6 @@
 import java.util.Map;
 import java.util.stream.IntStream;
 
-import org.apache.kafka.connect.sink.ErrantRecordReporter;
 import org.apache.kafka.connect.sink.SinkRecord;
 
 import com.mongodb.MongoBulkWriteException;
@@ -44,7 +43,7 @@
 public final class AnalyzedBatchFailedWithBulkWriteException {
   private final List<SinkRecord> batch;
   private final MongoBulkWriteException e;
-  private final ErrantRecordReporter errorReporter;
+  private final ErrorReporter errorReporter;
   private final Logger logger;
   private final Map<Integer, Map.Entry<SinkRecord, WriteException>> recordsFailedWithWriteError =
       new HashMap<>();
@@ -57,7 +56,7 @@ public AnalyzedBatchFailedWithBulkWriteException(
       final List<SinkRecord> batch,
       final boolean ordered,
       final MongoBulkWriteException e,
-      final ErrantRecordReporter errorReporter,
+      final ErrorReporter errorReporter,
       final Logger logger) {
     this.batch = batch;
     this.e = e;

src/main/java/com/mongodb/kafka/connect/sink/dlq/ErrorReporter.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed 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 com.mongodb.kafka.connect.sink.dlq;
+
+import org.apache.kafka.connect.sink.ErrantRecordReporter;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+/**
+ * This interface exists to avoid exposing our code to {@link ErrantRecordReporter}. This allows
+ * tolerating situations when the class {@link ErrantRecordReporter} is not available.
+ */
+public interface ErrorReporter {
+  void report(SinkRecord record, Exception e);
+}

src/test/java/com/mongodb/kafka/connect/sink/StartedMongoSinkTaskTest.java

@@ -48,16 +48,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.DataException;
-import org.apache.kafka.connect.sink.ErrantRecordReporter;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -81,6 +78,7 @@
 import com.mongodb.client.model.WriteModel;
 
 import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.ErrorTolerance;
+import com.mongodb.kafka.connect.sink.dlq.ErrorReporter;
 import com.mongodb.kafka.connect.sink.dlq.WriteConcernException;
 import com.mongodb.kafka.connect.sink.dlq.WriteException;
 import com.mongodb.kafka.connect.sink.dlq.WriteSkippedException;
@@ -357,13 +355,12 @@ static boolean match(
     }
   }
 
-  private static final class InMemoryErrorReporter implements ErrantRecordReporter {
+  private static final class InMemoryErrorReporter implements ErrorReporter {
     private final List<ReportedData> reported = new ArrayList<>();
 
     @Override
-    public Future<Void> report(final SinkRecord record, final Throwable e) {
+    public void report(final SinkRecord record, final Exception e) {
       reported.add(new ReportedData(record, e));
-      return CompletableFuture.completedFuture(null);
     }
 
     List<ReportedData> reported() {