Improved the error messaging for the missing resume tokens in the source connector.

KAFKA-91

Author: rozza

CHANGELOG.md

@@ -17,6 +17,7 @@
   - [KAFKA-75](https://jira.mongodb.org/browse/KAFKA-75) Added specific configuration for the id strategies: `ProvidedInKeyStrategy` and `ProvidedInValueStrategy`.
     Added `document.id.strategy.partial.value.projection.type`, `document.id.strategy.partial.value.projection.list`,
     `document.id.strategy.partial.key.projection.type` and `document.id.strategy.partial.key.projection.list`.
+  - [KAFKA-91](https://jira.mongodb.org/browse/KAFKA-91) Improved the error messaging for the missing resume tokens in the source connector.
 
 ## 1.1.0
   - [KAFKA-45](https://jira.mongodb.org/browse/KAFKA-45) Allow the Sink connector to ignore unused source record key or value fields.

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

@@ -273,7 +273,24 @@ private MongoCursor<BsonDocument> tryCreateCursor(
           return tryCreateCursor(sourceConfig, mongoClient, resumeToken);
         }
       }
-      LOGGER.info("Failed to resume change stream: {} {}", e.getErrorMessage(), e.getErrorCode());
+      LOGGER.warn(
+          "Failed to resume change stream: {} {}\n"
+              + "=====================================================================================\n"
+              + "If the resume token is no longer available then there is the potential for data loss.\n"
+              + "Saved resume tokens are managed by Kafka and stored with the offset data.\n\n"
+              + "When running Connect in standalone mode offsets are configured using the:\n"
+              + "`offset.storage.file.filename` configuration.\n"
+              + "When running Connect in distributed mode the offsets are stored in a topic.\n\n"
+              + "Use the `kafka-consumer-groups.sh` tool with the `--reset-offsets` flag to reset\n"
+              + "offsets.\n\n"
+              + "Resetting the offset will allow for the connector to be resume from the latest resume\n"
+              + "token. Using `copy.existing=true` ensures that all data will be outputted by the\n"
+              + "connector but it will duplicate existing data.\n"
+              + "Future releases will support a configurable `errors.tolerance` level for the source\n"
+              + "connector and make use of the `postBatchResumeToken`.\n"
+              + "=====================================================================================\n",
+          e.getErrorMessage(),
+          e.getErrorCode());
       return null;
     }
   }