Improved Illegal Operation error messages (#62)

Provides a relevant error message when users use an illegal pipeline
operation.

KAFKA-174

Author: rozza

CHANGELOG.md

@@ -10,6 +10,7 @@
   - [KAFKA-205](https://jira.mongodb.org/browse/KAFKA-205) Updated Source connector to use RawBsonDocuments.
   - [KAFKA-201](https://jira.mongodb.org/browse/KAFKA-201) Improved copy.existing namespace handling.
   - [KAFKA-207](https://jira.mongodb.org/browse/KAFKA-207) Improved efficiency of heartbeats by making them tombstone messages.
+  - [KAFKA-174](https://jira.mongodb.org/browse/KAFKA-174) Improved error messages when using invalid pipeline operators.
 
 ### Bug Fixes
   - [KAFKA-195](https://jira.mongodb.org/browse/KAFKA-195) Fixed topics.regex sink validation issue for synthetic config property

src/integrationTest/java/com/mongodb/kafka/connect/MongoSourceConnectorIntegrationTest.java

@@ -308,6 +308,31 @@ void testSourceUsesHeartbeatsForOffsets() {
     }
   }
 
+  @Test
+  @DisplayName("Ensure Source provides friendly error messages for invalid pipelines")
+  void testSourceHasFriendlyErrorMessagesForInvalidPipelines() {
+    assumeTrue(isGreaterThanFourDotZero());
+    try (LogCapture logCapture = new LogCapture(Logger.getLogger(MongoSourceTask.class))) {
+      MongoCollection<Document> coll = getAndCreateCollection();
+
+      Properties sourceProperties = new Properties();
+      sourceProperties.put(
+          MongoSourceConfig.DATABASE_CONFIG, coll.getNamespace().getDatabaseName());
+      sourceProperties.put(
+          MongoSourceConfig.COLLECTION_CONFIG, coll.getNamespace().getCollectionName());
+      sourceProperties.put(PIPELINE_CONFIG, "[{'$group': {_id: 1 }}]");
+
+      addSourceConnector(sourceProperties);
+
+      boolean containsIllegalChangeStreamOperation =
+          logCapture.getEvents().stream()
+              .map(e -> e.getMessage().toString())
+              .anyMatch(e -> e.startsWith("Illegal $changeStream operation"));
+
+      assertTrue(containsIllegalChangeStreamOperation);
+    }
+  }
+
   public static class KeyValueDeserializer implements Deserializer<Integer> {
 
     static final JsonDeserializer JSON_DESERIALIZER = new JsonDeserializer();

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

@@ -117,6 +117,7 @@ public final class MongoSourceTask extends SourceTask {
   private static final String NS_KEY = "ns";
   private static final String FULL_DOCUMENT = "fullDocument";
   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 UNKNOWN_FIELD_ERROR = 40415;
   private static final int FAILED_TO_PARSE_ERROR = 9;
@@ -404,6 +405,19 @@ private MongoChangeStreamCursor<? extends BsonDocument> tryCreateCursor(
       }
       if (e.getErrorCode() == NAMESPACE_NOT_FOUND_ERROR) {
         LOGGER.info("Namespace not found cursor closed.");
+      } else if (e.getErrorCode() == ILLEGAL_OPERATION_ERROR) {
+        LOGGER.warn(
+            "Illegal $changeStream operation: {} {}\n\n"
+                + "=====================================================================================\n"
+                + "{}\n\n"
+                + "Please Note: Not all aggregation pipeline operations are suitable for modifying the\n"
+                + "change stream output. For more information, please see the official documentation:\n"
+                + "   https://docs.mongodb.com/manual/changeStreams/\n"
+                + "=====================================================================================\n",
+            e.getErrorMessage(),
+            e.getErrorCode(),
+            e.getErrorMessage());
+        throw new ConnectException("Illegal $changeStream operation", e);
       } else {
         LOGGER.warn(
             "Failed to resume change stream: {} {}\n\n"