SERVER-54937 Removes DocumentSourceChangeStreamCloseCursor dependencies on invalidate events

1 files changed, 13 insertions, 0 deletions

diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp
index 90e2b822178..8827907d5a2 100644
--- a/src/mongo/s/query/cluster_aggregation_planner.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner.cpp
@@ -35,6 +35,7 @@
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/client/connpool.h"
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/pipeline/change_stream_invalidation_info.h"
 #include "mongo/db/pipeline/document_source_limit.h"
 #include "mongo/db/pipeline/document_source_skip.h"
 #include "mongo/db/pipeline/sharded_agg_helpers.h"
@@ -290,6 +291,18 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx,
             // error.
             cursorState = ClusterCursorManager::CursorState::Exhausted;
             break;
+        } catch (const ExceptionFor<ErrorCodes::ChangeStreamInvalidated>& ex) {
+            // This exception is thrown when a change-stream cursor is invalidated. Set the PBRT
+            // to the resume token of the invalidating event, and mark the cursor response as
+            // invalidated. We always expect to have ExtraInfo for this error code.
+            const auto extraInfo = ex.extraInfo<ChangeStreamInvalidationInfo>();
+            tassert(
+                5493706, "Missing ChangeStreamInvalidationInfo on exception", extraInfo != nullptr);
+
+            responseBuilder.setPostBatchResumeToken(extraInfo->getInvalidateResumeToken());
+            responseBuilder.setInvalidated();
+            cursorState = ClusterCursorManager::CursorState::Exhausted;
+            break;
         }
 
         // Check whether we have exhausted the pipeline's results.