Ensure resume token missing errors respect errors.tolerance

Throw an exception if there is no errors tolerance and
the resume token is missing.

KAFKA-105

Author: rozza

src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest.java

@@ -45,6 +45,7 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
@@ -402,10 +403,8 @@ void testSourceCanUseCustomOffsetPartitionNames() {
       when(offsetStorageReader.offset(singletonMap("ns", "oldPartitionName")))
           .thenReturn(INVALID_OFFSET);
       task.initialize(context);
-      task.start(cfg);
 
-      assertNull(task.poll());
-      task.stop();
+      assertThrows(ConnectException.class, () -> task.start(cfg));
 
       assertTrue(
           task.logCapture.getEvents().stream()

src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java

@@ -391,7 +391,7 @@ private MongoChangeStreamCursor<? extends BsonDocument> tryCreateCursor(
         } else if (doesNotSupportsStartAfter(e)) {
           supportsStartAfter = false;
           return tryCreateCursor(sourceConfig, mongoClient, resumeToken);
-        } else if (resumeTokenNotFound(e)) {
+        } else if (sourceConfig.tolerateErrors() && resumeTokenNotFound(e)) {
           LOGGER.warn(
               "Failed to resume change stream: {} {}\n"
                   + "===================================================================================\n"
@@ -422,6 +422,10 @@ private MongoChangeStreamCursor<? extends BsonDocument> tryCreateCursor(
                 + "=====================================================================================\n",
             e.getErrorMessage(),
             e.getErrorCode());
+        if (resumeTokenNotFound(e)) {
+          throw new ConnectException(
+              "ResumeToken not found. Cannot create a change stream cursor", e);
+        }
       }
       return null;
     }
@@ -437,12 +441,8 @@ private boolean invalidatedResumeToken(final MongoCommandException e) {
   }
 
   private boolean resumeTokenNotFound(final MongoCommandException e) {
-    if (!sourceConfig.tolerateErrors()) {
-      return false;
-    }
     String errorMessage = e.getErrorMessage().toLowerCase(Locale.ROOT);
-    return sourceConfig.tolerateErrors()
-        && errorMessage.contains(RESUME_TOKEN)
+    return errorMessage.contains(RESUME_TOKEN)
         && (errorMessage.contains(NOT_FOUND)
             || errorMessage.contains(DOES_NOT_EXIST)
             || errorMessage.contains(INVALID_RESUME_TOKEN));