Problem while using resumeToken for Azure CosmosDB

Question

Problem while using resumeToken for Azure CosmosDB

Leela Sankar Pinjala 0

Hi, I am using CosmosDB (Mongo API) change streams to update cache. I am persisting resumeToken, and using it when my app restart. I noticed that whenever app restarted and open stream with resumeToken, we are getting high number of old messages which are already processed. Some time days old messages.

ShaktiSingh-MSFT 16,276 Reputation points

2023-04-24T08:47:50.0633333+00:00

Hi Leela Sankar Pinjala •,

Welcome to Microsoft Q&A forum and thanks for using Azure Services.

As I understand, you are using Change Streams in Cosmos DB Mongo DB API to update cache and getting old messages.

Could you please share the sample code snippet used for this purpose?

Leela Sankar Pinjala 0

Hi Here is the code snippet. We use spring-data-mongodb

// change stream request builder
ChangeStreamRequest.ChangeStreamRequestBuilder<T> changeStreamRequestBuilder = ChangeStreamRequest.builder()
.collection(changeStreamDefinition.getCollectionName())
.filter(
Aggregation.newAggregation(
Aggregation.match(Criteria.where("operationType").in("insert", "update", "replace")),
Aggregation.project("_id", "ns", "documentKey", "fullDocument")
)
).fullDocumentLookup("updateLookup"));
// resume token
ResumeTokenManager defaultResumeTokenManager =
DefaultResumeTokenManagerFactory.create(changeStreamDefinition, mongoTemplate);
BsonDocument resumeToken = defaultResumeTokenManager.getResumeToken();
if( resumeToken != null ) {
 changeStreamRequestBuilder.resumeAfter(resumeToken);
}

// register
ChangeStreamRequest request = changeStreamRequestBuilder.build();

Subscription subscription = messageListenerContainer.register(changeStreamRequestBuilder.build(),        changeStreamDefinition.getType(), changeStreamDefinition.getErrorHandler());
// code to persist resume token
public void persistResumeToken(BsonDocument resumeToken) {  
long currentTime = System.currentTimeMillis();    

Long lastUpdatedTime = lastUpdateTime.putIfAbsent(streamName, currentTime); 
   
if( lastUpdatedTime != null && (currentTime - lastUpdatedTime) >= 10000)  {        
long modifiedCount = streamStatusRepository.updateResumeToken(streamName, resumeToken.toJson());        this.lastUpdateTime.put(streamName, currentTime);        

if( modifiedCount > 0) {            
logger.info("ResumeToken({}) updated for stream [{}].", resumeToken.toJson(), streamName);        }    
}
}

Leela Sankar Pinjala 0 Reputation points

2023-04-24T12:40:52.65+00:00

Updated code, please let me know if you need any further info
Leela Sankar Pinjala 0 Reputation points

2023-04-24T12:44:20.87+00:00

Every 10 sec, we persist resumeToken into DB. When Pod restarted we expect at most 10sec changes reprocessed. However we see very high number of changes whenever pod restarted
ShaktiSingh-MSFT 16,276 Reputation points

2023-04-25T04:14:18.19+00:00

Hi Leela Sankar Pinjala •,

Thanks for sharing the details.

I am checking on this and will let you know the status soon.

Appreciate your patience.
Leela Sankar Pinjala 0 Reputation points

2023-04-26T05:44:40.5233333+00:00

Hi @ShaktiSingh-MSFT Wondering if you could spot anything. I am also going through docs. It appears CosmosDB maintain one token per physical partition. But app only persisting single token. Could this be an issue?
ShaktiSingh-MSFT 16,276 Reputation points

2023-04-27T05:59:22.3133333+00:00

Hi Leela Sankar Pinjala •,

We appreciate your patience.

I am yet to get reply from internal team on this. Will keep you posted. Thanks.
ShaktiSingh-MSFT 16,276 Reputation points

2023-05-02T09:31:33.5633333+00:00

Hi Leela Sankar Pinjala •,

Thanks for your patience.

As discussed with the internal team, it is suggested that you should raise support case for this issue for deeper investigation and in case if you don't have a support plan, do let us know here so that we can check on other options to unblock you. Thanks

Your answer

ShaktiSingh-MSFT 16,276 Reputation points

2023-04-24T08:47:50.0633333+00:00

Hi Leela Sankar Pinjala •,

Welcome to Microsoft Q&A forum and thanks for using Azure Services.

As I understand, you are using Change Streams in Cosmos DB Mongo DB API to update cache and getting old messages.

Could you please share the sample code snippet used for this purpose?
Leela Sankar Pinjala 0 Reputation points

2023-04-24T10:11:10.34+00:00

Hi Here is the code snippet. We use spring-data-mongodb

// change stream request builder ChangeStreamRequest.ChangeStreamRequestBuilder<T> changeStreamRequestBuilder = ChangeStreamRequest.builder() .collection(changeStreamDefinition.getCollectionName()) .filter( Aggregation.newAggregation( Aggregation.match(Criteria.where("operationType").in("insert", "update", "replace")), Aggregation.project("_id", "ns", "documentKey", "fullDocument") ) ).fullDocumentLookup("updateLookup")); // resume token ResumeTokenManager defaultResumeTokenManager = DefaultResumeTokenManagerFactory.create(changeStreamDefinition, mongoTemplate); BsonDocument resumeToken = defaultResumeTokenManager.getResumeToken(); if( resumeToken != null ) { changeStreamRequestBuilder.resumeAfter(resumeToken); } // register ChangeStreamRequest request = changeStreamRequestBuilder.build(); Subscription subscription = messageListenerContainer.register(changeStreamRequestBuilder.build(), changeStreamDefinition.getType(), changeStreamDefinition.getErrorHandler()); // code to persist resume token public void persistResumeToken(BsonDocument resumeToken) { long currentTime = System.currentTimeMillis(); Long lastUpdatedTime = lastUpdateTime.putIfAbsent(streamName, currentTime); if( lastUpdatedTime != null && (currentTime - lastUpdatedTime) >= 10000) { long modifiedCount = streamStatusRepository.updateResumeToken(streamName, resumeToken.toJson()); this.lastUpdateTime.put(streamName, currentTime); if( modifiedCount > 0) { logger.info("ResumeToken({}) updated for stream [{}].", resumeToken.toJson(), streamName); } } }
Leela Sankar Pinjala 0 Reputation points

2023-04-24T12:40:52.65+00:00

Updated code, please let me know if you need any further info
Leela Sankar Pinjala 0 Reputation points

2023-04-24T12:44:20.87+00:00

Every 10 sec, we persist resumeToken into DB. When Pod restarted we expect at most 10sec changes reprocessed. However we see very high number of changes whenever pod restarted
ShaktiSingh-MSFT 16,276 Reputation points

2023-04-25T04:14:18.19+00:00

Hi Leela Sankar Pinjala •,

Thanks for sharing the details.

I am checking on this and will let you know the status soon.

Appreciate your patience.
Leela Sankar Pinjala 0 Reputation points

2023-04-26T05:44:40.5233333+00:00

Hi @ShaktiSingh-MSFT Wondering if you could spot anything. I am also going through docs. It appears CosmosDB maintain one token per physical partition. But app only persisting single token. Could this be an issue?
ShaktiSingh-MSFT 16,276 Reputation points

2023-04-27T05:59:22.3133333+00:00

Hi Leela Sankar Pinjala •,

We appreciate your patience.

I am yet to get reply from internal team on this. Will keep you posted. Thanks.
ShaktiSingh-MSFT 16,276 Reputation points

2023-05-02T09:31:33.5633333+00:00

Hi Leela Sankar Pinjala •,

Thanks for your patience.

As discussed with the internal team, it is suggested that you should raise support case for this issue for deeper investigation and in case if you don't have a support plan, do let us know here so that we can check on other options to unblock you. Thanks

Share via

Problem while using resumeToken for Azure CosmosDB

Your answer