You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "Michwo (via GitHub)" <gi...@apache.org> on 2026/04/17 12:03:22 UTC
[PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Michwo opened a new pull request, #22674:
URL: https://github.com/apache/camel/pull/22674
# Description
Adds option to configure the change streams resume token in change streams consumer endpoint (property `resumeToken`) and exposes the current resume token in exchange header (`MongoDBConstants.RESUME_TOKEN`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #22674:
URL: https://github.com/apache/camel/pull/22674
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Michwo (via GitHub)" <gi...@apache.org>.
Michwo closed pull request #22674: CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer)
URL: https://github.com/apache/camel/pull/22674
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4268134470
Please run a full mvn clean install -DskipTests from root folder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4279315409
Hi @Michwo I think that this feature can be implemented with Camel's [resume strategy (ResumeAware Consumer)]( Is it the resume documentation https://camel.apache.org/components/4.18.x/eips/resume-strategies.html). Using the resume strategy, the Camel route would look like:
```
from("mongodb:myDb?consumerType=changeStreams&database=mydb&collection=mycol")
.resumable(myStategy)
.process(this::process);
```
The resumeToken handling will be offloaded to the framework, and hopefully it will be easier to use.
The ResumeAware implementation can be done in another PR, no need to change the current implementation, but please open an issue about it.
fyi @orpiske
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #22674:
URL: https://github.com/apache/camel/pull/22674#discussion_r3464716913
##########
components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java:
##########
@@ -20,29 +20,67 @@
import java.util.concurrent.ExecutorService;
import org.apache.camel.Processor;
+import org.apache.camel.resume.ResumeAdapter;
+import org.apache.camel.resume.ResumeAware;
+import org.apache.camel.resume.ResumeStrategy;
+import org.apache.camel.spi.StateRepository;
import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.resume.ResumeStrategyHelper;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
import org.bson.BsonDocument;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static java.util.Collections.singletonList;
/**
* The MongoDb Change Streams consumer.
*/
-public class MongoDbChangeStreamsConsumer extends DefaultConsumer {
+public class MongoDbChangeStreamsConsumer extends DefaultConsumer implements ResumeAware<ResumeStrategy> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MongoDbChangeStreamsConsumer.class);
+ private static final String MONGODB_RESUME_PLACEHOLDER_ACTION = "mongodb-resume";
private final MongoDbEndpoint endpoint;
private ExecutorService executor;
private MongoDbChangeStreamsThread changeStreamsThread;
+ private ResumeStrategy resumeStrategy;
+ private boolean stopOffsetRepo;
+ private volatile BsonDocument startupResumeToken;
Review Comment:
The token repository is stopped here before the change streams thread is stopped (line 58). If the thread is mid-commit (`commitManager.commit()` → `repo.setState()`), the repo may already be shut down. Consider moving this block after the thread and executor shutdown to ensure the last token commit completes before the repo is closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4288281205
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `catalog/camel-catalog`
- `components/camel-mongodb`
- `dsl/camel-endpointdsl`
<details><summary>All tested modules (10 modules)</summary>
- Camel :: Catalog :: Camel Catalog
- Camel :: Endpoint DSL
- Camel :: JBang :: MCP
- Camel :: JBang :: Plugin :: Route Parser
- Camel :: JBang :: Plugin :: TUI
- Camel :: JBang :: Plugin :: Validate
- Camel :: Launcher :: Container
- Camel :: MongoDB
- Camel :: YAML DSL :: Validator
- Camel :: YAML DSL :: Validator Maven Plugin
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/24719900389)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Michwo (via GitHub)" <gi...@apache.org>.
Michwo commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4303606039
> Thanks for this contribution. However, I think this this could be implemented using our [Resume Strategies API](https://camel.apache.org/components/4.18.x/eips/resume-strategies.html).
> Hi @Michwo I think that this feature can be implemented with Camel's [resume strategy (ResumeAware Consumer)](https://camel.apache.org/components/4.18.x/eips/resume-strategies.html). Using the resume strategy, the Camel route would look like:
>
> ```
> from("mongodb:myDb?consumerType=changeStreams&database=mydb&collection=mycol")
> .resumable(myStategy)
> .process(this::process);
> ```
>
> The resumeToken handling will be offloaded to the framework, and hopefully it will be easier to use.
>
> The ResumeAware implementation can be done in another PR, no need to change the current implementation, but please open an issue about it.
>
> fyi @orpiske
Should I then abandon this PR and try to implement the resume strategy for camel-mongodb?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Michwo (via GitHub)" <gi...@apache.org>.
Michwo commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4780072303
I've finally found some time to work on that again - I've reimplemented this like @orpiske and @Croway have suggested, as an resume strategy eip.
I used the camel-kafka's implementation as a template for this.
Changes included:
- MongoDB change streams now work with the Resume API/EIP.
- Added MongoDB resume strategy + configuration builder support.
- Resume token (offset) key is now based on route id and collection (with endpoint fallback).
This was a decision based on the way WatchStreamConsumer is configured - more consumers per collection possible watching different changes based on BSON filter - therefore each consumer endpoint should be able to ingest the changes separately.
- Resume tokens are stored as JSON strings
- resumable completion updates the strategy from exchange offset
Startup token priority: resume strategy token first, then endpoint/repository
Test for the added code are included (value/integration).
The documentation is also updated to reflect the changes (camel-mongodb and Resume Strategies).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Michwo (via GitHub)" <gi...@apache.org>.
Michwo commented on code in PR #22674:
URL: https://github.com/apache/camel/pull/22674#discussion_r3102163520
##########
components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbChangeStreamsConsumerIT.java:
##########
@@ -167,6 +168,12 @@ public void operationTypeAndIdHeaderTest() throws Exception {
context.getRouteController().stopRoute(consumerRouteId);
}
+ @Order(5)
+ @Test
+ public void invalidResumeTokenShouldFailToStartRoute() {
+ assertThrows(Exception.class, () -> context.getRouteController().startRoute("invalidResumeTokenConsumer"));
+ }
+
Review Comment:
Makes sense, I will add that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4267877192
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Michwo (via GitHub)" <gi...@apache.org>.
Michwo commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4270101089
> you need to build the source code for the entire project as when you add new options there are code changed elsewhere
>
> ```
> modified: catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb.json
> modified: components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
> modified: dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MongoDbEndpointBuilderFactory.java
> ```
Hi, thanks for the info.
I think my problem is that I've done the build on Windows and it messed up all the line endings in generated source (over 3k files changed). I've commited the changes inside components/camel-mongodb but that seems not enough.
I will try to run the build on linux next week.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "Michwo (via GitHub)" <gi...@apache.org>.
Michwo commented on code in PR #22674:
URL: https://github.com/apache/camel/pull/22674#discussion_r3466591711
##########
components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java:
##########
@@ -20,29 +20,67 @@
import java.util.concurrent.ExecutorService;
import org.apache.camel.Processor;
+import org.apache.camel.resume.ResumeAdapter;
+import org.apache.camel.resume.ResumeAware;
+import org.apache.camel.resume.ResumeStrategy;
+import org.apache.camel.spi.StateRepository;
import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.resume.ResumeStrategyHelper;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
import org.bson.BsonDocument;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static java.util.Collections.singletonList;
/**
* The MongoDb Change Streams consumer.
*/
-public class MongoDbChangeStreamsConsumer extends DefaultConsumer {
+public class MongoDbChangeStreamsConsumer extends DefaultConsumer implements ResumeAware<ResumeStrategy> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MongoDbChangeStreamsConsumer.class);
+ private static final String MONGODB_RESUME_PLACEHOLDER_ACTION = "mongodb-resume";
private final MongoDbEndpoint endpoint;
private ExecutorService executor;
private MongoDbChangeStreamsThread changeStreamsThread;
+ private ResumeStrategy resumeStrategy;
+ private boolean stopOffsetRepo;
+ private volatile BsonDocument startupResumeToken;
Review Comment:
I've applied the fixes for points 1 and 3 and fixed the integration test. Turns out the integration tests were skipped on my local maven build therefore the oversight.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4268135156
you need to build the source code for the entire project as when you add new options there are code changed elsewhere
modified: catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb.json
modified: components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
modified: dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MongoDbEndpointBuilderFactory.java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "apupier (via GitHub)" <gi...@apache.org>.
apupier commented on code in PR #22674:
URL: https://github.com/apache/camel/pull/22674#discussion_r3100735551
##########
components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbChangeStreamsConsumerIT.java:
##########
@@ -167,6 +168,12 @@ public void operationTypeAndIdHeaderTest() throws Exception {
context.getRouteController().stopRoute(consumerRouteId);
}
+ @Order(5)
+ @Test
+ public void invalidResumeTokenShouldFailToStartRoute() {
+ assertThrows(Exception.class, () -> context.getRouteController().startRoute("invalidResumeTokenConsumer"));
+ }
+
Review Comment:
I think it will be worthy to add also a test when the resume token is valid to check for the happy path
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer) [camel]
Posted by "orpiske (via GitHub)" <gi...@apache.org>.
orpiske commented on PR #22674:
URL: https://github.com/apache/camel/pull/22674#issuecomment-4304328092
> Should I then abandon this Issue/PR and try to implement the resume strategy for camel-mongodb?
I think either way is fine. You are free to pick the way you find more convenient.
Personally, I'd prefer to reuse the PR, so we have the history of the conversation (specially because the idea **IS** good - we'd like to have this on the code - it's just the implementation that needs to be tweaked to fit into what we already have).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org