CAMEL-23336: camel-mongodb - Expose Resume Token (Change Streams Consumer)

components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointConfigurer.java

@@ -29,6 +29,10 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "authSource": target.setAuthSource(property(camelContext, java.lang.String.class, value)); return true;
         case "bridgeerrorhandler":
         case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+        case "changestreamtoken":
+        case "changeStreamToken": target.setChangeStreamToken(property(camelContext, java.lang.String.class, value)); return true;
+        case "changestreamtokenrepository":
+        case "changeStreamTokenRepository": target.setChangeStreamTokenRepository(property(camelContext, org.apache.camel.spi.StateRepository.class, value)); return true;
         case "collection": target.setCollection(property(camelContext, java.lang.String.class, value)); return true;
         case "collectionindex":
         case "collectionIndex": target.setCollectionIndex(property(camelContext, java.lang.String.class, value)); return true;
@@ -139,6 +143,10 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "authSource": return java.lang.String.class;
         case "bridgeerrorhandler":
         case "bridgeErrorHandler": return boolean.class;
+        case "changestreamtoken":
+        case "changeStreamToken": return java.lang.String.class;
+        case "changestreamtokenrepository":
+        case "changeStreamTokenRepository": return org.apache.camel.spi.StateRepository.class;
         case "collection": return java.lang.String.class;
         case "collectionindex":
         case "collectionIndex": return java.lang.String.class;
@@ -250,6 +258,10 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "authSource": return target.getAuthSource();
         case "bridgeerrorhandler":
         case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+        case "changestreamtoken":
+        case "changeStreamToken": return target.getChangeStreamToken();
+        case "changestreamtokenrepository":
+        case "changeStreamTokenRepository": return target.getChangeStreamTokenRepository();
         case "collection": return target.getCollection();
         case "collectionindex":
         case "collectionIndex": return target.getCollectionIndex();
@@ -350,5 +362,14 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         default: return null;
         }
     }
+
+    @Override
+    public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+        switch (ignoreCase ? name.toLowerCase() : name) {
+        case "changestreamtokenrepository":
+        case "changeStreamTokenRepository": return java.lang.String.class;
+        default: return null;
+        }
+    }
 }
 

components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointUriFactory.java

@@ -24,10 +24,12 @@ public class MongoDbEndpointUriFactory extends org.apache.camel.support.componen
     private static final Set<String> ENDPOINT_IDENTITY_PROPERTY_NAMES;
     private static final Map<String, String> MULTI_VALUE_PREFIXES;
     static {
-        Set<String> props = new HashSet<>(57);
+        Set<String> props = new HashSet<>(59);
         props.add("appName");
         props.add("authSource");
         props.add("bridgeErrorHandler");
+        props.add("changeStreamToken");
+        props.add("changeStreamTokenRepository");
         props.add("collection");
         props.add("collectionIndex");
         props.add("compressors");

components/camel-mongodb/src/generated/resources/META-INF/services/org/apache/camel/mongodb-adapter-factory

@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.mongodb.MongoDbResumeAdapter

components/camel-mongodb/src/generated/resources/META-INF/services/org/apache/camel/mongodb-resume-strategy

@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.processor.resume.mongodb.MongoDbResumeStrategy

components/camel-mongodb/src/main/docs/mongodb-component.adoc

@@ -1251,6 +1251,47 @@ YAML::
 TIP: You can externalize the streamFilter value into a property placeholder which allows the endpoint
 URI parameters to be _cleaner_ and easier to read.
 
+==== Change Streams Resume
+
+The Change Streams consumer supports resume in two complementary ways:
+
+* Endpoint options (`changeStreamTokenRepository`, `changeStreamToken`) to restore and persist the resume token.
+* The xref:eips:resume-strategies.adoc[Resume Strategies] EIP (`resumable()`) for route-level resume handling.
+
+When a route id is configured, the token key is computed as `<routeId>/<collection>`.
+If no route id is configured, Camel falls back to the endpoint URI.
+
+The resume token is a BSON document. It is persisted as a serialized JSON string and parsed back to a `BsonDocument` when restored.
+
+The following example configures endpoint-level token persistence:
+
+[source,java]
+----
+from("mongodb:myDb?consumerType=changeStreams"
+   + "&database=flights"
+   + "&collection=tickets"
+   + "&changeStreamTokenRepository=#changeStreamTokenRepo")
+  .id("resumeChangeStreamConsumer")
+  .to("mock:test");
+----
+
+The following example configures route-level resume strategy support:
+
+[source,java]
+----
+MongoDbResumeStrategyConfigurationBuilder configurationBuilder =
+    MongoDbResumeStrategyConfigurationBuilder.newBuilder()
+        .withStateRepository(memoryStateRepository)
+        .withResumeCache(new CaffeineCache<>(100));
+
+from("mongodb:myDb?consumerType=changeStreams&database=flights&collection=tickets")
+  .id("resumeChangeStreamConsumer")
+  .resumable().configuration(configurationBuilder)
+  .to("mock:test");
+----
+
+You can use only endpoint options, only route-level resumable configuration, or both depending on your deployment requirements.
+
 === Type conversions
 
 The `MongoDbBasicConverters` type converter included with the

components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/ChangeStreamCommitManager.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mongodb;
+
+import org.apache.camel.spi.StateRepository;
+import org.apache.camel.util.ObjectHelper;
+import org.bson.BsonDocument;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ChangeStreamCommitManager implements CommitManager {
+    private static final Logger LOG = LoggerFactory.getLogger(ChangeStreamCommitManager.class);
+
+    private final StateRepository<String, String> resumeTokenRepository;
+    private final String resumeTokenKey;
+    private final String explicitResumeToken;
+
+    private BsonDocument cachedResumeToken;
+
+    public ChangeStreamCommitManager(MongoDbChangeStreamsConsumer consumer, MongoDbEndpoint endpoint) {
+        this.resumeTokenRepository = endpoint.getChangeStreamTokenRepository();
+        this.explicitResumeToken = endpoint.getChangeStreamToken();
+
+        String routeId = consumer.getRouteId();
+        if (ObjectHelper.isEmpty(routeId)) {
+            routeId = endpoint.getEndpointUri();
+        }
+
+        this.resumeTokenKey = serializeResumeTokenKey(routeId, endpoint.getCollection());
+    }
+
+    @Override
+    public BsonDocument readResumeToken() {
+        if (ObjectHelper.isNotEmpty(explicitResumeToken)) {
+            return BsonDocument.parse(explicitResumeToken);
+        }
+
+        if (resumeTokenRepository == null) {
+            return null;
+        }
+
+        String serialized = resumeTokenRepository.getState(resumeTokenKey);
+        if (ObjectHelper.isEmpty(serialized)) {
+            return null;
+        }
+
+        return BsonDocument.parse(serialized);
+    }
+
+    @Override
+    public void recordResumeToken(BsonDocument resumeToken) {
+        this.cachedResumeToken = resumeToken;
+    }
+
+    @Override
+    public void commit() throws Exception {
+        if (cachedResumeToken == null) {
+            return;
+        }
+
+        if (resumeTokenRepository != null) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Saving resume token repository state for key {}", resumeTokenKey);
+            }
+            resumeTokenRepository.setState(resumeTokenKey, cachedResumeToken.toJson());
+        }
+    }
+
+    private static String serializeResumeTokenKey(String routeId, String collection) {
+        return routeId + '/' + collection;
+    }
+}

components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/CommitManager.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mongodb;
+
+import org.bson.BsonDocument;
+
+public interface CommitManager {
+
+    BsonDocument readResumeToken();
+
+    void recordResumeToken(BsonDocument resumeToken);
+
+    void commit() throws Exception;
+}

components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/CommitManagers.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mongodb;
+
+public final class CommitManagers {
+
+    private CommitManagers() {
+    }
+
+    public static CommitManager createCommitManager(MongoDbChangeStreamsConsumer consumer, MongoDbEndpoint endpoint) {
+        if (endpoint.getChangeStreamTokenRepository() == null && consumer.getResumeStrategy() == null) {
+            return new NoopCommitManager(endpoint);
+        }
+
+        return new ChangeStreamCommitManager(consumer, endpoint);
+    }
+}

components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java

@@ -20,53 +20,129 @@
 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;
 
     public MongoDbChangeStreamsConsumer(MongoDbEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
         this.endpoint = endpoint;
     }
 
+    @Override
+    public void setResumeStrategy(ResumeStrategy resumeStrategy) {
+        this.resumeStrategy = resumeStrategy;
+    }
+
+    @Override
+    public ResumeStrategy getResumeStrategy() {
+        return resumeStrategy;
+    }
+
+    @Override
+    public String adapterFactoryService() {
+        return "mongodb-adapter-factory";
+    }
+
     @Override
     protected void doStop() throws Exception {
         super.doStop();
+
         if (changeStreamsThread != null) {
             changeStreamsThread.stop();
         }
         if (executor != null) {
             endpoint.getCamelContext().getExecutorServiceManager().shutdown(executor);
             executor = null;
         }
+
+        if (stopOffsetRepo) {
+            StateRepository<String, String> repo = endpoint.getChangeStreamTokenRepository();
+            LOG.debug("Stopping ChangeStreamTokenRepository: {}", repo);
+            ServiceHelper.stopAndShutdownService(repo);
+        }
     }
 
     @Override
     protected void doStart() throws Exception {
         super.doStart();
 
+        // Is the change stream token repository already started?
+        StateRepository<String, String> repo = endpoint.getChangeStreamTokenRepository();
+        if (repo instanceof ServiceSupport serviceSupport) {
+            boolean started = serviceSupport.isStarted();
+            // If not already started then start and mark to stop when stopping the consumer
+            if (!started) {
+                stopOffsetRepo = true;
+                LOG.debug("Starting ChangeStreamTokenRepository: {}", repo);
+                ServiceHelper.startService(endpoint.getChangeStreamTokenRepository());
+            }
+        }
+
         String streamFilter = endpoint.getStreamFilter();
         List<BsonDocument> bsonFilter = null;
         if (ObjectHelper.isNotEmpty(streamFilter)) {
             bsonFilter = singletonList(BsonDocument.parse(streamFilter));
         }
 
+        if (resumeStrategy != null) {
+            ResumeAdapter resumeAdapter = resumeStrategy.getAdapter();
+            if (resumeAdapter instanceof MongoDbResumeAdapter adapter) {
+                adapter.setResumeTokenKey(getResumeTokenKey());
+                adapter.setConsumer(this);
+            }
+
+            ResumeStrategyHelper.resume(getEndpoint().getCamelContext(), this, resumeStrategy,
+                    MONGODB_RESUME_PLACEHOLDER_ACTION);
+        }
+
         executor = endpoint.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this,
                 endpoint.getEndpointUri(), 1);
         changeStreamsThread = new MongoDbChangeStreamsThread(endpoint, this, bsonFilter);
         changeStreamsThread.init();
         executor.execute(changeStreamsThread);
     }
 
+    BsonDocument getStartupResumeToken() {
+        return startupResumeToken;
+    }
+
+    void setStartupResumeToken(BsonDocument startupResumeToken) {
+        this.startupResumeToken = startupResumeToken;
+    }
+
+    String getResumeTokenKey() {
+        String routeId = getRouteId();
+        if (ObjectHelper.isEmpty(routeId)) {
+            routeId = endpoint.getEndpointUri();
+        }
+        return routeId + '/' + endpoint.getCollection();
+    }
+
 }