Ensure filtered change streams survive invalidate events

KAFKA-83

Author: rozza

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

@@ -309,6 +309,30 @@ void testSourceCanHandleNonExistentCollectionAndSurviveDropping() throws Interru
         assertProduced(201, coll);
     }
 
+    @Test
+    @DisplayName("Ensure source can handle non existent collection and survive dropping")
+    void testSourceCanSurviveDroppingWithPipelineWatchingInsertsOnly() throws InterruptedException {
+        MongoCollection<Document> coll = getDatabaseWithPostfix().getCollection("coll");
+
+        Properties sourceProperties = new Properties();
+        sourceProperties.put(MongoSourceConfig.DATABASE_CONFIG, coll.getNamespace().getDatabaseName());
+        sourceProperties.put(MongoSourceConfig.COLLECTION_CONFIG, coll.getNamespace().getCollectionName());
+        sourceProperties.put(MongoSourceConfig.PIPELINE_CONFIG, "[{\"$match\": {\"operationType\": \"insert\"}}]");
+        addSourceConnector(sourceProperties);
+
+        Thread.sleep(5000);
+        assertProduced(0, coll);
+
+        insertMany(rangeClosed(1, 50), coll);
+        assertProduced(50, coll);
+
+        coll.drop();
+        Thread.sleep(5000);
+
+        insertMany(rangeClosed(1, 50), coll);
+        assertProduced(100, coll);
+    }
+
     @Test
     @DisplayName("Ensure source loads data from collection and outputs documents only")
     void testSourceLoadsDataFromCollectionDocumentOnly() {

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

@@ -46,7 +46,6 @@
 
 import org.bson.BsonDocument;
 import org.bson.BsonDocumentWrapper;
-import org.bson.BsonString;
 import org.bson.Document;
 
 import com.mongodb.MongoClientSettings;
@@ -90,7 +89,6 @@
  */
 public class MongoSourceTask extends SourceTask {
     private static final Logger LOGGER = LoggerFactory.getLogger(MongoSourceTask.class);
-    private static final String INVALIDATE = "invalidate";
 
     private final Time time;
     private final AtomicBoolean isRunning = new AtomicBoolean();
@@ -189,12 +187,7 @@ public List<SourceRecord> poll() {
                             Schema.STRING_SCHEMA, json));
                 });
 
-                // If the cursor is invalidated add the record and return calls
-                if (changeStreamDocument.getString("operationType", new BsonString("")).getValue().equalsIgnoreCase(INVALIDATE)) {
-                    LOGGER.info("Cursor has been invalidated.");
-                    cursor = null;
-                    return sourceRecords;
-                } else if (sourceRecords.size() == maxBatchSize) {
+                if (sourceRecords.size() == maxBatchSize) {
                     LOGGER.debug("Reached max batch size: {}, returning records", maxBatchSize);
                     return sourceRecords;
                 }
@@ -317,7 +310,12 @@ private Optional<BsonDocument> getNextDocument() {
         }
 
         try {
-            return Optional.ofNullable(cursor.tryNext());
+            BsonDocument next = cursor.tryNext();
+            if (next == null && cursor.getServerCursor() == null) {
+                cursor.close();
+                cursor = null;
+            }
+            return Optional.ofNullable(next);
         } catch (Exception e) {
             cursor.close();
             cursor = null;