debezium-connector-mongodb fails when using cursor.pipeline [DBZ-9274]

Migrated from [DBZ-9274](https://issues.redhat.com/browse/DBZ-9274)

h2. What Debezium connector do you use and what version?

mongodb connector, 2.3.3.Final, via spring cloud stream debezium-source-kafka docker image (version 4.0.0)

[https://hub.docker.com/r/springcloudstream/debezium-source-kafka]
h2. What is the connector configuration?

Using cursor.pipeline that specifies a $changeStream step seems to trigger the error, works otherwise.

database.include.list and collection.include.list are also correctly populated, not specifying the admin db anywhere.
{code:java}
...
cursor.pipeline = [{ $changeStream: { startAtOperationTime: { $timestamp: { t: 1753269600, i: 1 } } } }]
database.include.list = "my_db"
collection.include.list = "my_db.coll1"{code}
 

 
h2. What is the captured database version and mode of deployment?

Mongodb 5.1, on premise
h2. What behavior do you expect?

I would expect the $changeStream step to be applied to the events extracted from the watched collections
h2. What behavior do you see?

Looks like debezium tries to register the pipeline on the local admin db too (?)
{code:java}
Caused by: com.mongodb.MongoCommandException: Command failed with error 73 (InvalidNamespace): '$changeStream may not be opened on the internal admin database' on server *****:27017. The full response is {"ok": 0.0, "errmsg": "$changeStream may not be opened on the internal admin database", "code": 73, "codeName": "InvalidNamespace", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1753341831, "i": 99}}, "signature": {"hash": {"$binary": {"base64": "fQmDLsfOaoDhnO97eRCD9BXuRu0=", "subType": "00"}}, "keyId": 7495709948706291772}}, "operationTime": {"$timestamp": {"t": 1753341831, "i": 99}}}
  at com.mongodb.internal.connection.ProtocolHelper.getCommandFailureException(ProtocolHelper.java:205) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.InternalStreamConnection.receiveCommandMessageResponse(InternalStreamConnection.java:443) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:365) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.UsageTrackingInternalConnection.sendAndReceive(UsageTrackingInternalConnection.java:114) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection.sendAndReceive(DefaultConnectionPool.java:643) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.CommandProtocolImpl.execute(CommandProtocolImpl.java:73) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:204) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:122) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:87) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:76) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.connection.DefaultServer$OperationCountTrackingConnection.command(DefaultServer.java:288) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.CommandOperationHelper.createReadCommandAndExecute(CommandOperationHelper.java:239) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.CommandOperationHelper.lambda$executeRetryableRead$4(CommandOperationHelper.java:220) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.OperationHelper.lambda$withSourceAndConnection$0(OperationHelper.java:358) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.OperationHelper.withSuppliedResource(OperationHelper.java:383) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.OperationHelper.lambda$withSourceAndConnection$1(OperationHelper.java:357) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.OperationHelper.withSuppliedResource(OperationHelper.java:383) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.OperationHelper.withSourceAndConnection(OperationHelper.java:356) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.CommandOperationHelper.lambda$executeRetryableRead$5(CommandOperationHelper.java:218) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.async.function.RetryingSyncSupplier.get(RetryingSyncSupplier.java:67) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.CommandOperationHelper.executeRetryableRead(CommandOperationHelper.java:223) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.CommandOperationHelper.executeRetryableRead(CommandOperationHelper.java:204) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.AggregateOperationImpl.execute(AggregateOperationImpl.java:191) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.ChangeStreamOperation.lambda$execute$0(ChangeStreamOperation.java:187) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.OperationHelper.withReadConnectionSource(OperationHelper.java:321) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.ChangeStreamOperation.execute(ChangeStreamOperation.java:185) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.internal.operation.ChangeStreamOperation.execute(ChangeStreamOperation.java:55) ~[mongodb-driver-core-4.9.1.jar:na]
  at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.execute(MongoClientDelegate.java:185) ~[mongodb-driver-sync-4.9.1.jar:na]
  at com.mongodb.client.internal.ChangeStreamIterableImpl.execute(ChangeStreamIterableImpl.java:212) ~[mongodb-driver-sync-4.9.1.jar:na]
  at com.mongodb.client.internal.ChangeStreamIterableImpl.cursor(ChangeStreamIterableImpl.java:187) ~[mongodb-driver-sync-4.9.1.jar:na]
  at io.debezium.connector.mongodb.MongoDbStreamingChangeEventSource.readChangeStream(MongoDbStreamingChangeEventSource.java:170) ~[debezium-connector-mongodb-2.3.3.Final.jar:2.3.3.Final]
  at io.debezium.connector.mongodb.MongoDbStreamingChangeEventSource.lambda$streamChangesForReplicaSet$0(MongoDbStreamingChangeEventSource.java:99) ~[debezium-connector-mongodb-2.3.3.Final.jar:2.3.3.Final]
  at io.debezium.connector.mongodb.connection.MongoDbConnection.lambda$execute$3(MongoDbConnection.java:104) ~[debezium-connector-mongodb-2.3.3.Final.jar:2.3.3.Final]
  at io.debezium.connector.mongodb.connection.MongoDbConnection.execute(MongoDbConnection.java:120) ~[debezium-connector-mongodb-2.3.3.Final.jar:2.3.3.Final]
  ... 12 common frames omitted  {code}
 

 
h2. Do you see the same behaviour using the latest released Debezium version?

Could not verify
h2. Do you have the connector logs, ideally from start till finish?

I can provide the full log of a failing run, prior to obscuring some details, but the log I pasted above seems to be the only relevant entry

Start at MongoDbStreamingChangeEventSource.readChangeStream and trace how cursor.pipeline is applied through MongoDbConnection. Reproduce with the supplied MongoDB 5.1 configuration, then verify that the $changeStream stage is used for the watched collections without attempting to open it on the internal admin database.

Written by the indexing model from the issue text.