Improve error logging and handling (#150)

When looping a cursor fails log the error at an
a error log level.

If errors aren't tolerated then throw an exception instead of
continuing to loop.

KAFKA-396

Author: rozza

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

@@ -31,6 +31,7 @@
 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.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertIterableEquals;
@@ -53,7 +54,6 @@
 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;
@@ -483,6 +483,7 @@ void testSourceCanUseCustomOffsetPartitionNames() {
             {
               put(MongoSourceConfig.DATABASE_CONFIG, coll.getNamespace().getDatabaseName());
               put(MongoSourceConfig.COLLECTION_CONFIG, coll.getNamespace().getCollectionName());
+              put(MongoSourceConfig.OVERRIDE_ERRORS_TOLERANCE_CONFIG, "all");
               put(MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG, "50");
               put(MongoSourceConfig.POLL_AWAIT_TIME_MS_CONFIG, "5000");
               put(MongoSourceConfig.OFFSET_PARTITION_NAME_CONFIG, "oldPartitionName");
@@ -494,8 +495,7 @@ void testSourceCanUseCustomOffsetPartitionNames() {
           .thenReturn(INVALID_OFFSET);
       task.initialize(context);
 
-      assertThrows(
-          ConnectException.class,
+      assertDoesNotThrow(
           () -> {
             task.start(cfg);
             getNextBatch(task);

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

@@ -599,11 +599,18 @@ private List<BsonDocument> getNextBatch() {
     } catch (MongoException e) {
       closeCursor();
       if (isRunning) {
-        if (sourceConfig.tolerateErrors() && changeStreamNotValid(e)) {
-          cursor = tryRecreateCursor(e);
+        if (sourceConfig.tolerateErrors()) {
+          if (changeStreamNotValid(e)) {
+            cursor = tryRecreateCursor(e);
+          } else {
+            LOGGER.error(
+                "An exception occurred when trying to get the next item from the Change Stream", e);
+          }
         } else {
-          LOGGER.info(
-              "An exception occurred when trying to get the next item from the Change Stream", e);
+          throw new ConnectException(
+              "An exception occurred when trying to get the next item from the Change Stream: "
+                  + e.getMessage(),
+              e);
         }
       }
     } catch (Exception e) {