SERVER-34705: Whole-DB or whole-cluster change streams may not provide a total ordering if resumed after a drop

3 files changed, 35 insertions, 5 deletions

diff --git a/jstests/sharding/change_stream_invalidation.js b/jstests/sharding/change_stream_invalidation.js
index ebfc59cd8c6..a50d8eb88ea 100644
--- a/jstests/sharding/change_stream_invalidation.js
+++ b/jstests/sharding/change_stream_invalidation.js
@@ -71,8 +71,6 @@
     assert.soon(() => changeStream.hasNext());
     next = changeStream.next();
     assert.eq(next.operationType, "insert");
-    // TODO SERVER-34705: Extract the shard key from the resume token and include in the documentKey
-    // for inserts.
     assert.eq(next.documentKey, {_id: 2});
 
     assert.soon(() => changeStream.hasNext());
@@ -93,7 +91,7 @@
     assert.soon(() => changeStream.hasNext());
     next = changeStream.next();
     assert.eq(next.operationType, "insert");
-    assert.eq(next.documentKey, {_id: 2});
+    assert.eq(next.documentKey, {shardKey: 2, _id: 2});
 
     assert.soon(() => changeStream.hasNext());
     next = changeStream.next();
diff --git a/jstests/sharding/change_streams_primary_shard_unaware.js b/jstests/sharding/change_streams_primary_shard_unaware.js
index f59775cacb1..6975678d18d 100644
--- a/jstests/sharding/change_streams_primary_shard_unaware.js
+++ b/jstests/sharding/change_streams_primary_shard_unaware.js
@@ -165,7 +165,7 @@
     cstMongos2.assertNextChangesEqual({
         cursor: cursorMongos2,
         expectedChanges: [{
-            documentKey: {_id: 2},
+            documentKey: {_id: 2, a: 2},
             fullDocument: {_id: 2, a: 2},
             ns: {db: mongos2DB.getName(), coll: mongos2Coll.getName()},
             operationType: "insert",
diff --git a/jstests/sharding/change_streams_whole_db.js b/jstests/sharding/change_streams_whole_db.js
index d43294e7772..bc7d559610a 100644
--- a/jstests/sharding/change_streams_whole_db.js
+++ b/jstests/sharding/change_streams_whole_db.js
@@ -5,6 +5,7 @@
     load('jstests/replsets/libs/two_phase_drops.js');  // For TwoPhaseDropCollectionTest.
     load('jstests/aggregation/extras/utils.js');       // For assertErrorCode().
     load('jstests/libs/change_stream_util.js');        // For ChangeStreamTest.
+    load("jstests/libs/collection_drop_recreate.js");  // For assertDropCollection.
 
     // For supportsMajorityReadConcern().
     load("jstests/multiVersion/libs/causal_consistency_helpers.js");
@@ -151,7 +152,38 @@
           operationType: "insert",
         },
     ];
-    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: expected});
+
+    const results = cst.assertNextChangesEqual({cursor: cursor, expectedChanges: expected});
+    // Store the resume token of the first insert to use after dropping the collection.
+    const resumeTokenBeforeDrop = results[0]._id;
+
+    // Write one more document to the collection that will be dropped, to be returned after
+    // resuming.
+    assert.writeOK(mongosCollShardedOnX.insert({_id: 4, x: 4}));
+
+    // Drop the collection, invalidating the open change stream.
+    assertDropCollection(mongosDB, mongosCollShardedOnX.getName());
+
+    // Resume the change stream from before the collection drop, and verify that the documentKey
+    // field contains the extracted shard key from the resume token.
+    cursor = cst.startWatchingChanges({
+        pipeline: [
+            {$changeStream: {resumeAfter: resumeTokenBeforeDrop}},
+            {$match: {"ns.coll": mongosCollShardedOnX.getName()}}
+        ],
+        collection: 1
+    });
+    cst.assertNextChangesEqual({
+        cursor: cursor,
+        expectedChanges: [
+            {
+              documentKey: {_id: 4, x: 4},
+              fullDocument: {_id: 4, x: 4},
+              ns: {db: mongosDB.getName(), coll: mongosCollShardedOnX.getName()},
+              operationType: "insert",
+            },
+        ]
+    });
 
     cst.cleanUp();