Resume token issue after reconnection [DATAMONGO-2537]

sydneyhenrard opened DATAMONGO-2537 and commented

I have 2 collections data and nocs. The application is started without a resume token.

Change Stream started for data collection with resume token null

One document is written in data collection

message: Document`_id=5eb515fb8cd21a4628dbe06e, test=new doc`

Check the oplog

rs.printReplicationInfo()
configured oplog size:   990MB
log length start to end: 506secs (0.14hrs)
oplog first event time:  Fri May 08 2020 11:04:04 GMT+0200 (W. Europe Standard Time)
oplog last event time:   Fri May 08 2020 11:12:30 GMT+0200 (W. Europe Standard Time)
now:                     Fri May 08 2020 11:12:30 GMT+0200 (W. Europe Standard Time)

Grow the oplog by writing into nocs collection so the resume token becomes invalid

rs.printReplicationInfo()
configured oplog size:   990MB
log length start to end: 273secs (0.08hrs)
oplog first event time:  Fri May 08 2020 11:14:07 GMT+0200 (W. Europe Standard Time)
oplog last event time:   Fri May 08 2020 11:18:40 GMT+0200 (W. Europe Standard Time)
now:                     Fri May 08 2020 11:18:44 GMT+0200 (W. Europe Standard Time)

Kill the cluster to simulate a network issue

2020-05-08 11:25:23.120  INFO 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Wait for 10 seconds
2020-05-08 11:25:33.121 ERROR 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Error in change stream listener; will reconnect: Timed out after 30000 ms while waiting for a server that matches ReadPreferenceServerSelector{readPreference=primary}. Client view of cluster state is {type=UNKNOWN, servers=[{address=localhost:9051, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketOpenException: Exception opening socket}, caused by {java.net.ConnectException: Connection refused: connect}}]; nested exception is com.mongodb.MongoTimeoutException: Timed out after 30000 ms while waiting for a server that matches ReadPreferenceServerSelector{readPreference=primary}. Client view of cluster state is {type=UNKNOWN, servers=[{address=localhost:9051, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketOpenException: Exception opening socket}, caused by {java.net.ConnectException: Connection refused: connect}}]
2020-05-08 11:25:33.121  INFO 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Reconnecting

Restart the cluster

2020-05-08 11:29:06.973  INFO 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Wait for 10 seconds
2020-05-08 11:29:16.973 ERROR 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Error in change stream listener; will reconnect: Timed out after 30000 ms while waiting to connect. Client view of cluster state is {type=UNKNOWN, servers=[{address=localhost:9051, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketOpenException: Exception opening socket}, caused by {java.net.ConnectException: Connection refused: connect}}]; nested exception is com.mongodb.MongoTimeoutException: Timed out after 30000 ms while waiting to connect. Client view of cluster state is {type=UNKNOWN, servers=[{address=localhost:9051, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketOpenException: Exception opening socket}, caused by {java.net.ConnectException: Connection refused: connect}}]
2020-05-08 11:29:16.973  INFO 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Reconnecting
2020-05-08 11:29:16.974  INFO 13852 --- [pool-1-thread-1] org.mongodb.driver.cluster               : Cluster description not yet available. Waiting for 30000 ms before timing out
2020-05-08 11:29:16.978  INFO 13852 --- [-localhost:9051] org.mongodb.driver.connection            : Opened connection [connectionId{localValue:73, serverValue:9}] to localhost:9051
2020-05-08 11:29:16.979  INFO 13852 --- [-localhost:9051] org.mongodb.driver.cluster               : Monitor thread successfully connected to server with description ServerDescription{address=localhost:9051, type=REPLICA_SET_SECONDARY, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 9]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=684000, setName='rs', canonicalAddress=mongo4:9051, hosts=[mongo6:9053, mongo5:9052, mongo4:9051], passives=[], arbiters=[], primary='null', tagSet=TagSet{[]}, electionId=null, setVersion=1, lastWriteDate=Fri May 08 11:27:55 CEST 2020, lastUpdateTimeNanos=229268558630200}
2020-05-08 11:29:16.981  INFO 13852 --- [pool-1-thread-1] org.mongodb.driver.connection            : Opened connection [connectionId{localValue:74, serverValue:10}] to localhost:9051
2020-05-08 11:29:26.981  INFO 13852 --- [-localhost:9051] org.mongodb.driver.cluster               : Monitor thread successfully connected to server with description ServerDescription{address=localhost:9051, type=REPLICA_SET_SECONDARY, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 9]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=770160, setName='rs', canonicalAddress=mongo4:9051, hosts=[mongo6:9053, mongo5:9052, mongo4:9051], passives=[], arbiters=[], primary='mongo5:9052', tagSet=TagSet{[]}, electionId=null, setVersion=1, lastWriteDate=Fri May 08 11:29:24 CEST 2020, lastUpdateTimeNanos=229278560870700}
2020-05-08 11:29:36.983  INFO 13852 --- [-localhost:9051] org.mongodb.driver.cluster               : Monitor thread successfully connected to server with description ServerDescription{address=localhost:9051, type=REPLICA_SET_PRIMARY, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 9]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=809308, setName='rs', canonicalAddress=mongo4:9051, hosts=[mongo6:9053, mongo5:9052, mongo4:9051], passives=[], arbiters=[], primary='mongo4:9051', tagSet=TagSet{[]}, electionId=7fffffff000000000000000e, setVersion=1, lastWriteDate=Fri May 08 11:29:35 CEST 2020, lastUpdateTimeNanos=229288562777200}

Write in the collection data

For some reason the error handler is triggered because the resume token is invalid, and tries to reconnect without the resume token but it means that the change stream has been lost.

It's like when the connection is up again, it takes the existing MessageListenerContainer and try to get a new subscription. It kind of makes sense because the behavior is the same when an application starts. Maybe the fix is to have a bigger oplog.

2020-05-08 11:31:06.419  INFO 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Wait for 10 seconds
2020-05-08 11:31:16.420 ERROR 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Reconnection without resume token

com.mongodb.MongoCommandException: Command failed with error 40585 (Location40585): 'resume of change stream was not possible, as the resume token was not found. {_data: BinData(0, "825EB526D60000000146645F696400645EB526D625C06C18241B78D8005A10040D318B64E9804CABB61ED8C31C1C3F2804")}' on server localhost:9051. The full response is {"operationTime": {"$timestamp": {"t": 1588930262, "i": 1}}, "ok": 0.0, "errmsg": "resume of change stream was not possible, as the resume token was not found. {_data: BinData(0, \"825EB526D60000000146645F696400645EB526D625C06C18241B78D8005A10040D318B64E9804CABB61ED8C31C1C3F2804\")}", "code": 40585, "codeName": "Location40585", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1588930262, "i": 1}}, "signature": {"hash": {"$binary": "AAAAAAAAAAAAAAAAAAAAAAAAAAA=", "$type": "00"}, "keyId": {"$numberLong": "0"}}}}

2020-05-08 11:31:16.421  INFO 13852 --- [pool-1-thread-1] c.e.m.RetryableMessageListenerContainer  : Reconnecting

Write in the collection data

message: Document`_id=5eb5291a25c06c18241b802e, test=reconnection without resume token`

I attached the implementation in case it's the cause.

---

Affects: 2.2.7 (Moore SR7)

Attachments:

• ChangeStreamService.java (3.57 kB)
• MongoChangeStreamApplication.java (356 bytes)
• ResumeToken.java (1019 bytes)
• ResumeTokenRepository.java (295 bytes)
• RetryableMessageListenerContainer.java (4.35 kB)