Fixed Source Connector recovery during getMore call (#79)

rozza · web-flow · commit 2f1b034437cb · 2021-07-09T09:43:22.000+01:00
KAFKA-230 KAFKA-219
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -15,6 +15,8 @@
 ### Bug Fixes
   - [KAFKA-218](https://jira.mongodb.org/browse/KAFKA-218) Fixed bug in LazyBsonDocument#clone ignoring any changes made once unwrapped.
   - [KAFKA-220](https://jira.mongodb.org/browse/KAFKA-220) Fixed bug with timestamp integer overflow.
+  - [KAFKA-219](https://jira.mongodb.org/browse/KAFKA-219) Fixed Source Connector unable to recover from broken change stream due to event > 16MB
+  - [KAFKA-230](https://jira.mongodb.org/browse/KAFKA-230) Fixed Source Connector recovery during getMore call
 
 ## 1.5.1
 
diff --git a/src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest.java b/src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest.java
@@ -25,6 +25,7 @@
 import static java.util.Collections.singletonList;
 import static java.util.Collections.singletonMap;
 import static java.util.stream.Collectors.toList;
+import static java.util.stream.IntStream.range;
 import static java.util.stream.IntStream.rangeClosed;
 import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -69,6 +70,7 @@
 
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.Updates;
 
 import com.mongodb.kafka.connect.log.LogCapture;
 import com.mongodb.kafka.connect.mongodb.ChangeStreamOperations.ChangeStreamOperation;
@@ -750,6 +752,49 @@ void testErrorToleranceNoneSupport() {
     }
   }
 
+  @Test
+  @DisplayName("Ensure source honours error tolerance all and > 16mb change stream message")
+  void testErrorToleranceAllSupport16MbError() {
+    try (AutoCloseableSourceTask task = createSourceTask(Logger.getLogger(MongoSourceTask.class))) {
+      MongoCollection<Document> coll = getAndCreateCollection();
+      HashMap<String, String> cfg =
+          new HashMap<String, String>() {
+            {
+              put(MongoSourceConfig.DATABASE_CONFIG, coll.getNamespace().getDatabaseName());
+              put(MongoSourceConfig.COLLECTION_CONFIG, coll.getNamespace().getCollectionName());
+              put(MongoSourceConfig.FULL_DOCUMENT_CONFIG, "updateLookup");
+              put(MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG, "5");
+              put(MongoSourceConfig.ERRORS_TOLERANCE_CONFIG, ErrorTolerance.ALL.value());
+            }
+          };
+
+      task.start(cfg);
+
+      insertMany(rangeClosed(1, 5), coll);
+      List<SourceRecord> poll = getNextResults(task);
+      assertEquals(5, poll.size());
+
+      // Poison the change stream
+      coll.updateOne(new Document("_id", 3), Updates.set("y", new byte[(1024 * 1024 * 16) - 30]));
+      task.poll(); // Use poll directly as no results are expected.
+
+      // Insert some new data and confirm new events are available post change stream restart
+      insertMany(range(10, 15), coll);
+      poll = getNextResults(task);
+
+      assertEquals(5, poll.size());
+      assertTrue(
+          task.logCapture.getEvents().stream()
+              .filter(e -> e.getLevel().equals(Level.WARN))
+              .anyMatch(
+                  e ->
+                      e.getMessage()
+                          .toString()
+                          .startsWith(
+                              "Failed to resume change stream: Query failed with error code 10334")));
+    }
+  }
+
   private void assertSourceRecordValues(
       final List<? extends ChangeStreamOperation> expectedChangeStreamOperations,
       final List<SourceRecord> allSourceRecords,
diff --git a/src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java b/src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java
@@ -32,15 +32,18 @@
 import static com.mongodb.kafka.connect.util.ConfigHelper.getMongoDriverInformation;
 import static com.mongodb.kafka.connect.util.ServerApiConfig.setServerApi;
 import static java.lang.String.format;
+import static java.util.Arrays.asList;
 import static java.util.Collections.singletonMap;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 
@@ -63,6 +66,7 @@
 import com.mongodb.ConnectionString;
 import com.mongodb.MongoClientSettings;
 import com.mongodb.MongoCommandException;
+import com.mongodb.MongoException;
 import com.mongodb.client.ChangeStreamIterable;
 import com.mongodb.client.MongoChangeStreamCursor;
 import com.mongodb.client.MongoClient;
@@ -119,12 +123,24 @@ public final class MongoSourceTask extends SourceTask {
   private static final int NAMESPACE_NOT_FOUND_ERROR = 26;
   private static final int ILLEGAL_OPERATION_ERROR = 20;
   private static final int INVALIDATED_RESUME_TOKEN_ERROR = 260;
+  private static final int CHANGE_STREAM_FATAL_ERROR = 280;
+  private static final int CHANGE_STREAM_HISTORY_LOST = 286;
+  private static final int BSON_OBJECT_TOO_LARGE = 10334;
+  private static final Set<Integer> INVALID_CHANGE_STREAM_ERRORS =
+      new HashSet<>(
+          asList(
+              INVALIDATED_RESUME_TOKEN_ERROR,
+              CHANGE_STREAM_FATAL_ERROR,
+              CHANGE_STREAM_HISTORY_LOST,
+              BSON_OBJECT_TOO_LARGE));
   private static final int UNKNOWN_FIELD_ERROR = 40415;
   private static final int FAILED_TO_PARSE_ERROR = 9;
   private static final String RESUME_TOKEN = "resume token";
+  private static final String RESUME_POINT = "resume point";
   private static final String NOT_FOUND = "not found";
   private static final String DOES_NOT_EXIST = "does not exist";
   private static final String INVALID_RESUME_TOKEN = "invalid resume token";
+  private static final String NO_LONGER_IN_THE_OPLOG = "no longer be in the oplog";
 
   private final Time time;
   private final AtomicBoolean isRunning = new AtomicBoolean();
@@ -366,6 +382,28 @@ MongoChangeStreamCursor<? extends BsonDocument> createCursor(
     return tryCreateCursor(sourceConfig, mongoClient, getResumeToken(sourceConfig));
   }
 
+  private MongoChangeStreamCursor<? extends BsonDocument> tryRecreateCursor(
+      final MongoException e) {
+    int errorCode =
+        e instanceof MongoCommandException
+            ? ((MongoCommandException) e).getErrorCode()
+            : e.getCode();
+    String errorMessage =
+        e instanceof MongoCommandException
+            ? ((MongoCommandException) e).getErrorMessage()
+            : e.getMessage();
+    LOGGER.warn(
+        "Failed to resume change stream: {} {}\n"
+            + "===================================================================================\n"
+            + "When the resume token is no longer available there is the potential for data loss.\n\n"
+            + "Restarting the change stream with no resume token because `errors.tolerance=all`.\n"
+            + "===================================================================================\n",
+        errorMessage,
+        errorCode);
+    invalidatedCursor = true;
+    return tryCreateCursor(sourceConfig, mongoClient, null);
+  }
+
   private MongoChangeStreamCursor<? extends BsonDocument> tryCreateCursor(
       final MongoSourceConfig sourceConfig,
       final MongoClient mongoClient,
@@ -394,17 +432,8 @@ private MongoChangeStreamCursor<? extends BsonDocument> tryCreateCursor(
         } else if (doesNotSupportsStartAfter(e)) {
           supportsStartAfter = false;
           return tryCreateCursor(sourceConfig, mongoClient, resumeToken);
-        } else if (sourceConfig.tolerateErrors() && resumeTokenNotFound(e)) {
-          LOGGER.warn(
-              "Failed to resume change stream: {} {}\n"
-                  + "===================================================================================\n"
-                  + "When the resume token is no longer available there is the potential for data loss.\n\n"
-                  + "Restarting the change stream with no resume token because `errors.tolerance=all`.\n"
-                  + "===================================================================================\n",
-              e.getErrorMessage(),
-              e.getErrorCode());
-          invalidatedCursor = true;
-          return tryCreateCursor(sourceConfig, mongoClient, null);
+        } else if (sourceConfig.tolerateErrors() && changeStreamNotValid(e)) {
+          return tryRecreateCursor(e);
         }
       }
       if (e.getErrorCode() == NAMESPACE_NOT_FOUND_ERROR) {
@@ -438,7 +467,7 @@ private MongoChangeStreamCursor<? extends BsonDocument> tryCreateCursor(
                 + "=====================================================================================\n",
             e.getErrorMessage(),
             e.getErrorCode());
-        if (resumeTokenNotFound(e)) {
+        if (changeStreamNotValid(e)) {
           throw new ConnectException(
               "ResumeToken not found. Cannot create a change stream cursor", e);
         }
@@ -456,12 +485,19 @@ private boolean invalidatedResumeToken(final MongoCommandException e) {
     return e.getErrorCode() == INVALIDATED_RESUME_TOKEN_ERROR;
   }
 
-  private boolean resumeTokenNotFound(final MongoCommandException e) {
-    String errorMessage = e.getErrorMessage().toLowerCase(Locale.ROOT);
-    return errorMessage.contains(RESUME_TOKEN)
+  private boolean changeStreamNotValid(final MongoException e) {
+    if (INVALID_CHANGE_STREAM_ERRORS.contains(e.getCode())) {
+      return true;
+    }
+    String errorMessage =
+        e instanceof MongoCommandException
+            ? ((MongoCommandException) e).getErrorMessage().toLowerCase(Locale.ROOT)
+            : e.getMessage().toLowerCase(Locale.ROOT);
+    return (errorMessage.contains(RESUME_TOKEN) || errorMessage.contains(RESUME_POINT))
         && (errorMessage.contains(NOT_FOUND)
             || errorMessage.contains(DOES_NOT_EXIST)
-            || errorMessage.contains(INVALID_RESUME_TOKEN));
+            || errorMessage.contains(INVALID_RESUME_TOKEN)
+            || errorMessage.contains(NO_LONGER_IN_THE_OPLOG));
   }
 
   Map<String, Object> createPartitionMap(final MongoSourceConfig sourceConfig) {
@@ -588,26 +624,38 @@ private Optional<BsonDocument> getNextDocument() {
           next = cursor != null ? cursor.tryNext() : null;
         }
         return Optional.ofNullable(next);
-      } catch (Exception e) {
-        if (cursor != null) {
-          try {
-            cursor.close();
-          } catch (Exception e1) {
-            // ignore
+      } catch (MongoException e) {
+        closeCursor();
+        if (isRunning.get()) {
+          if (sourceConfig.tolerateErrors() && changeStreamNotValid(e)) {
+            cursor = tryRecreateCursor(e);
+          } else {
+            LOGGER.info(
+                "An exception occurred when trying to get the next item from the Change Stream", e);
           }
-          cursor = null;
         }
+        return Optional.empty();
+      } catch (Exception e) {
+        closeCursor();
         if (isRunning.get()) {
-          LOGGER.info(
-              "An exception occurred when trying to get the next item from the Change Stream: {}",
-              e.getMessage());
+          throw new ConnectException("Unexpected error: " + e.getMessage(), e);
         }
-        return Optional.empty();
       }
     }
     return Optional.empty();
   }
 
+  private void closeCursor() {
+    if (cursor != null) {
+      try {
+        cursor.close();
+      } catch (Exception e1) {
+        // ignore
+      }
+      cursor = null;
+    }
+  }
+
   private void invalidateCursorAndReinitialize() {
     invalidatedCursor = true;
     cursor.close();
