diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb.json index c4f64a29d3875..4dd51b35272ff 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/mongodb.json @@ -69,52 +69,54 @@ "mongoConnection": { "index": 7, "kind": "parameter", "displayName": "Mongo Connection", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.mongodb.client.MongoClient", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the connection bean used as a client for connecting to a database." }, "operation": { "index": 8, "kind": "parameter", "displayName": "Operation", "group": "common", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.mongodb.MongoDbOperation", "enum": [ "findById", "findOneAndDelete", "findOneAndReplace", "findOneAndUpdate", "findOneByQuery", "findAll", "findDistinct", "insert", "save", "update", "remove", "bulkWrite", "aggregate", "getDbStats", "getColStats", "count", "command" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the operation this endpoint will execute against MongoDB." }, "outputType": { "index": 9, "kind": "parameter", "displayName": "Output Type", "group": "common", "label": "common", "required": false, "type": "enum", "javaType": "org.apache.camel.component.mongodb.MongoDbOutputType", "enum": [ "DocumentList", "Document", "MongoIterable" ], "deprecated": false, "autowired": false, "secret": false, "description": "Convert the output of the producer to the selected type: DocumentList Document or MongoIterable. DocumentList or MongoIterable applies to findAll and aggregate. Document applies to all other operations." }, - "consumerType": { "index": 10, "kind": "parameter", "displayName": "Consumer Type", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "tailable", "changeStreams" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "tailable", "description": "Consumer type." }, - "fullDocument": { "index": 11, "kind": "parameter", "displayName": "Full Document", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "com.mongodb.client.model.changestream.FullDocument", "enum": [ "default", "updateLookup", "required", "whenAvailable" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "default", "description": "Specifies whether changeStream consumer include a copy of the full document when modified by update operations. Possible values are default, updateLookup, required and whenAvailable." }, - "persistentId": { "index": 12, "kind": "parameter", "displayName": "Persistent Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "One tail tracking collection can host many trackers for several tailable consumers. To keep them separate, each tracker should have its own unique persistentId." }, - "persistentTailTracking": { "index": 13, "kind": "parameter", "displayName": "Persistent Tail Tracking", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable persistent tail tracking, which is a mechanism to keep track of the last consumed message across system restarts. The next time the system is up, the endpoint will recover the cursor from the point where it last stopped slurping records." }, - "tailTrackCollection": { "index": 14, "kind": "parameter", "displayName": "Tail Track Collection", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Collection where tail tracking information will be persisted. If not specified, MongoDbTailTrackingConfig#DEFAULT_COLLECTION will be used by default." }, - "tailTrackDb": { "index": 15, "kind": "parameter", "displayName": "Tail Track Db", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Indicates what database the tail tracking mechanism will persist to. If not specified, the current database will be picked by default. Dynamicity will not be taken into account even if enabled, i.e., the tail tracking database will not vary past endpoint initialization." }, - "tailTrackField": { "index": 16, "kind": "parameter", "displayName": "Tail Track Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Field where the last tracked value will be placed. If not specified, MongoDbTailTrackingConfig#DEFAULT_FIELD will be used by default." }, - "tailTrackIncreasingField": { "index": 17, "kind": "parameter", "displayName": "Tail Track Increasing Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Correlation field in the incoming record which is of increasing nature and will be used to position the tailing cursor every time it is generated. The cursor will be (re)created with a query of type: tailTrackIncreasingField greater than lastValue (possibly recovered from persistent tail tracking). Can be of type Integer, Date, String, etc. NOTE: No support for dot notation at the current time, so the field should be at the top level of the document." }, - "bridgeErrorHandler": { "index": 18, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "exceptionHandler": { "index": 19, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "exchangePattern": { "index": 20, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, - "lazyStartProducer": { "index": 21, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, - "appName": { "index": 22, "kind": "parameter", "displayName": "App Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the logical name of the application. The application name may be used by the client to identify the application to the server, for use in server logs, slow query logs, and profile collection. Default: null" }, - "compressors": { "index": 23, "kind": "parameter", "displayName": "Compressors", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies one or more compression algorithms that the driver will attempt to use to compress requests sent to the connected MongoDB instance. Possible values include: zlib, snappy, and zstd. Default: null" }, - "connectTimeoutMS": { "index": 24, "kind": "parameter", "displayName": "Connect Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10000, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver waits for a connection to open before timing out. A value of 0 instructs the driver to never time out while waiting for a connection to open. Default: 10000 (10 seconds)" }, - "cursorRegenerationDelay": { "index": 25, "kind": "parameter", "displayName": "Cursor Regeneration Delay", "group": "advanced", "label": "advanced", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "MongoDB tailable cursors will block until new data arrives. If no new data is inserted, after some time the cursor will be automatically freed and closed by the MongoDB server. The client is expected to regenerate the cursor if needed. This value specifies the time to wait before attempting to fetch a new cursor, and if the attempt fails, how long before the next attempt is made. Default value is 1000ms." }, - "directConnection": { "index": 26, "kind": "parameter", "displayName": "Direct Connection", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver must connect to the host directly. Default: false" }, - "dynamicity": { "index": 27, "kind": "parameter", "displayName": "Dynamicity", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether this endpoint will attempt to dynamically resolve the target database and collection from the incoming Exchange properties. Can be used to override at runtime the database and collection specified on the otherwise static endpoint URI. It is disabled by default to boost performance. Enabling it will take a minimal performance hit." }, - "heartbeatFrequencyMS": { "index": 28, "kind": "parameter", "displayName": "Heartbeat Frequency MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "heartbeatFrequencyMS controls when the driver checks the state of the MongoDB deployment. Specify the interval (in milliseconds) between checks, counted from the end of the previous check until the beginning of the next one. Default: Single-threaded drivers: 60 seconds. Multithreaded drivers: 10 seconds." }, - "loadBalanced": { "index": 29, "kind": "parameter", "displayName": "Load Balanced", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If true the driver will assume that it's connecting to MongoDB through a load balancer." }, - "localThresholdMS": { "index": 30, "kind": "parameter", "displayName": "Local Threshold MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 15, "description": "The size (in milliseconds) of the latency window for selecting among multiple suitable MongoDB instances. Default: 15 milliseconds." }, - "maxConnecting": { "index": 31, "kind": "parameter", "displayName": "Max Connecting", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 2, "description": "Specifies the maximum number of connections a pool may be establishing concurrently. Default: 2" }, - "maxIdleTimeMS": { "index": 32, "kind": "parameter", "displayName": "Max Idle Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will allow a pooled connection to idle before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can allow a pooled collection to be idle. Default: 0" }, - "maxLifeTimeMS": { "index": 33, "kind": "parameter", "displayName": "Max Life Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will continue to use a pooled connection before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can keep a pooled connection open. Default: 0" }, - "maxPoolSize": { "index": 34, "kind": "parameter", "displayName": "Max Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 100, "description": "The maximum number of connections in the connection pool. The default value is 100." }, - "maxStalenessSeconds": { "index": 35, "kind": "parameter", "displayName": "Max Staleness Seconds", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": -1, "description": "Specifies, in seconds, how stale a secondary can be before the driver stops communicating with that secondary. The minimum value is either 90 seconds or the heartbeat frequency plus 10 seconds, whichever is greater. For more information, see the server documentation for the maxStalenessSeconds option. Not providing a parameter or explicitly specifying -1 indicates that there should be no staleness check for secondaries. Default: -1" }, - "minPoolSize": { "index": 36, "kind": "parameter", "displayName": "Min Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the minimum number of connections that must exist at any moment in a single connection pool. Default: 0" }, - "readPreference": { "index": 37, "kind": "parameter", "displayName": "Read Preference", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "PRIMARY", "PRIMARY_PREFERRED", "SECONDARY", "SECONDARY_PREFERRED", "NEAREST" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PRIMARY", "description": "Configure how MongoDB clients route read operations to the members of a replica set. Possible values are PRIMARY, PRIMARY_PREFERRED, SECONDARY, SECONDARY_PREFERRED or NEAREST" }, - "readPreferenceTags": { "index": 38, "kind": "parameter", "displayName": "Read Preference Tags", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "A representation of a tag set as a comma-separated list of colon-separated key-value pairs, e.g. dc:ny,rack:1. Spaces are stripped from the beginning and end of all keys and values. To specify a list of tag sets, using multiple readPreferenceTags, e.g., readPreferenceTags=dc:ny,rack:1;readPreferenceTags=dc:ny;readPreferenceTags= Note the empty value for the last one, which means match any secondary as a last resort. Order matters when using multiple readPreferenceTags." }, - "replicaSet": { "index": 39, "kind": "parameter", "displayName": "Replica Set", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies that the connection string provided includes multiple hosts. When specified, the driver attempts to find all members of that set." }, - "retryReads": { "index": 40, "kind": "parameter", "displayName": "Retry Reads", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported read operations if they fail due to a network error. Default: true" }, - "retryWrites": { "index": 41, "kind": "parameter", "displayName": "Retry Writes", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported write operations if they fail due to a network error. Default: true" }, - "serverSelectionTimeoutMS": { "index": 42, "kind": "parameter", "displayName": "Server Selection Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "Specifies how long (in milliseconds) to block for server selection before throwing an exception. Default: 30,000 milliseconds." }, - "socketTimeoutMS": { "index": 43, "kind": "parameter", "displayName": "Socket Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will wait to send or receive a request before timing out. A value of 0 instructs the driver to never time out while waiting to send or receive a request. Default: 0" }, - "srvMaxHosts": { "index": 44, "kind": "parameter", "displayName": "Srv Max Hosts", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "The maximum number of hosts from the SRV record to connect to." }, - "srvServiceName": { "index": 45, "kind": "parameter", "displayName": "Srv Service Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "mongodb", "description": "Specifies the service name of the SRV resource recordsthe driver retrieves to construct your seed list. You must use the DNS Seed List Connection Format in your connection URI to use this option. Default: mongodb" }, - "waitQueueTimeoutMS": { "index": 46, "kind": "parameter", "displayName": "Wait Queue Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 120000, "description": "Specifies the maximum amount of time, in milliseconds that a thread may wait for a connection to become available. Default: 120000 (120 seconds)" }, - "writeConcern": { "index": 47, "kind": "parameter", "displayName": "Write Concern", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "ACKNOWLEDGED", "W1", "W2", "W3", "UNACKNOWLEDGED", "JOURNALED", "MAJORITY" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ACKNOWLEDGED", "description": "Configure the connection bean with the level of acknowledgment requested from MongoDB for write operations to a standalone mongod, replicaset or cluster. Possible values are ACKNOWLEDGED, W1, W2, W3, UNACKNOWLEDGED, JOURNALED or MAJORITY." }, - "writeResultAsHeader": { "index": 48, "kind": "parameter", "displayName": "Write Result As Header", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "In write operations, it determines whether instead of returning WriteResult as the body of the OUT message, we transfer the IN message to the OUT and attach the WriteResult as a header." }, - "zlibCompressionLevel": { "index": 49, "kind": "parameter", "displayName": "Zlib Compression Level", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies the degree of compression that Zlib should use to decrease the size of requests to the connected MongoDB instance. The level can range from -1 to 9, with lower values compressing faster (but resulting in larger requests) and larger values compressing slower (but resulting in smaller requests). Default: null" }, - "streamFilter": { "index": 50, "kind": "parameter", "displayName": "Stream Filter", "group": "changeStream", "label": "consumer,changeStream", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Filter condition for change streams consumer." }, - "authSource": { "index": 51, "kind": "parameter", "displayName": "Auth Source", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The database name associated with the user's credentials." }, - "password": { "index": 52, "kind": "parameter", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "User password for mongodb connection" }, - "sslContextParameters": { "index": 53, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "SSL configuration using a Camel SSLContextParameters object. When configured, TLS is automatically enabled on the connection." }, - "tls": { "index": 54, "kind": "parameter", "displayName": "Tls", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that all communication with MongoDB instances should use TLS. Supersedes the ssl option. Default: false" }, - "tlsAllowInvalidHostnames": { "index": 55, "kind": "parameter", "displayName": "Tls Allow Invalid Hostnames", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver should allow invalid hostnames in the certificate for TLS connections. Supersedes sslInvalidHostNameAllowed. Has the same effect as tlsInsecure by setting tlsAllowInvalidHostnames to true. Default: false" }, - "username": { "index": 56, "kind": "parameter", "displayName": "Username", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Username for mongodb connection" } + "changeStreamToken": { "index": 10, "kind": "parameter", "displayName": "Change Stream Token", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "A change stream token as a serialized BSON document JSON string." }, + "changeStreamTokenRepository": { "index": 11, "kind": "parameter", "displayName": "Change Stream Token Repository", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.spi.StateRepository", "deprecated": false, "autowired": false, "secret": false, "description": "The repository to store change stream tokens." }, + "consumerType": { "index": 12, "kind": "parameter", "displayName": "Consumer Type", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "tailable", "changeStreams" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "tailable", "description": "Consumer type." }, + "fullDocument": { "index": 13, "kind": "parameter", "displayName": "Full Document", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "com.mongodb.client.model.changestream.FullDocument", "enum": [ "default", "updateLookup", "required", "whenAvailable" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "default", "description": "Specifies whether changeStream consumer include a copy of the full document when modified by update operations. Possible values are default, updateLookup, required and whenAvailable." }, + "persistentId": { "index": 14, "kind": "parameter", "displayName": "Persistent Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "One tail tracking collection can host many trackers for several tailable consumers. To keep them separate, each tracker should have its own unique persistentId." }, + "persistentTailTracking": { "index": 15, "kind": "parameter", "displayName": "Persistent Tail Tracking", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable persistent tail tracking, which is a mechanism to keep track of the last consumed message across system restarts. The next time the system is up, the endpoint will recover the cursor from the point where it last stopped slurping records." }, + "tailTrackCollection": { "index": 16, "kind": "parameter", "displayName": "Tail Track Collection", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Collection where tail tracking information will be persisted. If not specified, MongoDbTailTrackingConfig#DEFAULT_COLLECTION will be used by default." }, + "tailTrackDb": { "index": 17, "kind": "parameter", "displayName": "Tail Track Db", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Indicates what database the tail tracking mechanism will persist to. If not specified, the current database will be picked by default. Dynamicity will not be taken into account even if enabled, i.e., the tail tracking database will not vary past endpoint initialization." }, + "tailTrackField": { "index": 18, "kind": "parameter", "displayName": "Tail Track Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Field where the last tracked value will be placed. If not specified, MongoDbTailTrackingConfig#DEFAULT_FIELD will be used by default." }, + "tailTrackIncreasingField": { "index": 19, "kind": "parameter", "displayName": "Tail Track Increasing Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Correlation field in the incoming record which is of increasing nature and will be used to position the tailing cursor every time it is generated. The cursor will be (re)created with a query of type: tailTrackIncreasingField greater than lastValue (possibly recovered from persistent tail tracking). Can be of type Integer, Date, String, etc. NOTE: No support for dot notation at the current time, so the field should be at the top level of the document." }, + "bridgeErrorHandler": { "index": 20, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, + "exceptionHandler": { "index": 21, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, + "exchangePattern": { "index": 22, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, + "lazyStartProducer": { "index": 23, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, + "appName": { "index": 24, "kind": "parameter", "displayName": "App Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the logical name of the application. The application name may be used by the client to identify the application to the server, for use in server logs, slow query logs, and profile collection. Default: null" }, + "compressors": { "index": 25, "kind": "parameter", "displayName": "Compressors", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies one or more compression algorithms that the driver will attempt to use to compress requests sent to the connected MongoDB instance. Possible values include: zlib, snappy, and zstd. Default: null" }, + "connectTimeoutMS": { "index": 26, "kind": "parameter", "displayName": "Connect Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10000, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver waits for a connection to open before timing out. A value of 0 instructs the driver to never time out while waiting for a connection to open. Default: 10000 (10 seconds)" }, + "cursorRegenerationDelay": { "index": 27, "kind": "parameter", "displayName": "Cursor Regeneration Delay", "group": "advanced", "label": "advanced", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "MongoDB tailable cursors will block until new data arrives. If no new data is inserted, after some time the cursor will be automatically freed and closed by the MongoDB server. The client is expected to regenerate the cursor if needed. This value specifies the time to wait before attempting to fetch a new cursor, and if the attempt fails, how long before the next attempt is made. Default value is 1000ms." }, + "directConnection": { "index": 28, "kind": "parameter", "displayName": "Direct Connection", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver must connect to the host directly. Default: false" }, + "dynamicity": { "index": 29, "kind": "parameter", "displayName": "Dynamicity", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether this endpoint will attempt to dynamically resolve the target database and collection from the incoming Exchange properties. Can be used to override at runtime the database and collection specified on the otherwise static endpoint URI. It is disabled by default to boost performance. Enabling it will take a minimal performance hit." }, + "heartbeatFrequencyMS": { "index": 30, "kind": "parameter", "displayName": "Heartbeat Frequency MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "heartbeatFrequencyMS controls when the driver checks the state of the MongoDB deployment. Specify the interval (in milliseconds) between checks, counted from the end of the previous check until the beginning of the next one. Default: Single-threaded drivers: 60 seconds. Multithreaded drivers: 10 seconds." }, + "loadBalanced": { "index": 31, "kind": "parameter", "displayName": "Load Balanced", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If true the driver will assume that it's connecting to MongoDB through a load balancer." }, + "localThresholdMS": { "index": 32, "kind": "parameter", "displayName": "Local Threshold MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 15, "description": "The size (in milliseconds) of the latency window for selecting among multiple suitable MongoDB instances. Default: 15 milliseconds." }, + "maxConnecting": { "index": 33, "kind": "parameter", "displayName": "Max Connecting", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 2, "description": "Specifies the maximum number of connections a pool may be establishing concurrently. Default: 2" }, + "maxIdleTimeMS": { "index": 34, "kind": "parameter", "displayName": "Max Idle Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will allow a pooled connection to idle before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can allow a pooled collection to be idle. Default: 0" }, + "maxLifeTimeMS": { "index": 35, "kind": "parameter", "displayName": "Max Life Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will continue to use a pooled connection before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can keep a pooled connection open. Default: 0" }, + "maxPoolSize": { "index": 36, "kind": "parameter", "displayName": "Max Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 100, "description": "The maximum number of connections in the connection pool. The default value is 100." }, + "maxStalenessSeconds": { "index": 37, "kind": "parameter", "displayName": "Max Staleness Seconds", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": -1, "description": "Specifies, in seconds, how stale a secondary can be before the driver stops communicating with that secondary. The minimum value is either 90 seconds or the heartbeat frequency plus 10 seconds, whichever is greater. For more information, see the server documentation for the maxStalenessSeconds option. Not providing a parameter or explicitly specifying -1 indicates that there should be no staleness check for secondaries. Default: -1" }, + "minPoolSize": { "index": 38, "kind": "parameter", "displayName": "Min Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the minimum number of connections that must exist at any moment in a single connection pool. Default: 0" }, + "readPreference": { "index": 39, "kind": "parameter", "displayName": "Read Preference", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "PRIMARY", "PRIMARY_PREFERRED", "SECONDARY", "SECONDARY_PREFERRED", "NEAREST" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PRIMARY", "description": "Configure how MongoDB clients route read operations to the members of a replica set. Possible values are PRIMARY, PRIMARY_PREFERRED, SECONDARY, SECONDARY_PREFERRED or NEAREST" }, + "readPreferenceTags": { "index": 40, "kind": "parameter", "displayName": "Read Preference Tags", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "A representation of a tag set as a comma-separated list of colon-separated key-value pairs, e.g. dc:ny,rack:1. Spaces are stripped from the beginning and end of all keys and values. To specify a list of tag sets, using multiple readPreferenceTags, e.g., readPreferenceTags=dc:ny,rack:1;readPreferenceTags=dc:ny;readPreferenceTags= Note the empty value for the last one, which means match any secondary as a last resort. Order matters when using multiple readPreferenceTags." }, + "replicaSet": { "index": 41, "kind": "parameter", "displayName": "Replica Set", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies that the connection string provided includes multiple hosts. When specified, the driver attempts to find all members of that set." }, + "retryReads": { "index": 42, "kind": "parameter", "displayName": "Retry Reads", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported read operations if they fail due to a network error. Default: true" }, + "retryWrites": { "index": 43, "kind": "parameter", "displayName": "Retry Writes", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported write operations if they fail due to a network error. Default: true" }, + "serverSelectionTimeoutMS": { "index": 44, "kind": "parameter", "displayName": "Server Selection Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "Specifies how long (in milliseconds) to block for server selection before throwing an exception. Default: 30,000 milliseconds." }, + "socketTimeoutMS": { "index": 45, "kind": "parameter", "displayName": "Socket Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will wait to send or receive a request before timing out. A value of 0 instructs the driver to never time out while waiting to send or receive a request. Default: 0" }, + "srvMaxHosts": { "index": 46, "kind": "parameter", "displayName": "Srv Max Hosts", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "The maximum number of hosts from the SRV record to connect to." }, + "srvServiceName": { "index": 47, "kind": "parameter", "displayName": "Srv Service Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "mongodb", "description": "Specifies the service name of the SRV resource recordsthe driver retrieves to construct your seed list. You must use the DNS Seed List Connection Format in your connection URI to use this option. Default: mongodb" }, + "waitQueueTimeoutMS": { "index": 48, "kind": "parameter", "displayName": "Wait Queue Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 120000, "description": "Specifies the maximum amount of time, in milliseconds that a thread may wait for a connection to become available. Default: 120000 (120 seconds)" }, + "writeConcern": { "index": 49, "kind": "parameter", "displayName": "Write Concern", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "ACKNOWLEDGED", "W1", "W2", "W3", "UNACKNOWLEDGED", "JOURNALED", "MAJORITY" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ACKNOWLEDGED", "description": "Configure the connection bean with the level of acknowledgment requested from MongoDB for write operations to a standalone mongod, replicaset or cluster. Possible values are ACKNOWLEDGED, W1, W2, W3, UNACKNOWLEDGED, JOURNALED or MAJORITY." }, + "writeResultAsHeader": { "index": 50, "kind": "parameter", "displayName": "Write Result As Header", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "In write operations, it determines whether instead of returning WriteResult as the body of the OUT message, we transfer the IN message to the OUT and attach the WriteResult as a header." }, + "zlibCompressionLevel": { "index": 51, "kind": "parameter", "displayName": "Zlib Compression Level", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies the degree of compression that Zlib should use to decrease the size of requests to the connected MongoDB instance. The level can range from -1 to 9, with lower values compressing faster (but resulting in larger requests) and larger values compressing slower (but resulting in smaller requests). Default: null" }, + "streamFilter": { "index": 52, "kind": "parameter", "displayName": "Stream Filter", "group": "changeStream", "label": "consumer,changeStream", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Filter condition for change streams consumer." }, + "authSource": { "index": 53, "kind": "parameter", "displayName": "Auth Source", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The database name associated with the user's credentials." }, + "password": { "index": 54, "kind": "parameter", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "User password for mongodb connection" }, + "sslContextParameters": { "index": 55, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "SSL configuration using a Camel SSLContextParameters object. When configured, TLS is automatically enabled on the connection." }, + "tls": { "index": 56, "kind": "parameter", "displayName": "Tls", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that all communication with MongoDB instances should use TLS. Supersedes the ssl option. Default: false" }, + "tlsAllowInvalidHostnames": { "index": 57, "kind": "parameter", "displayName": "Tls Allow Invalid Hostnames", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver should allow invalid hostnames in the certificate for TLS connections. Supersedes sslInvalidHostNameAllowed. Has the same effect as tlsInsecure by setting tlsAllowInvalidHostnames to true. Default: false" }, + "username": { "index": 58, "kind": "parameter", "displayName": "Username", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Username for mongodb connection" } } } diff --git a/components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointConfigurer.java b/components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointConfigurer.java index b153c2f3b11c0..7b6608f49f70f 100644 --- a/components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointConfigurer.java +++ b/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; + } + } } diff --git a/components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointUriFactory.java b/components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointUriFactory.java index c6cd565c44c4c..4416b532f7e9f 100644 --- a/components/camel-mongodb/src/generated/java/org/apache/camel/component/mongodb/MongoDbEndpointUriFactory.java +++ b/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 ENDPOINT_IDENTITY_PROPERTY_NAMES; private static final Map MULTI_VALUE_PREFIXES; static { - Set props = new HashSet<>(57); + Set 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"); diff --git a/components/camel-mongodb/src/generated/resources/META-INF/org/apache/camel/component/mongodb/mongodb.json b/components/camel-mongodb/src/generated/resources/META-INF/org/apache/camel/component/mongodb/mongodb.json index c4f64a29d3875..4dd51b35272ff 100644 --- a/components/camel-mongodb/src/generated/resources/META-INF/org/apache/camel/component/mongodb/mongodb.json +++ b/components/camel-mongodb/src/generated/resources/META-INF/org/apache/camel/component/mongodb/mongodb.json @@ -69,52 +69,54 @@ "mongoConnection": { "index": 7, "kind": "parameter", "displayName": "Mongo Connection", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.mongodb.client.MongoClient", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the connection bean used as a client for connecting to a database." }, "operation": { "index": 8, "kind": "parameter", "displayName": "Operation", "group": "common", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.mongodb.MongoDbOperation", "enum": [ "findById", "findOneAndDelete", "findOneAndReplace", "findOneAndUpdate", "findOneByQuery", "findAll", "findDistinct", "insert", "save", "update", "remove", "bulkWrite", "aggregate", "getDbStats", "getColStats", "count", "command" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the operation this endpoint will execute against MongoDB." }, "outputType": { "index": 9, "kind": "parameter", "displayName": "Output Type", "group": "common", "label": "common", "required": false, "type": "enum", "javaType": "org.apache.camel.component.mongodb.MongoDbOutputType", "enum": [ "DocumentList", "Document", "MongoIterable" ], "deprecated": false, "autowired": false, "secret": false, "description": "Convert the output of the producer to the selected type: DocumentList Document or MongoIterable. DocumentList or MongoIterable applies to findAll and aggregate. Document applies to all other operations." }, - "consumerType": { "index": 10, "kind": "parameter", "displayName": "Consumer Type", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "tailable", "changeStreams" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "tailable", "description": "Consumer type." }, - "fullDocument": { "index": 11, "kind": "parameter", "displayName": "Full Document", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "com.mongodb.client.model.changestream.FullDocument", "enum": [ "default", "updateLookup", "required", "whenAvailable" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "default", "description": "Specifies whether changeStream consumer include a copy of the full document when modified by update operations. Possible values are default, updateLookup, required and whenAvailable." }, - "persistentId": { "index": 12, "kind": "parameter", "displayName": "Persistent Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "One tail tracking collection can host many trackers for several tailable consumers. To keep them separate, each tracker should have its own unique persistentId." }, - "persistentTailTracking": { "index": 13, "kind": "parameter", "displayName": "Persistent Tail Tracking", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable persistent tail tracking, which is a mechanism to keep track of the last consumed message across system restarts. The next time the system is up, the endpoint will recover the cursor from the point where it last stopped slurping records." }, - "tailTrackCollection": { "index": 14, "kind": "parameter", "displayName": "Tail Track Collection", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Collection where tail tracking information will be persisted. If not specified, MongoDbTailTrackingConfig#DEFAULT_COLLECTION will be used by default." }, - "tailTrackDb": { "index": 15, "kind": "parameter", "displayName": "Tail Track Db", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Indicates what database the tail tracking mechanism will persist to. If not specified, the current database will be picked by default. Dynamicity will not be taken into account even if enabled, i.e., the tail tracking database will not vary past endpoint initialization." }, - "tailTrackField": { "index": 16, "kind": "parameter", "displayName": "Tail Track Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Field where the last tracked value will be placed. If not specified, MongoDbTailTrackingConfig#DEFAULT_FIELD will be used by default." }, - "tailTrackIncreasingField": { "index": 17, "kind": "parameter", "displayName": "Tail Track Increasing Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Correlation field in the incoming record which is of increasing nature and will be used to position the tailing cursor every time it is generated. The cursor will be (re)created with a query of type: tailTrackIncreasingField greater than lastValue (possibly recovered from persistent tail tracking). Can be of type Integer, Date, String, etc. NOTE: No support for dot notation at the current time, so the field should be at the top level of the document." }, - "bridgeErrorHandler": { "index": 18, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "exceptionHandler": { "index": 19, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, - "exchangePattern": { "index": 20, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, - "lazyStartProducer": { "index": 21, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, - "appName": { "index": 22, "kind": "parameter", "displayName": "App Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the logical name of the application. The application name may be used by the client to identify the application to the server, for use in server logs, slow query logs, and profile collection. Default: null" }, - "compressors": { "index": 23, "kind": "parameter", "displayName": "Compressors", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies one or more compression algorithms that the driver will attempt to use to compress requests sent to the connected MongoDB instance. Possible values include: zlib, snappy, and zstd. Default: null" }, - "connectTimeoutMS": { "index": 24, "kind": "parameter", "displayName": "Connect Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10000, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver waits for a connection to open before timing out. A value of 0 instructs the driver to never time out while waiting for a connection to open. Default: 10000 (10 seconds)" }, - "cursorRegenerationDelay": { "index": 25, "kind": "parameter", "displayName": "Cursor Regeneration Delay", "group": "advanced", "label": "advanced", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "MongoDB tailable cursors will block until new data arrives. If no new data is inserted, after some time the cursor will be automatically freed and closed by the MongoDB server. The client is expected to regenerate the cursor if needed. This value specifies the time to wait before attempting to fetch a new cursor, and if the attempt fails, how long before the next attempt is made. Default value is 1000ms." }, - "directConnection": { "index": 26, "kind": "parameter", "displayName": "Direct Connection", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver must connect to the host directly. Default: false" }, - "dynamicity": { "index": 27, "kind": "parameter", "displayName": "Dynamicity", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether this endpoint will attempt to dynamically resolve the target database and collection from the incoming Exchange properties. Can be used to override at runtime the database and collection specified on the otherwise static endpoint URI. It is disabled by default to boost performance. Enabling it will take a minimal performance hit." }, - "heartbeatFrequencyMS": { "index": 28, "kind": "parameter", "displayName": "Heartbeat Frequency MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "heartbeatFrequencyMS controls when the driver checks the state of the MongoDB deployment. Specify the interval (in milliseconds) between checks, counted from the end of the previous check until the beginning of the next one. Default: Single-threaded drivers: 60 seconds. Multithreaded drivers: 10 seconds." }, - "loadBalanced": { "index": 29, "kind": "parameter", "displayName": "Load Balanced", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If true the driver will assume that it's connecting to MongoDB through a load balancer." }, - "localThresholdMS": { "index": 30, "kind": "parameter", "displayName": "Local Threshold MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 15, "description": "The size (in milliseconds) of the latency window for selecting among multiple suitable MongoDB instances. Default: 15 milliseconds." }, - "maxConnecting": { "index": 31, "kind": "parameter", "displayName": "Max Connecting", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 2, "description": "Specifies the maximum number of connections a pool may be establishing concurrently. Default: 2" }, - "maxIdleTimeMS": { "index": 32, "kind": "parameter", "displayName": "Max Idle Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will allow a pooled connection to idle before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can allow a pooled collection to be idle. Default: 0" }, - "maxLifeTimeMS": { "index": 33, "kind": "parameter", "displayName": "Max Life Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will continue to use a pooled connection before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can keep a pooled connection open. Default: 0" }, - "maxPoolSize": { "index": 34, "kind": "parameter", "displayName": "Max Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 100, "description": "The maximum number of connections in the connection pool. The default value is 100." }, - "maxStalenessSeconds": { "index": 35, "kind": "parameter", "displayName": "Max Staleness Seconds", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": -1, "description": "Specifies, in seconds, how stale a secondary can be before the driver stops communicating with that secondary. The minimum value is either 90 seconds or the heartbeat frequency plus 10 seconds, whichever is greater. For more information, see the server documentation for the maxStalenessSeconds option. Not providing a parameter or explicitly specifying -1 indicates that there should be no staleness check for secondaries. Default: -1" }, - "minPoolSize": { "index": 36, "kind": "parameter", "displayName": "Min Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the minimum number of connections that must exist at any moment in a single connection pool. Default: 0" }, - "readPreference": { "index": 37, "kind": "parameter", "displayName": "Read Preference", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "PRIMARY", "PRIMARY_PREFERRED", "SECONDARY", "SECONDARY_PREFERRED", "NEAREST" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PRIMARY", "description": "Configure how MongoDB clients route read operations to the members of a replica set. Possible values are PRIMARY, PRIMARY_PREFERRED, SECONDARY, SECONDARY_PREFERRED or NEAREST" }, - "readPreferenceTags": { "index": 38, "kind": "parameter", "displayName": "Read Preference Tags", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "A representation of a tag set as a comma-separated list of colon-separated key-value pairs, e.g. dc:ny,rack:1. Spaces are stripped from the beginning and end of all keys and values. To specify a list of tag sets, using multiple readPreferenceTags, e.g., readPreferenceTags=dc:ny,rack:1;readPreferenceTags=dc:ny;readPreferenceTags= Note the empty value for the last one, which means match any secondary as a last resort. Order matters when using multiple readPreferenceTags." }, - "replicaSet": { "index": 39, "kind": "parameter", "displayName": "Replica Set", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies that the connection string provided includes multiple hosts. When specified, the driver attempts to find all members of that set." }, - "retryReads": { "index": 40, "kind": "parameter", "displayName": "Retry Reads", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported read operations if they fail due to a network error. Default: true" }, - "retryWrites": { "index": 41, "kind": "parameter", "displayName": "Retry Writes", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported write operations if they fail due to a network error. Default: true" }, - "serverSelectionTimeoutMS": { "index": 42, "kind": "parameter", "displayName": "Server Selection Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "Specifies how long (in milliseconds) to block for server selection before throwing an exception. Default: 30,000 milliseconds." }, - "socketTimeoutMS": { "index": 43, "kind": "parameter", "displayName": "Socket Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will wait to send or receive a request before timing out. A value of 0 instructs the driver to never time out while waiting to send or receive a request. Default: 0" }, - "srvMaxHosts": { "index": 44, "kind": "parameter", "displayName": "Srv Max Hosts", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "The maximum number of hosts from the SRV record to connect to." }, - "srvServiceName": { "index": 45, "kind": "parameter", "displayName": "Srv Service Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "mongodb", "description": "Specifies the service name of the SRV resource recordsthe driver retrieves to construct your seed list. You must use the DNS Seed List Connection Format in your connection URI to use this option. Default: mongodb" }, - "waitQueueTimeoutMS": { "index": 46, "kind": "parameter", "displayName": "Wait Queue Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 120000, "description": "Specifies the maximum amount of time, in milliseconds that a thread may wait for a connection to become available. Default: 120000 (120 seconds)" }, - "writeConcern": { "index": 47, "kind": "parameter", "displayName": "Write Concern", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "ACKNOWLEDGED", "W1", "W2", "W3", "UNACKNOWLEDGED", "JOURNALED", "MAJORITY" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ACKNOWLEDGED", "description": "Configure the connection bean with the level of acknowledgment requested from MongoDB for write operations to a standalone mongod, replicaset or cluster. Possible values are ACKNOWLEDGED, W1, W2, W3, UNACKNOWLEDGED, JOURNALED or MAJORITY." }, - "writeResultAsHeader": { "index": 48, "kind": "parameter", "displayName": "Write Result As Header", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "In write operations, it determines whether instead of returning WriteResult as the body of the OUT message, we transfer the IN message to the OUT and attach the WriteResult as a header." }, - "zlibCompressionLevel": { "index": 49, "kind": "parameter", "displayName": "Zlib Compression Level", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies the degree of compression that Zlib should use to decrease the size of requests to the connected MongoDB instance. The level can range from -1 to 9, with lower values compressing faster (but resulting in larger requests) and larger values compressing slower (but resulting in smaller requests). Default: null" }, - "streamFilter": { "index": 50, "kind": "parameter", "displayName": "Stream Filter", "group": "changeStream", "label": "consumer,changeStream", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Filter condition for change streams consumer." }, - "authSource": { "index": 51, "kind": "parameter", "displayName": "Auth Source", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The database name associated with the user's credentials." }, - "password": { "index": 52, "kind": "parameter", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "User password for mongodb connection" }, - "sslContextParameters": { "index": 53, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "SSL configuration using a Camel SSLContextParameters object. When configured, TLS is automatically enabled on the connection." }, - "tls": { "index": 54, "kind": "parameter", "displayName": "Tls", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that all communication with MongoDB instances should use TLS. Supersedes the ssl option. Default: false" }, - "tlsAllowInvalidHostnames": { "index": 55, "kind": "parameter", "displayName": "Tls Allow Invalid Hostnames", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver should allow invalid hostnames in the certificate for TLS connections. Supersedes sslInvalidHostNameAllowed. Has the same effect as tlsInsecure by setting tlsAllowInvalidHostnames to true. Default: false" }, - "username": { "index": 56, "kind": "parameter", "displayName": "Username", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Username for mongodb connection" } + "changeStreamToken": { "index": 10, "kind": "parameter", "displayName": "Change Stream Token", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "A change stream token as a serialized BSON document JSON string." }, + "changeStreamTokenRepository": { "index": 11, "kind": "parameter", "displayName": "Change Stream Token Repository", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.spi.StateRepository", "deprecated": false, "autowired": false, "secret": false, "description": "The repository to store change stream tokens." }, + "consumerType": { "index": 12, "kind": "parameter", "displayName": "Consumer Type", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "tailable", "changeStreams" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "tailable", "description": "Consumer type." }, + "fullDocument": { "index": 13, "kind": "parameter", "displayName": "Full Document", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "com.mongodb.client.model.changestream.FullDocument", "enum": [ "default", "updateLookup", "required", "whenAvailable" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "default", "description": "Specifies whether changeStream consumer include a copy of the full document when modified by update operations. Possible values are default, updateLookup, required and whenAvailable." }, + "persistentId": { "index": 14, "kind": "parameter", "displayName": "Persistent Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "One tail tracking collection can host many trackers for several tailable consumers. To keep them separate, each tracker should have its own unique persistentId." }, + "persistentTailTracking": { "index": 15, "kind": "parameter", "displayName": "Persistent Tail Tracking", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable persistent tail tracking, which is a mechanism to keep track of the last consumed message across system restarts. The next time the system is up, the endpoint will recover the cursor from the point where it last stopped slurping records." }, + "tailTrackCollection": { "index": 16, "kind": "parameter", "displayName": "Tail Track Collection", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Collection where tail tracking information will be persisted. If not specified, MongoDbTailTrackingConfig#DEFAULT_COLLECTION will be used by default." }, + "tailTrackDb": { "index": 17, "kind": "parameter", "displayName": "Tail Track Db", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Indicates what database the tail tracking mechanism will persist to. If not specified, the current database will be picked by default. Dynamicity will not be taken into account even if enabled, i.e., the tail tracking database will not vary past endpoint initialization." }, + "tailTrackField": { "index": 18, "kind": "parameter", "displayName": "Tail Track Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Field where the last tracked value will be placed. If not specified, MongoDbTailTrackingConfig#DEFAULT_FIELD will be used by default." }, + "tailTrackIncreasingField": { "index": 19, "kind": "parameter", "displayName": "Tail Track Increasing Field", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Correlation field in the incoming record which is of increasing nature and will be used to position the tailing cursor every time it is generated. The cursor will be (re)created with a query of type: tailTrackIncreasingField greater than lastValue (possibly recovered from persistent tail tracking). Can be of type Integer, Date, String, etc. NOTE: No support for dot notation at the current time, so the field should be at the top level of the document." }, + "bridgeErrorHandler": { "index": 20, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, + "exceptionHandler": { "index": 21, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." }, + "exchangePattern": { "index": 22, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." }, + "lazyStartProducer": { "index": 23, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." }, + "appName": { "index": 24, "kind": "parameter", "displayName": "App Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the logical name of the application. The application name may be used by the client to identify the application to the server, for use in server logs, slow query logs, and profile collection. Default: null" }, + "compressors": { "index": 25, "kind": "parameter", "displayName": "Compressors", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies one or more compression algorithms that the driver will attempt to use to compress requests sent to the connected MongoDB instance. Possible values include: zlib, snappy, and zstd. Default: null" }, + "connectTimeoutMS": { "index": 26, "kind": "parameter", "displayName": "Connect Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10000, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver waits for a connection to open before timing out. A value of 0 instructs the driver to never time out while waiting for a connection to open. Default: 10000 (10 seconds)" }, + "cursorRegenerationDelay": { "index": 27, "kind": "parameter", "displayName": "Cursor Regeneration Delay", "group": "advanced", "label": "advanced", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000", "description": "MongoDB tailable cursors will block until new data arrives. If no new data is inserted, after some time the cursor will be automatically freed and closed by the MongoDB server. The client is expected to regenerate the cursor if needed. This value specifies the time to wait before attempting to fetch a new cursor, and if the attempt fails, how long before the next attempt is made. Default value is 1000ms." }, + "directConnection": { "index": 28, "kind": "parameter", "displayName": "Direct Connection", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver must connect to the host directly. Default: false" }, + "dynamicity": { "index": 29, "kind": "parameter", "displayName": "Dynamicity", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether this endpoint will attempt to dynamically resolve the target database and collection from the incoming Exchange properties. Can be used to override at runtime the database and collection specified on the otherwise static endpoint URI. It is disabled by default to boost performance. Enabling it will take a minimal performance hit." }, + "heartbeatFrequencyMS": { "index": 30, "kind": "parameter", "displayName": "Heartbeat Frequency MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "heartbeatFrequencyMS controls when the driver checks the state of the MongoDB deployment. Specify the interval (in milliseconds) between checks, counted from the end of the previous check until the beginning of the next one. Default: Single-threaded drivers: 60 seconds. Multithreaded drivers: 10 seconds." }, + "loadBalanced": { "index": 31, "kind": "parameter", "displayName": "Load Balanced", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If true the driver will assume that it's connecting to MongoDB through a load balancer." }, + "localThresholdMS": { "index": 32, "kind": "parameter", "displayName": "Local Threshold MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 15, "description": "The size (in milliseconds) of the latency window for selecting among multiple suitable MongoDB instances. Default: 15 milliseconds." }, + "maxConnecting": { "index": 33, "kind": "parameter", "displayName": "Max Connecting", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 2, "description": "Specifies the maximum number of connections a pool may be establishing concurrently. Default: 2" }, + "maxIdleTimeMS": { "index": 34, "kind": "parameter", "displayName": "Max Idle Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will allow a pooled connection to idle before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can allow a pooled collection to be idle. Default: 0" }, + "maxLifeTimeMS": { "index": 35, "kind": "parameter", "displayName": "Max Life Time MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will continue to use a pooled connection before closing the connection. A value of 0 indicates that there is no upper bound on how long the driver can keep a pooled connection open. Default: 0" }, + "maxPoolSize": { "index": 36, "kind": "parameter", "displayName": "Max Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 100, "description": "The maximum number of connections in the connection pool. The default value is 100." }, + "maxStalenessSeconds": { "index": 37, "kind": "parameter", "displayName": "Max Staleness Seconds", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": -1, "description": "Specifies, in seconds, how stale a secondary can be before the driver stops communicating with that secondary. The minimum value is either 90 seconds or the heartbeat frequency plus 10 seconds, whichever is greater. For more information, see the server documentation for the maxStalenessSeconds option. Not providing a parameter or explicitly specifying -1 indicates that there should be no staleness check for secondaries. Default: -1" }, + "minPoolSize": { "index": 38, "kind": "parameter", "displayName": "Min Pool Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the minimum number of connections that must exist at any moment in a single connection pool. Default: 0" }, + "readPreference": { "index": 39, "kind": "parameter", "displayName": "Read Preference", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "PRIMARY", "PRIMARY_PREFERRED", "SECONDARY", "SECONDARY_PREFERRED", "NEAREST" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PRIMARY", "description": "Configure how MongoDB clients route read operations to the members of a replica set. Possible values are PRIMARY, PRIMARY_PREFERRED, SECONDARY, SECONDARY_PREFERRED or NEAREST" }, + "readPreferenceTags": { "index": 40, "kind": "parameter", "displayName": "Read Preference Tags", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "A representation of a tag set as a comma-separated list of colon-separated key-value pairs, e.g. dc:ny,rack:1. Spaces are stripped from the beginning and end of all keys and values. To specify a list of tag sets, using multiple readPreferenceTags, e.g., readPreferenceTags=dc:ny,rack:1;readPreferenceTags=dc:ny;readPreferenceTags= Note the empty value for the last one, which means match any secondary as a last resort. Order matters when using multiple readPreferenceTags." }, + "replicaSet": { "index": 41, "kind": "parameter", "displayName": "Replica Set", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies that the connection string provided includes multiple hosts. When specified, the driver attempts to find all members of that set." }, + "retryReads": { "index": 42, "kind": "parameter", "displayName": "Retry Reads", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported read operations if they fail due to a network error. Default: true" }, + "retryWrites": { "index": 43, "kind": "parameter", "displayName": "Retry Writes", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies that the driver must retry supported write operations if they fail due to a network error. Default: true" }, + "serverSelectionTimeoutMS": { "index": 44, "kind": "parameter", "displayName": "Server Selection Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "Specifies how long (in milliseconds) to block for server selection before throwing an exception. Default: 30,000 milliseconds." }, + "socketTimeoutMS": { "index": 45, "kind": "parameter", "displayName": "Socket Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies the maximum amount of time, in milliseconds, the Java driver will wait to send or receive a request before timing out. A value of 0 instructs the driver to never time out while waiting to send or receive a request. Default: 0" }, + "srvMaxHosts": { "index": 46, "kind": "parameter", "displayName": "Srv Max Hosts", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "The maximum number of hosts from the SRV record to connect to." }, + "srvServiceName": { "index": 47, "kind": "parameter", "displayName": "Srv Service Name", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "mongodb", "description": "Specifies the service name of the SRV resource recordsthe driver retrieves to construct your seed list. You must use the DNS Seed List Connection Format in your connection URI to use this option. Default: mongodb" }, + "waitQueueTimeoutMS": { "index": 48, "kind": "parameter", "displayName": "Wait Queue Timeout MS", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 120000, "description": "Specifies the maximum amount of time, in milliseconds that a thread may wait for a connection to become available. Default: 120000 (120 seconds)" }, + "writeConcern": { "index": 49, "kind": "parameter", "displayName": "Write Concern", "group": "advanced", "label": "advanced", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "ACKNOWLEDGED", "W1", "W2", "W3", "UNACKNOWLEDGED", "JOURNALED", "MAJORITY" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ACKNOWLEDGED", "description": "Configure the connection bean with the level of acknowledgment requested from MongoDB for write operations to a standalone mongod, replicaset or cluster. Possible values are ACKNOWLEDGED, W1, W2, W3, UNACKNOWLEDGED, JOURNALED or MAJORITY." }, + "writeResultAsHeader": { "index": 50, "kind": "parameter", "displayName": "Write Result As Header", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "In write operations, it determines whether instead of returning WriteResult as the body of the OUT message, we transfer the IN message to the OUT and attach the WriteResult as a header." }, + "zlibCompressionLevel": { "index": 51, "kind": "parameter", "displayName": "Zlib Compression Level", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Specifies the degree of compression that Zlib should use to decrease the size of requests to the connected MongoDB instance. The level can range from -1 to 9, with lower values compressing faster (but resulting in larger requests) and larger values compressing slower (but resulting in smaller requests). Default: null" }, + "streamFilter": { "index": 52, "kind": "parameter", "displayName": "Stream Filter", "group": "changeStream", "label": "consumer,changeStream", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Filter condition for change streams consumer." }, + "authSource": { "index": 53, "kind": "parameter", "displayName": "Auth Source", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The database name associated with the user's credentials." }, + "password": { "index": 54, "kind": "parameter", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "User password for mongodb connection" }, + "sslContextParameters": { "index": 55, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "SSL configuration using a Camel SSLContextParameters object. When configured, TLS is automatically enabled on the connection." }, + "tls": { "index": 56, "kind": "parameter", "displayName": "Tls", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that all communication with MongoDB instances should use TLS. Supersedes the ssl option. Default: false" }, + "tlsAllowInvalidHostnames": { "index": 57, "kind": "parameter", "displayName": "Tls Allow Invalid Hostnames", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies that the driver should allow invalid hostnames in the certificate for TLS connections. Supersedes sslInvalidHostNameAllowed. Has the same effect as tlsInsecure by setting tlsAllowInvalidHostnames to true. Default: false" }, + "username": { "index": 58, "kind": "parameter", "displayName": "Username", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Username for mongodb connection" } } } diff --git a/components/camel-mongodb/src/generated/resources/META-INF/services/org/apache/camel/mongodb-adapter-factory b/components/camel-mongodb/src/generated/resources/META-INF/services/org/apache/camel/mongodb-adapter-factory new file mode 100644 index 0000000000000..4adbf933ff931 --- /dev/null +++ b/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 diff --git a/components/camel-mongodb/src/generated/resources/META-INF/services/org/apache/camel/mongodb-resume-strategy b/components/camel-mongodb/src/generated/resources/META-INF/services/org/apache/camel/mongodb-resume-strategy new file mode 100644 index 0000000000000..a991d3b7c51e3 --- /dev/null +++ b/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 diff --git a/components/camel-mongodb/src/main/docs/mongodb-component.adoc b/components/camel-mongodb/src/main/docs/mongodb-component.adoc index 17b754f2a3c2f..c4517c7b59e8e 100644 --- a/components/camel-mongodb/src/main/docs/mongodb-component.adoc +++ b/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 `/`. +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 diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/ChangeStreamCommitManager.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/ChangeStreamCommitManager.java new file mode 100644 index 0000000000000..efeeb57b7224d --- /dev/null +++ b/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 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; + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/CommitManager.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/CommitManager.java new file mode 100644 index 0000000000000..c12f850acf24b --- /dev/null +++ b/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; +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/CommitManagers.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/CommitManagers.java new file mode 100644 index 0000000000000..53f5df49bf7b9 --- /dev/null +++ b/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); + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java index 044d370887a0a..3cc048c89c800 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java @@ -20,29 +20,60 @@ 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 { + + 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(); } @@ -50,18 +81,47 @@ protected void doStop() throws Exception { endpoint.getCamelContext().getExecutorServiceManager().shutdown(executor); executor = null; } + + if (stopOffsetRepo) { + StateRepository 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 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 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); @@ -69,4 +129,20 @@ protected void doStart() throws Exception { 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(); + } + } diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java index 4be722baa59b1..1fb8cacae69c6 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java @@ -34,6 +34,8 @@ class MongoDbChangeStreamsThread extends MongoAbstractConsumerThread { private List bsonFilter; private BsonDocument resumeToken; + private CommitManager commitManager; + private String resumeTokenKey; MongoDbChangeStreamsThread(MongoDbEndpoint endpoint, MongoDbChangeStreamsConsumer consumer, List bsonFilter) { @@ -43,6 +45,15 @@ class MongoDbChangeStreamsThread extends MongoAbstractConsumerThread { @Override protected void init() { + MongoDbChangeStreamsConsumer changeStreamsConsumer = (MongoDbChangeStreamsConsumer) consumer; + commitManager = CommitManagers.createCommitManager(changeStreamsConsumer, endpoint); + resumeTokenKey = changeStreamsConsumer.getResumeTokenKey(); + + // Strategy-provided resume token has precedence, then endpoint/repository fallback. + resumeToken = changeStreamsConsumer.getStartupResumeToken(); + if (resumeToken == null) { + resumeToken = commitManager.readResumeToken(); + } cursor = initializeCursor(); } @@ -77,8 +88,13 @@ protected void doRun() { ObjectId documentId = dbObj.getDocumentKey().getObjectId(MONGO_ID).getValue(); OperationType operationType = dbObj.getOperationType(); + BsonDocument currentResumeToken = dbObj.getResumeToken(); + exchange.getIn().setHeader(MongoDbConstants.STREAM_OPERATION_TYPE, operationType.getValue()); exchange.getIn().setHeader(MongoDbConstants.MONGO_ID, documentId); + if (currentResumeToken != null) { + exchange.getIn().setHeader(Exchange.OFFSET, MongoDbResumable.of(resumeTokenKey, currentResumeToken)); + } if (operationType == OperationType.DELETE) { exchange.getIn().setBody(new Document(MONGO_ID, documentId)); } @@ -88,10 +104,11 @@ protected void doRun() { log.trace("Sending exchange: {}, ObjectId: {}", exchange, dbObj.getFullDocument().get(MONGO_ID)); } consumer.getProcessor().process(exchange); + this.resumeToken = currentResumeToken; + commitManager.recordResumeToken(currentResumeToken); + commitManager.commit(); } catch (Exception ignored) { } - - this.resumeToken = dbObj.getResumeToken(); } } catch (MongoException e) { // cursor.hasNext() opens socket and waiting for data diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java index a22ce4e723d53..97949cd2eb88d 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java @@ -42,6 +42,7 @@ import org.apache.camel.Producer; import org.apache.camel.spi.EndpointServiceLocation; import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.StateRepository; import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.spi.UriPath; @@ -195,6 +196,10 @@ public class MongoDbEndpoint extends DefaultEndpoint implements EndpointServiceL @UriParam(label = "advanced") private boolean loadBalanced; //additional properties + @UriParam(label = "consumer") + private StateRepository changeStreamTokenRepository; + @UriParam(label = "consumer") + private String changeStreamToken; @UriParam(description = "Set the whole Connection String/Uri for mongodb endpoint.", label = "common") private String connectionUriString; @@ -1179,6 +1184,28 @@ public boolean isLoadBalanced() { return loadBalanced; } + /** + * The repository to store change stream tokens. + */ + public void setChangeStreamTokenRepository(StateRepository changeStreamTokenRepository) { + this.changeStreamTokenRepository = changeStreamTokenRepository; + } + + public StateRepository getChangeStreamTokenRepository() { + return changeStreamTokenRepository; + } + + /** + * A change stream token as a serialized BSON document JSON string. + */ + public void setChangeStreamToken(String changeStreamToken) { + this.changeStreamToken = changeStreamToken; + } + + public String getChangeStreamToken() { + return changeStreamToken; + } + /** * Set the whole Connection String/Uri for mongodb endpoint. To be flexible and future proof about supporting all * the mongodb client options diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbResumable.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbResumable.java new file mode 100644 index 0000000000000..e8fa66b6fbdbf --- /dev/null +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbResumable.java @@ -0,0 +1,48 @@ +/* + * 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.resume.Offset; +import org.apache.camel.resume.OffsetKey; +import org.apache.camel.resume.Resumable; +import org.apache.camel.support.resume.OffsetKeys; +import org.apache.camel.support.resume.Offsets; +import org.bson.BsonDocument; + +public final class MongoDbResumable implements Resumable { + private final String addressable; + private final String resumeToken; + + private MongoDbResumable(String addressable, String resumeToken) { + this.addressable = addressable; + this.resumeToken = resumeToken; + } + + @Override + public OffsetKey getOffsetKey() { + return OffsetKeys.unmodifiableOf(addressable); + } + + @Override + public Offset getLastOffset() { + return Offsets.of(resumeToken); + } + + public static MongoDbResumable of(String addressable, BsonDocument resumeToken) { + return new MongoDbResumable(addressable, resumeToken.toJson()); + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbResumeAdapter.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbResumeAdapter.java new file mode 100644 index 0000000000000..8c866453deefb --- /dev/null +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbResumeAdapter.java @@ -0,0 +1,84 @@ +/* + * 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.resume.Cacheable; +import org.apache.camel.resume.Offset; +import org.apache.camel.resume.OffsetKey; +import org.apache.camel.resume.ResumeAdapter; +import org.apache.camel.resume.cache.ResumeCache; +import org.apache.camel.spi.annotations.JdkService; +import org.apache.camel.util.ObjectHelper; +import org.bson.BsonDocument; + +@JdkService("mongodb-adapter-factory") +public class MongoDbResumeAdapter implements ResumeAdapter, Cacheable { + private ResumeCache resumeCache; + private String resumeTokenKey; + private MongoDbChangeStreamsConsumer consumer; + + @Override + public void resume() { + if (resumeCache == null || ObjectHelper.isEmpty(resumeTokenKey) || consumer == null) { + return; + } + + String serializedToken = resumeCache.get(resumeTokenKey, String.class); + if (ObjectHelper.isNotEmpty(serializedToken)) { + consumer.setStartupResumeToken(BsonDocument.parse(serializedToken)); + } + } + + @SuppressWarnings("unchecked") + @Override + public void setCache(ResumeCache cache) { + this.resumeCache = (ResumeCache) cache; + } + + @Override + public ResumeCache getCache() { + return resumeCache; + } + + @Override + public boolean add(OffsetKey key, Offset offset) { + if (resumeCache == null) { + return false; + } + + String keyValue = String.valueOf(key.getValue()); + String offsetValue = offset.getValue(String.class); + if (ObjectHelper.isNotEmpty(keyValue) && ObjectHelper.isNotEmpty(offsetValue)) { + resumeCache.add(keyValue, offsetValue); + return true; + } + + return false; + } + + public void setResumeTokenKey(String resumeTokenKey) { + this.resumeTokenKey = resumeTokenKey; + } + + public String getResumeTokenKey() { + return resumeTokenKey; + } + + public void setConsumer(MongoDbChangeStreamsConsumer consumer) { + this.consumer = consumer; + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/NoopCommitManager.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/NoopCommitManager.java new file mode 100644 index 0000000000000..e1d0bacd22ccf --- /dev/null +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/NoopCommitManager.java @@ -0,0 +1,48 @@ +/* + * 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.util.ObjectHelper; +import org.bson.BsonDocument; + +public class NoopCommitManager implements CommitManager { + + private final MongoDbEndpoint endpoint; + + public NoopCommitManager(MongoDbEndpoint endpoint) { + this.endpoint = endpoint; + } + + @Override + public BsonDocument readResumeToken() { + if (ObjectHelper.isEmpty(endpoint.getChangeStreamToken())) { + return null; + } + + return BsonDocument.parse(endpoint.getChangeStreamToken()); + } + + @Override + public void recordResumeToken(BsonDocument resumeToken) { + // NO-OP + } + + @Override + public void commit() { + // NO-OP + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategy.java b/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategy.java new file mode 100644 index 0000000000000..7028ee8ab0b9a --- /dev/null +++ b/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategy.java @@ -0,0 +1,161 @@ +/* + * 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.processor.resume.mongodb; + +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.resume.Offset; +import org.apache.camel.resume.OffsetKey; +import org.apache.camel.resume.Resumable; +import org.apache.camel.resume.ResumeAdapter; +import org.apache.camel.resume.ResumeStrategy; +import org.apache.camel.resume.ResumeStrategyConfiguration; +import org.apache.camel.resume.cache.ResumeCache; +import org.apache.camel.spi.StateRepository; +import org.apache.camel.spi.annotations.JdkService; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.apache.camel.util.ObjectHelper; + +@JdkService("mongodb-resume-strategy") +public class MongoDbResumeStrategy implements ResumeStrategy { + private ResumeAdapter adapter; + private MongoDbResumeStrategyConfiguration configuration = new MongoDbResumeStrategyConfiguration(); + private boolean stopRepository; + + @Override + public void start() { + StateRepository repository = configuration.getStateRepository(); + if (repository instanceof ServiceSupport serviceSupport && !serviceSupport.isStarted()) { + stopRepository = true; + try { + ServiceHelper.startService(repository); + } catch (Exception e) { + throw new RuntimeCamelException("Unable to start MongoDB resume state repository", e); + } + } + } + + @Override + public void stop() { + if (stopRepository) { + try { + ServiceHelper.stopAndShutdownService(configuration.getStateRepository()); + } catch (Exception e) { + throw new RuntimeCamelException("Unable to stop MongoDB resume state repository", e); + } + } + } + + @Override + public void setAdapter(ResumeAdapter adapter) { + this.adapter = adapter; + } + + @Override + public ResumeAdapter getAdapter() { + return adapter; + } + + @Override + public void loadCache() { + ResumeCache resumeCache = getStringResumeCache(); + StateRepository repository = configuration.getStateRepository(); + + if (resumeCache == null || repository == null) { + return; + } + + if (adapter instanceof org.apache.camel.component.mongodb.MongoDbResumeAdapter mongoDbResumeAdapter) { + String key = mongoDbResumeAdapter.getResumeTokenKey(); + if (ObjectHelper.isNotEmpty(key)) { + String value = repository.getState(key); + if (ObjectHelper.isNotEmpty(value)) { + resumeCache.add(key, value); + } + } + } + } + + @Override + public void updateLastOffset(T offset) { + String key = String.valueOf(offset.getOffsetKey().getValue()); + String value = offset.getLastOffset().getValue(String.class); + updateStoredOffset(key, value); + } + + @Override + public void updateLastOffset(T offset, UpdateCallBack updateCallBack) { + updateLastOffset(offset); + } + + @Override + public void updateLastOffset(OffsetKey offsetKey, Offset offsetValue) { + String key = String.valueOf(offsetKey.getValue()); + String value = offsetValue.getValue(String.class); + updateStoredOffset(key, value); + } + + @Override + public void updateLastOffset(OffsetKey offsetKey, Offset offset, UpdateCallBack updateCallBack) { + updateLastOffset(offsetKey, offset); + } + + @Override + public void setResumeStrategyConfiguration(ResumeStrategyConfiguration resumeStrategyConfiguration) { + this.configuration = (MongoDbResumeStrategyConfiguration) resumeStrategyConfiguration; + } + + @Override + public ResumeStrategyConfiguration getResumeStrategyConfiguration() { + return configuration; + } + + public String getLastOffset(String key) { + StateRepository repository = configuration.getStateRepository(); + if (repository != null) { + return repository.getState(key); + } + + ResumeCache resumeCache = getStringResumeCache(); + if (resumeCache != null) { + return resumeCache.get(key, String.class); + } + + return null; + } + + private void updateStoredOffset(String key, String value) { + if (ObjectHelper.isEmpty(key) || ObjectHelper.isEmpty(value)) { + return; + } + + StateRepository repository = configuration.getStateRepository(); + if (repository != null) { + repository.setState(key, value); + } + + ResumeCache resumeCache = getStringResumeCache(); + if (resumeCache != null) { + resumeCache.add(key, value); + } + } + + @SuppressWarnings("unchecked") + private ResumeCache getStringResumeCache() { + return (ResumeCache) configuration.getResumeCache(); + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyConfiguration.java b/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyConfiguration.java new file mode 100644 index 0000000000000..83f9fb8ffd05c --- /dev/null +++ b/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyConfiguration.java @@ -0,0 +1,37 @@ +/* + * 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.processor.resume.mongodb; + +import org.apache.camel.resume.ResumeStrategyConfiguration; +import org.apache.camel.spi.StateRepository; + +public class MongoDbResumeStrategyConfiguration extends ResumeStrategyConfiguration { + private StateRepository stateRepository; + + @Override + public String resumeStrategyService() { + return "mongodb-resume-strategy"; + } + + public StateRepository getStateRepository() { + return stateRepository; + } + + public void setStateRepository(StateRepository stateRepository) { + this.stateRepository = stateRepository; + } +} diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyConfigurationBuilder.java b/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyConfigurationBuilder.java new file mode 100644 index 0000000000000..d0741c2f395fb --- /dev/null +++ b/components/camel-mongodb/src/main/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyConfigurationBuilder.java @@ -0,0 +1,47 @@ +/* + * 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.processor.resume.mongodb; + +import org.apache.camel.spi.StateRepository; +import org.apache.camel.support.resume.BasicResumeStrategyConfigurationBuilder; + +public class MongoDbResumeStrategyConfigurationBuilder + extends + BasicResumeStrategyConfigurationBuilder { + + private StateRepository stateRepository; + + private MongoDbResumeStrategyConfigurationBuilder() { + } + + public static MongoDbResumeStrategyConfigurationBuilder newBuilder() { + return new MongoDbResumeStrategyConfigurationBuilder(); + } + + public MongoDbResumeStrategyConfigurationBuilder withStateRepository(StateRepository stateRepository) { + this.stateRepository = stateRepository; + return this; + } + + @Override + public MongoDbResumeStrategyConfiguration build() { + MongoDbResumeStrategyConfiguration configuration = new MongoDbResumeStrategyConfiguration(); + buildCommonConfiguration(configuration); + configuration.setStateRepository(stateRepository); + return configuration; + } +} diff --git a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/ChangeStreamCommitManagerTest.java b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/ChangeStreamCommitManagerTest.java new file mode 100644 index 0000000000000..4d086e821de20 --- /dev/null +++ b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/ChangeStreamCommitManagerTest.java @@ -0,0 +1,74 @@ +/* + * 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.CamelContext; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.processor.state.MemoryStateRepository; +import org.bson.BsonDocument; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class ChangeStreamCommitManagerTest { + + private static MongoDbEndpoint createEndpoint() { + CamelContext context = new DefaultCamelContext(); + MongoDbComponent component = new MongoDbComponent(context); + MongoDbEndpoint endpoint = new MongoDbEndpoint("mongodb:myDb", component); + endpoint.setCollection("camelTest"); + return endpoint; + } + + @Test + public void shouldReadExplicitResumeTokenBeforeRepositoryToken() { + MemoryStateRepository repository = new MemoryStateRepository(); + repository.setState("mongodb:myDb/camelTest", "{\"_data\":\"repo-token\"}"); + + MongoDbEndpoint endpoint = createEndpoint(); + endpoint.setChangeStreamTokenRepository(repository); + endpoint.setChangeStreamToken("{\"_data\":\"explicit-token\"}"); + + MongoDbChangeStreamsConsumer consumer = new MongoDbChangeStreamsConsumer(endpoint, e -> { + }); + ChangeStreamCommitManager manager = new ChangeStreamCommitManager(consumer, endpoint); + + BsonDocument token = manager.readResumeToken(); + assertNotNull(token); + assertEquals("explicit-token", token.getString("_data").getValue()); + } + + @Test + public void shouldCommitSerializedTokenToRepository() throws Exception { + MemoryStateRepository repository = new MemoryStateRepository(); + + MongoDbEndpoint endpoint = createEndpoint(); + endpoint.setChangeStreamTokenRepository(repository); + + MongoDbChangeStreamsConsumer consumer = new MongoDbChangeStreamsConsumer(endpoint, e -> { + }); + + ChangeStreamCommitManager manager = new ChangeStreamCommitManager(consumer, endpoint); + BsonDocument token = BsonDocument.parse("{\"_data\":\"new-token\"}"); + + manager.recordResumeToken(token); + manager.commit(); + + assertEquals(token.toJson(), repository.getState("mongodb:myDb/camelTest")); + } +} diff --git a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbResumeAdapterTest.java b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbResumeAdapterTest.java new file mode 100644 index 0000000000000..a95dc5051fa55 --- /dev/null +++ b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbResumeAdapterTest.java @@ -0,0 +1,75 @@ +/* + * 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.CamelContext; +import org.apache.camel.component.mongodb.support.InMemoryStringResumeCache; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.processor.resume.mongodb.MongoDbResumeStrategy; +import org.apache.camel.processor.resume.mongodb.MongoDbResumeStrategyConfigurationBuilder; +import org.apache.camel.resume.ResumeAdapter; +import org.apache.camel.support.processor.state.MemoryStateRepository; +import org.apache.camel.support.resume.ResumeStrategyHelper; +import org.bson.BsonDocument; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class MongoDbResumeAdapterTest { + + private static MongoDbEndpoint createEndpoint(CamelContext context) { + MongoDbComponent component = new MongoDbComponent(context); + MongoDbEndpoint endpoint = new MongoDbEndpoint("mongodb:myDb", component); + endpoint.setCollection("camelTest"); + return endpoint; + } + + @Test + public void shouldSetStartupTokenFromStrategyCacheLoad() throws Exception { + CamelContext context = new DefaultCamelContext(); + MongoDbEndpoint endpoint = createEndpoint(context); + MongoDbChangeStreamsConsumer consumer = new MongoDbChangeStreamsConsumer(endpoint, e -> { + }); + + String key = endpoint.getEndpointUri() + "/camelTest"; + String expected = "{\"_data\":\"strategy-token\"}"; + InMemoryStringResumeCache resumeCache = new InMemoryStringResumeCache(); + + MemoryStateRepository repository = new MemoryStateRepository(); + repository.setState(key, expected); + + MongoDbResumeStrategy strategy = new MongoDbResumeStrategy(); + strategy.setResumeStrategyConfiguration(MongoDbResumeStrategyConfigurationBuilder.newBuilder() + .withStateRepository(repository) + .withResumeCache(resumeCache) + .build()); + + MongoDbResumeAdapter adapter = new MongoDbResumeAdapter(); + adapter.setCache(resumeCache); + adapter.setConsumer(consumer); + adapter.setResumeTokenKey(key); + + strategy.setAdapter(adapter); + + ResumeStrategyHelper.resume(context, consumer, strategy, "mongodb-resume", ResumeAdapter.class); + + BsonDocument startupToken = consumer.getStartupResumeToken(); + assertNotNull(startupToken); + assertEquals("strategy-token", startupToken.getString("_data").getValue()); + } +} diff --git a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbChangeStreamsResumeIT.java b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbChangeStreamsResumeIT.java new file mode 100644 index 0000000000000..d0753d77ec585 --- /dev/null +++ b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbChangeStreamsResumeIT.java @@ -0,0 +1,103 @@ +/* + * 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.integration; + +import com.mongodb.client.MongoCollection; +import com.mongodb.client.model.CreateCollectionOptions; +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.support.processor.state.MemoryStateRepository; +import org.apache.camel.test.infra.core.annotations.RouteFixture; +import org.apache.camel.test.infra.core.api.ConfigurableRoute; +import org.bson.Document; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class MongoDbChangeStreamsResumeIT extends AbstractMongoDbITSupport implements ConfigurableRoute { + + private final MemoryStateRepository resumeTokenRepository = new MemoryStateRepository(); + private static final String ROUTE_ID = "resumeChangeStreamConsumer"; + private MongoCollection mongoCollection; + + @AfterEach + protected void doPostSetup() { + super.doPostSetup(); + + mongoCollection = db.getCollection(AbstractMongoDbITSupport.testCollectionName, Document.class); + mongoCollection.drop(); + + CreateCollectionOptions collectionOptions = new CreateCollectionOptions(); + db.createCollection(AbstractMongoDbITSupport.testCollectionName, collectionOptions); + mongoCollection = db.getCollection(AbstractMongoDbITSupport.testCollectionName, Document.class); + } + + @Test + public void shouldResumeAfterRestartFromStoredResumeToken() throws Exception { + mongoCollection = db.getCollection(AbstractMongoDbITSupport.testCollectionName, Document.class); + resumeTokenRepository.start(); + + MockEndpoint mock = contextExtension.getMockEndpoint("mock:test"); + mock.expectedMessageCount(2); + + context.getRouteController().startRoute(ROUTE_ID); + + mongoCollection.insertOne(new Document("increasing", 0).append("string", "value0")); + mongoCollection.insertOne(new Document("increasing", 1).append("string", "value1")); + + mock.assertIsSatisfied(); + context.getRouteController().stopRoute(ROUTE_ID); + + String key = ROUTE_ID + '/' + AbstractMongoDbITSupport.testCollectionName; + String storedToken = resumeTokenRepository.getState(key); + assertNotNull(storedToken); + + // Insert while the route is stopped. The resume token should make these consumable on restart. + mongoCollection.insertOne(new Document("increasing", 2).append("string", "value2")); + mongoCollection.insertOne(new Document("increasing", 3).append("string", "value3")); + + mock.reset(); + mock.expectedMessageCount(2); + + context.getRouteController().startRoute(ROUTE_ID); + mock.assertIsSatisfied(); + context.getRouteController().stopRoute(ROUTE_ID); + + assertEquals("value2", mock.getExchanges().get(0).getIn().getBody(Document.class).getString("string")); + assertEquals("value3", mock.getExchanges().get(1).getIn().getBody(Document.class).getString("string")); + } + + @RouteFixture + @Override + public void createRouteBuilder(CamelContext context) throws Exception { + context.getRegistry().bind("changeStreamTokenRepo", resumeTokenRepository); + + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("mongodb:myDb?consumerType=changeStreams&database={{mongodb.testDb}}" + + "&collection={{mongodb.testCollection}}&changeStreamTokenRepository=#changeStreamTokenRepo") + .id(ROUTE_ID) + .autoStartup(false) + .to("mock:test"); + } + }); + } +} diff --git a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java new file mode 100644 index 0000000000000..c113c26bf6e96 --- /dev/null +++ b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java @@ -0,0 +1,78 @@ +/* + * 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.support; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.BiFunction; +import java.util.function.Function; + +import org.apache.camel.resume.cache.ResumeCache; + +public class InMemoryStringResumeCache implements ResumeCache { + private final Map data = new HashMap<>(); + + @Override + public Object computeIfAbsent(String key, Function mapping) { + return data.computeIfAbsent(key, mapping::apply); + } + + @Override + public Object computeIfPresent(String key, BiFunction remapping) { + return data.computeIfPresent(key, remapping::apply); + } + + @Override + public boolean contains(String key, Object entry) { + return Objects.equals(data.get(key), entry); + } + + @Override + public void add(String key, Object offsetValue) { + data.put(key, offsetValue); + } + + @Override + public boolean isFull() { + return false; + } + + @Override + public long capacity() { + return Long.MAX_VALUE; + } + + @Override + public T get(String key, Class clazz) { + return clazz.cast(data.get(key)); + } + + @Override + public Object get(String key) { + return data.get(key); + } + + @Override + public void forEach(BiFunction action) { + for (Map.Entry entry : data.entrySet()) { + if (!action.apply(entry.getKey(), entry.getValue())) { + break; + } + } + } +} diff --git a/components/camel-mongodb/src/test/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyTest.java b/components/camel-mongodb/src/test/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyTest.java new file mode 100644 index 0000000000000..a83b75d8e2ea2 --- /dev/null +++ b/components/camel-mongodb/src/test/java/org/apache/camel/processor/resume/mongodb/MongoDbResumeStrategyTest.java @@ -0,0 +1,72 @@ +/* + * 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.processor.resume.mongodb; + +import org.apache.camel.component.mongodb.MongoDbResumable; +import org.apache.camel.component.mongodb.support.InMemoryStringResumeCache; +import org.apache.camel.resume.Cacheable; +import org.apache.camel.support.processor.state.MemoryStateRepository; +import org.bson.BsonDocument; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +public class MongoDbResumeStrategyTest { + + @Test + public void shouldStoreAndLoadOffsetsUsingStateRepository() throws Exception { + MemoryStateRepository repository = new MemoryStateRepository(); + + MongoDbResumeStrategyConfiguration configuration = MongoDbResumeStrategyConfigurationBuilder.newBuilder() + .withStateRepository(repository) + .build(); + MongoDbResumeStrategy strategy = new MongoDbResumeStrategy(); + strategy.setResumeStrategyConfiguration(configuration); + + String key = "routeA/testCollection"; + BsonDocument token = BsonDocument.parse("{\"_data\":\"token-1\"}"); + + strategy.updateLastOffset(MongoDbResumable.of(key, token)); + + assertEquals(token.toJson(), repository.getState(key)); + assertEquals(token.toJson(), strategy.getLastOffset(key)); + } + + @Test + public void shouldStoreAndLoadOffsetsUsingResumeCache() throws Exception { + InMemoryStringResumeCache resumeCache = new InMemoryStringResumeCache(); + + MongoDbResumeStrategyConfiguration configuration = MongoDbResumeStrategyConfigurationBuilder.newBuilder() + .withResumeCache(resumeCache) + .withCacheFillPolicy(Cacheable.FillPolicy.MAXIMIZING) + .build(); + + MongoDbResumeStrategy strategy = new MongoDbResumeStrategy(); + strategy.setResumeStrategyConfiguration(configuration); + + String key = "routeB/testCollection"; + BsonDocument token = BsonDocument.parse("{\"_data\":\"token-2\"}"); + + strategy.updateLastOffset(MongoDbResumable.of(key, token)); + + assertEquals(token.toJson(), resumeCache.get(key, String.class)); + assertEquals(token.toJson(), strategy.getLastOffset(key)); + assertNull(strategy.getLastOffset("missing")); + } + +} diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc index 4d11ab735f8f0..8bfdcdb7434da 100644 --- a/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc +++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc @@ -23,6 +23,7 @@ Support for resume varies according to the component. Initially, the support is * xref:components::couchdb-component.adoc[camel-couchdb] * xref:components::file-component.adoc[camel-file] * xref:components::kafka-component.adoc[camel-kafka] +* xref:components::mongodb-component.adoc[camel-mongodb] * xref:components::rss-component.adoc[camel-rss] The resume strategies comes in three parts: diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MongoDbEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MongoDbEndpointBuilderFactory.java index fec25ddc101dd..93b98e3d4b50a 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MongoDbEndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MongoDbEndpointBuilderFactory.java @@ -248,6 +248,52 @@ default MongoDbEndpointConsumerBuilder outputType(String outputType) { doSetProperty("outputType", outputType); return this; } + /** + * A change stream token as a serialized BSON document JSON string. + * + * The option is a: java.lang.String type. + * + * Group: consumer + * + * @param changeStreamToken the value to set + * @return the dsl builder + */ + default MongoDbEndpointConsumerBuilder changeStreamToken(String changeStreamToken) { + doSetProperty("changeStreamToken", changeStreamToken); + return this; + } + /** + * The repository to store change stream tokens. + * + * The option is a: + * org.apache.camel.spi.StateRepository<java.lang.String, + * java.lang.String> type. + * + * Group: consumer + * + * @param changeStreamTokenRepository the value to set + * @return the dsl builder + */ + default MongoDbEndpointConsumerBuilder changeStreamTokenRepository(org.apache.camel.spi.StateRepository changeStreamTokenRepository) { + doSetProperty("changeStreamTokenRepository", changeStreamTokenRepository); + return this; + } + /** + * The repository to store change stream tokens. + * + * The option will be converted to a + * org.apache.camel.spi.StateRepository<java.lang.String, + * java.lang.String> type. + * + * Group: consumer + * + * @param changeStreamTokenRepository the value to set + * @return the dsl builder + */ + default MongoDbEndpointConsumerBuilder changeStreamTokenRepository(String changeStreamTokenRepository) { + doSetProperty("changeStreamTokenRepository", changeStreamTokenRepository); + return this; + } /** * Consumer type. * diff --git a/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties b/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties index 3706ea862f89e..4b6ae8f817900 100644 --- a/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties +++ b/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties @@ -31,6 +31,8 @@ META-INF/services/org/apache/camel/mdc-service=camel:mdc META-INF/services/org/apache/camel/micrometer-observability-tracer=camel:micrometer-observability META-INF/services/org/apache/camel/micrometer-prometheus=camel:micrometer-prometheus META-INF/services/org/apache/camel/mock-send-to-endpoint-strategy-factory=camel:mock +META-INF/services/org/apache/camel/mongodb-adapter-factory=camel:mongodb +META-INF/services/org/apache/camel/mongodb-resume-strategy=camel:mongodb META-INF/services/org/apache/camel/oauth-client-authentication-factory=camel:oauth META-INF/services/org/apache/camel/oauth-token-validation-factory=camel:oauth META-INF/services/org/apache/camel/opentelemetry-tracer-2=camel:opentelemetry2