You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by oa...@apache.org on 2020/03/24 12:44:50 UTC
[camel] branch master updated: CAMEL-14773: Upgrade debezium to
1.1.0.Final and regenerate the configurations
This is an automated email from the ASF dual-hosted git repository.
oalsafi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 49c98cb CAMEL-14773: Upgrade debezium to 1.1.0.Final and regenerate the configurations
new 0f4f79a Merge pull request #3673 from omarsmak/CAMEL-14773
49c98cb is described below
commit 49c98cbb48342332b96814be816079dac8b51058
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Tue Mar 24 13:34:24 2020 +0100
CAMEL-14773: Upgrade debezium to 1.1.0.Final and regenerate the configurations
---
.../DebeziumMongodbComponentConfigurer.java | 5 ++
.../DebeziumMongodbEndpointConfigurer.java | 5 ++
...goDbConnectorEmbeddedDebeziumConfiguration.java | 18 ++++++-
.../camel/component/debezium/debezium-mongodb.json | 6 ++-
.../src/main/docs/debezium-mongodb-component.adoc | 10 ++--
.../debezium/DebeziumMySqlComponentConfigurer.java | 5 ++
.../debezium/DebeziumMySqlEndpointConfigurer.java | 5 ++
...ySqlConnectorEmbeddedDebeziumConfiguration.java | 30 +++++++++--
.../camel/component/debezium/debezium-mysql.json | 10 ++--
.../src/main/docs/debezium-mysql-component.adoc | 14 ++---
.../DebeziumPostgresComponentConfigurer.java | 15 ++++++
.../DebeziumPostgresEndpointConfigurer.java | 15 ++++++
...gresConnectorEmbeddedDebeziumConfiguration.java | 52 ++++++++++++++++++-
.../component/debezium/debezium-postgres.json | 10 +++-
.../src/main/docs/debezium-postgres-component.adoc | 14 +++--
.../DebeziumSqlserverComponentConfigurer.java | 10 ++++
.../DebeziumSqlserverEndpointConfigurer.java | 10 ++++
...rverConnectorEmbeddedDebeziumConfiguration.java | 34 ++++++++++++
.../component/debezium/debezium-sqlserver.json | 4 ++
.../main/docs/debezium-sqlserver-component.adoc | 8 ++-
.../DebeziumMongodbComponentBuilderFactory.java | 17 +++++-
.../dsl/DebeziumMysqlComponentBuilderFactory.java | 27 ++++++++--
.../DebeziumPostgresComponentBuilderFactory.java | 50 +++++++++++++++++-
.../DebeziumSqlserverComponentBuilderFactory.java | 33 ++++++++++++
.../dsl/DebeziumMongodbEndpointBuilderFactory.java | 18 ++++++-
.../dsl/DebeziumMySqlEndpointBuilderFactory.java | 26 ++++++++--
.../DebeziumPostgresEndpointBuilderFactory.java | 60 +++++++++++++++++++++-
.../DebeziumSqlserverEndpointBuilderFactory.java | 44 ++++++++++++++++
parent/pom.xml | 2 +-
29 files changed, 509 insertions(+), 48 deletions(-)
diff --git a/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbComponentConfigurer.java b/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbComponentConfigurer.java
index edbe4b1..7bafd4f 100644
--- a/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbComponentConfigurer.java
+++ b/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbComponentConfigurer.java
@@ -99,6 +99,8 @@ public class DebeziumMongodbComponentConfigurer extends PropertyConfigurerSuppor
case "offsetStorageTopic": getOrCreateConfiguration(target).setOffsetStorageTopic(property(camelContext, java.lang.String.class, value)); return true;
case "pollintervalms":
case "pollIntervalMs": getOrCreateConfiguration(target).setPollIntervalMs(property(camelContext, long.class, value)); return true;
+ case "skippedoperations":
+ case "skippedOperations": getOrCreateConfiguration(target).setSkippedOperations(property(camelContext, java.lang.String.class, value)); return true;
case "snapshotdelayms":
case "snapshotDelayMs": getOrCreateConfiguration(target).setSnapshotDelayMs(property(camelContext, long.class, value)); return true;
case "snapshotfetchsize":
@@ -153,6 +155,7 @@ public class DebeziumMongodbComponentConfigurer extends PropertyConfigurerSuppor
answer.put("offsetStorageReplicationFactor", int.class);
answer.put("offsetStorageTopic", java.lang.String.class);
answer.put("pollIntervalMs", long.class);
+ answer.put("skippedOperations", java.lang.String.class);
answer.put("snapshotDelayMs", long.class);
answer.put("snapshotFetchSize", int.class);
answer.put("snapshotMode", java.lang.String.class);
@@ -238,6 +241,8 @@ public class DebeziumMongodbComponentConfigurer extends PropertyConfigurerSuppor
case "offsetStorageTopic": return getOrCreateConfiguration(target).getOffsetStorageTopic();
case "pollintervalms":
case "pollIntervalMs": return getOrCreateConfiguration(target).getPollIntervalMs();
+ case "skippedoperations":
+ case "skippedOperations": return getOrCreateConfiguration(target).getSkippedOperations();
case "snapshotdelayms":
case "snapshotDelayMs": return getOrCreateConfiguration(target).getSnapshotDelayMs();
case "snapshotfetchsize":
diff --git a/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbEndpointConfigurer.java b/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbEndpointConfigurer.java
index 2c38de6..3d76b39 100644
--- a/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbEndpointConfigurer.java
+++ b/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/DebeziumMongodbEndpointConfigurer.java
@@ -95,6 +95,8 @@ public class DebeziumMongodbEndpointConfigurer extends PropertyConfigurerSupport
case "offsetStorageTopic": target.getConfiguration().setOffsetStorageTopic(property(camelContext, java.lang.String.class, value)); return true;
case "pollintervalms":
case "pollIntervalMs": target.getConfiguration().setPollIntervalMs(property(camelContext, long.class, value)); return true;
+ case "skippedoperations":
+ case "skippedOperations": target.getConfiguration().setSkippedOperations(property(camelContext, java.lang.String.class, value)); return true;
case "snapshotdelayms":
case "snapshotDelayMs": target.getConfiguration().setSnapshotDelayMs(property(camelContext, long.class, value)); return true;
case "snapshotfetchsize":
@@ -151,6 +153,7 @@ public class DebeziumMongodbEndpointConfigurer extends PropertyConfigurerSupport
answer.put("offsetStorageReplicationFactor", int.class);
answer.put("offsetStorageTopic", java.lang.String.class);
answer.put("pollIntervalMs", long.class);
+ answer.put("skippedOperations", java.lang.String.class);
answer.put("snapshotDelayMs", long.class);
answer.put("snapshotFetchSize", int.class);
answer.put("snapshotMode", java.lang.String.class);
@@ -240,6 +243,8 @@ public class DebeziumMongodbEndpointConfigurer extends PropertyConfigurerSupport
case "offsetStorageTopic": return target.getConfiguration().getOffsetStorageTopic();
case "pollintervalms":
case "pollIntervalMs": return target.getConfiguration().getPollIntervalMs();
+ case "skippedoperations":
+ case "skippedOperations": return target.getConfiguration().getSkippedOperations();
case "snapshotdelayms":
case "snapshotDelayMs": return target.getConfiguration().getSnapshotDelayMs();
case "snapshotfetchsize":
diff --git a/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/configuration/MongoDbConnectorEmbeddedDebeziumConfiguration.java b/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/configuration/MongoDbConnectorEmbeddedDebeziumConfiguration.java
index 203e679..5894b31 100644
--- a/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/configuration/MongoDbConnectorEmbeddedDebeziumConfiguration.java
+++ b/components/camel-debezium-mongodb/src/generated/java/org/apache/camel/component/debezium/configuration/MongoDbConnectorEmbeddedDebeziumConfiguration.java
@@ -66,6 +66,8 @@ public class MongoDbConnectorEmbeddedDebeziumConfiguration
private boolean mongodbSslInvalidHostnameAllowed = false;
@UriParam(label = LABEL_NAME)
private String databaseBlacklist;
+ @UriParam(label = LABEL_NAME)
+ private String skippedOperations;
@UriParam(label = LABEL_NAME, defaultValue = "2048")
private int maxBatchSize = 2048;
@UriParam(label = LABEL_NAME, defaultValue = "initial")
@@ -96,7 +98,7 @@ public class MongoDbConnectorEmbeddedDebeziumConfiguration
}
/**
- * Maximum number of threads used to perform an intial sync of the
+ * Maximum number of threads used to perform an initial sync of the
* collections in a replica set. Defaults to 1.
*/
public void setInitialSyncMaxThreads(int initialSyncMaxThreads) {
@@ -393,6 +395,19 @@ public class MongoDbConnectorEmbeddedDebeziumConfiguration
}
/**
+ * The comma-separated list of operations to skip during streaming, defined
+ * as: 'i' for inserts; 'u' for updates; 'd' for deletes. By default, no
+ * operations will be skipped.
+ */
+ public void setSkippedOperations(String skippedOperations) {
+ this.skippedOperations = skippedOperations;
+ }
+
+ public String getSkippedOperations() {
+ return skippedOperations;
+ }
+
+ /**
* Maximum size of each batch of source records. Defaults to 2048.
*/
public void setMaxBatchSize(int maxBatchSize) {
@@ -447,6 +462,7 @@ public class MongoDbConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "connect.backoff.max.delay.ms", connectBackoffMaxDelayMs);
addPropertyIfNotNull(configBuilder, "mongodb.ssl.invalid.hostname.allowed", mongodbSslInvalidHostnameAllowed);
addPropertyIfNotNull(configBuilder, "database.blacklist", databaseBlacklist);
+ addPropertyIfNotNull(configBuilder, "skipped.operations", skippedOperations);
addPropertyIfNotNull(configBuilder, "max.batch.size", maxBatchSize);
addPropertyIfNotNull(configBuilder, "snapshot.mode", snapshotMode);
diff --git a/components/camel-debezium-mongodb/src/generated/resources/org/apache/camel/component/debezium/debezium-mongodb.json b/components/camel-debezium-mongodb/src/generated/resources/org/apache/camel/component/debezium/debezium-mongodb.json
index 617d8d6c..3a99ec3 100644
--- a/components/camel-debezium-mongodb/src/generated/resources/org/apache/camel/component/debezium/debezium-mongodb.json
+++ b/components/camel-debezium-mongodb/src/generated/resources/org/apache/camel/component/debezium/debezium-mongodb.json
@@ -45,7 +45,7 @@
"fieldRenames": { "kind": "property", "displayName": "Field Renames", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Description is not available here, please check Debezium website for corresponding key [...]
"heartbeatIntervalMs": { "kind": "property", "displayName": "Heartbeat Interval Ms", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Length of an interval in milli-seconds in in which the connect [...]
"heartbeatTopicsPrefix": { "kind": "property", "displayName": "Heartbeat Topics Prefix", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is used to [...]
- "initialSyncMaxThreads": { "kind": "property", "displayName": "Initial Sync Max Threads", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum number of threads used to perform an intial sync [...]
+ "initialSyncMaxThreads": { "kind": "property", "displayName": "Initial Sync Max Threads", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum number of threads used to perform an initial sync [...]
"maxBatchSize": { "kind": "property", "displayName": "Max Batch Size", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
"maxQueueSize": { "kind": "property", "displayName": "Max Queue Size", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "8192", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of the queue for change events read from the database log bu [...]
"mongodbHosts": { "kind": "property", "displayName": "Mongodb Hosts", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The hostname and port pairs (in the form 'host' or 'host:port') of the MongoDB server( [...]
@@ -56,6 +56,7 @@
"mongodbSslInvalidHostnameAllowed": { "kind": "property", "displayName": "Mongodb Ssl Invalid Hostname Allowed", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Whether invalid host names [...]
"mongodbUser": { "kind": "property", "displayName": "Mongodb User", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Database user for connecting to MongoDB, if necessary." },
"pollIntervalMs": { "kind": "property", "displayName": "Poll Interval Ms", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Frequency in milliseconds to wait for new change events to appear aft [...]
+ "skippedOperations": { "kind": "property", "displayName": "Skipped Operations", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The comma-separated list of operations to skip during streaming, defined as: [...]
"snapshotDelayMs": { "kind": "property", "displayName": "Snapshot Delay Ms", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The number of milliseconds to delay before a snapshot will begin." },
"snapshotFetchSize": { "kind": "property", "displayName": "Snapshot Fetch Size", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The maximum number of records that should be loaded into memory while performing a snap [...]
"snapshotMode": { "kind": "property", "displayName": "Snapshot Mode", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "initial", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The criteria for running a snapshot upon startup of the con [...]
@@ -92,7 +93,7 @@
"fieldRenames": { "kind": "parameter", "displayName": "Field Renames", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Description is not available here, please check Debezium website for corresponding ke [...]
"heartbeatIntervalMs": { "kind": "parameter", "displayName": "Heartbeat Interval Ms", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Length of an interval in milli-seconds in in which the connec [...]
"heartbeatTopicsPrefix": { "kind": "parameter", "displayName": "Heartbeat Topics Prefix", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is used to [...]
- "initialSyncMaxThreads": { "kind": "parameter", "displayName": "Initial Sync Max Threads", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum number of threads used to perform an intial sync [...]
+ "initialSyncMaxThreads": { "kind": "parameter", "displayName": "Initial Sync Max Threads", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum number of threads used to perform an initial syn [...]
"maxBatchSize": { "kind": "parameter", "displayName": "Max Batch Size", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
"maxQueueSize": { "kind": "parameter", "displayName": "Max Queue Size", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "8192", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of the queue for change events read from the database log b [...]
"mongodbHosts": { "kind": "parameter", "displayName": "Mongodb Hosts", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The hostname and port pairs (in the form 'host' or 'host:port') of the MongoDB server [...]
@@ -103,6 +104,7 @@
"mongodbSslInvalidHostnameAllowed": { "kind": "parameter", "displayName": "Mongodb Ssl Invalid Hostname Allowed", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Whether invalid host name [...]
"mongodbUser": { "kind": "parameter", "displayName": "Mongodb User", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Database user for connecting to MongoDB, if necessary." },
"pollIntervalMs": { "kind": "parameter", "displayName": "Poll Interval Ms", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Frequency in milliseconds to wait for new change events to appear af [...]
+ "skippedOperations": { "kind": "parameter", "displayName": "Skipped Operations", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The comma-separated list of operations to skip during streaming, defined as [...]
"snapshotDelayMs": { "kind": "parameter", "displayName": "Snapshot Delay Ms", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The number of milliseconds to delay before a snapshot will begin." },
"snapshotFetchSize": { "kind": "parameter", "displayName": "Snapshot Fetch Size", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The maximum number of records that should be loaded into memory while performing a sna [...]
"snapshotMode": { "kind": "parameter", "displayName": "Snapshot Mode", "group": "mongodb", "label": "consumer,mongodb", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "initial", "configurationClass": "org.apache.camel.component.debezium.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The criteria for running a snapshot upon startup of the co [...]
diff --git a/components/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc b/components/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
index ccb7272..e0442e5 100644
--- a/components/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
+++ b/components/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
@@ -42,7 +42,7 @@ debezium-mongodb:name[?options]
// component options: START
-The Debezium MongoDB Connector component supports 42 options, which are listed below.
+The Debezium MongoDB Connector component supports 43 options, which are listed below.
@@ -75,7 +75,7 @@ The Debezium MongoDB Connector component supports 42 options, which are listed b
| *fieldRenames* (mongodb) | Description is not available here, please check Debezium website for corresponding key 'field.renames' description. | | String
| *heartbeatIntervalMs* (mongodb) | Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default. | 0 | int
| *heartbeatTopicsPrefix* (mongodb) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
-| *initialSyncMaxThreads* (mongodb) | Maximum number of threads used to perform an intial sync of the collections in a replica set. Defaults to 1. | 1 | int
+| *initialSyncMaxThreads* (mongodb) | Maximum number of threads used to perform an initial sync of the collections in a replica set. Defaults to 1. | 1 | int
| *maxBatchSize* (mongodb) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
| *maxQueueSize* (mongodb) | Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size. | 8192 | int
| *mongodbHosts* (mongodb) | The hostname and port pairs (in the form 'host' or 'host:port') of the MongoDB server(s) in the replica set. | | String
@@ -86,6 +86,7 @@ The Debezium MongoDB Connector component supports 42 options, which are listed b
| *mongodbSslInvalidHostname Allowed* (mongodb) | Whether invalid host names are allowed when using SSL. If true the connection will not prevent man-in-the-middle attacks | false | boolean
| *mongodbUser* (mongodb) | Database user for connecting to MongoDB, if necessary. | | String
| *pollIntervalMs* (mongodb) | Frequency in milliseconds to wait for new change events to appear after receiving no events. Defaults to 500ms. | 500 | long
+| *skippedOperations* (mongodb) | The comma-separated list of operations to skip during streaming, defined as: 'i' for inserts; 'u' for updates; 'd' for deletes. By default, no operations will be skipped. | | String
| *snapshotDelayMs* (mongodb) | The number of milliseconds to delay before a snapshot will begin. | 0 | long
| *snapshotFetchSize* (mongodb) | The maximum number of records that should be loaded into memory while performing a snapshot | | int
| *snapshotMode* (mongodb) | The criteria for running a snapshot upon startup of the connector. Options include: 'initial' (the default) to specify the connector should always perform an initial sync when required; 'never' to specify the connector should never perform an initial sync | initial | String
@@ -114,7 +115,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (44 parameters):
+=== Query Parameters (45 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -148,7 +149,7 @@ with the following path and query parameters:
| *fieldRenames* (mongodb) | Description is not available here, please check Debezium website for corresponding key 'field.renames' description. | | String
| *heartbeatIntervalMs* (mongodb) | Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default. | 0 | int
| *heartbeatTopicsPrefix* (mongodb) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
-| *initialSyncMaxThreads* (mongodb) | Maximum number of threads used to perform an intial sync of the collections in a replica set. Defaults to 1. | 1 | int
+| *initialSyncMaxThreads* (mongodb) | Maximum number of threads used to perform an initial sync of the collections in a replica set. Defaults to 1. | 1 | int
| *maxBatchSize* (mongodb) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
| *maxQueueSize* (mongodb) | Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size. | 8192 | int
| *mongodbHosts* (mongodb) | The hostname and port pairs (in the form 'host' or 'host:port') of the MongoDB server(s) in the replica set. | | String
@@ -159,6 +160,7 @@ with the following path and query parameters:
| *mongodbSslInvalidHostname Allowed* (mongodb) | Whether invalid host names are allowed when using SSL. If true the connection will not prevent man-in-the-middle attacks | false | boolean
| *mongodbUser* (mongodb) | Database user for connecting to MongoDB, if necessary. | | String
| *pollIntervalMs* (mongodb) | Frequency in milliseconds to wait for new change events to appear after receiving no events. Defaults to 500ms. | 500 | long
+| *skippedOperations* (mongodb) | The comma-separated list of operations to skip during streaming, defined as: 'i' for inserts; 'u' for updates; 'd' for deletes. By default, no operations will be skipped. | | String
| *snapshotDelayMs* (mongodb) | The number of milliseconds to delay before a snapshot will begin. | 0 | long
| *snapshotFetchSize* (mongodb) | The maximum number of records that should be loaded into memory while performing a snapshot | | int
| *snapshotMode* (mongodb) | The criteria for running a snapshot upon startup of the connector. Options include: 'initial' (the default) to specify the connector should always perform an initial sync when required; 'never' to specify the connector should never perform an initial sync | initial | String
diff --git a/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlComponentConfigurer.java b/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlComponentConfigurer.java
index 5e4b4bb..6f3bff2 100644
--- a/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlComponentConfigurer.java
+++ b/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlComponentConfigurer.java
@@ -99,6 +99,8 @@ public class DebeziumMySqlComponentConfigurer extends PropertyConfigurerSupport
case "enableTimeAdjuster": getOrCreateConfiguration(target).setEnableTimeAdjuster(property(camelContext, boolean.class, value)); return true;
case "eventdeserializationfailurehandlingmode":
case "eventDeserializationFailureHandlingMode": getOrCreateConfiguration(target).setEventDeserializationFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": getOrCreateConfiguration(target).setEventProcessingFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
case "gtidnewchannelposition":
case "gtidNewChannelPosition": getOrCreateConfiguration(target).setGtidNewChannelPosition(property(camelContext, java.lang.String.class, value)); return true;
case "gtidsourceexcludes":
@@ -213,6 +215,7 @@ public class DebeziumMySqlComponentConfigurer extends PropertyConfigurerSupport
answer.put("decimalHandlingMode", java.lang.String.class);
answer.put("enableTimeAdjuster", boolean.class);
answer.put("eventDeserializationFailureHandlingMode", java.lang.String.class);
+ answer.put("eventProcessingFailureHandlingMode", java.lang.String.class);
answer.put("gtidNewChannelPosition", java.lang.String.class);
answer.put("gtidSourceExcludes", java.lang.String.class);
answer.put("gtidSourceFilterDmlEvents", boolean.class);
@@ -328,6 +331,8 @@ public class DebeziumMySqlComponentConfigurer extends PropertyConfigurerSupport
case "enableTimeAdjuster": return getOrCreateConfiguration(target).isEnableTimeAdjuster();
case "eventdeserializationfailurehandlingmode":
case "eventDeserializationFailureHandlingMode": return getOrCreateConfiguration(target).getEventDeserializationFailureHandlingMode();
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": return getOrCreateConfiguration(target).getEventProcessingFailureHandlingMode();
case "gtidnewchannelposition":
case "gtidNewChannelPosition": return getOrCreateConfiguration(target).getGtidNewChannelPosition();
case "gtidsourceexcludes":
diff --git a/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlEndpointConfigurer.java b/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlEndpointConfigurer.java
index 3f5f9e3..d046751 100644
--- a/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlEndpointConfigurer.java
+++ b/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/DebeziumMySqlEndpointConfigurer.java
@@ -91,6 +91,8 @@ public class DebeziumMySqlEndpointConfigurer extends PropertyConfigurerSupport i
case "enableTimeAdjuster": target.getConfiguration().setEnableTimeAdjuster(property(camelContext, boolean.class, value)); return true;
case "eventdeserializationfailurehandlingmode":
case "eventDeserializationFailureHandlingMode": target.getConfiguration().setEventDeserializationFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": target.getConfiguration().setEventProcessingFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
case "exceptionhandler":
case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true;
case "exchangepattern":
@@ -209,6 +211,7 @@ public class DebeziumMySqlEndpointConfigurer extends PropertyConfigurerSupport i
answer.put("decimalHandlingMode", java.lang.String.class);
answer.put("enableTimeAdjuster", boolean.class);
answer.put("eventDeserializationFailureHandlingMode", java.lang.String.class);
+ answer.put("eventProcessingFailureHandlingMode", java.lang.String.class);
answer.put("exceptionHandler", org.apache.camel.spi.ExceptionHandler.class);
answer.put("exchangePattern", org.apache.camel.ExchangePattern.class);
answer.put("gtidNewChannelPosition", java.lang.String.class);
@@ -326,6 +329,8 @@ public class DebeziumMySqlEndpointConfigurer extends PropertyConfigurerSupport i
case "enableTimeAdjuster": return target.getConfiguration().isEnableTimeAdjuster();
case "eventdeserializationfailurehandlingmode":
case "eventDeserializationFailureHandlingMode": return target.getConfiguration().getEventDeserializationFailureHandlingMode();
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": return target.getConfiguration().getEventProcessingFailureHandlingMode();
case "exceptionhandler":
case "exceptionHandler": return target.getExceptionHandler();
case "exchangepattern":
diff --git a/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/configuration/MySqlConnectorEmbeddedDebeziumConfiguration.java b/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/configuration/MySqlConnectorEmbeddedDebeziumConfiguration.java
index 3329205..bbd9f89 100644
--- a/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/configuration/MySqlConnectorEmbeddedDebeziumConfiguration.java
+++ b/components/camel-debezium-mysql/src/generated/java/org/apache/camel/component/debezium/configuration/MySqlConnectorEmbeddedDebeziumConfiguration.java
@@ -58,8 +58,8 @@ public class MySqlConnectorEmbeddedDebeziumConfiguration
private String inconsistentSchemaHandlingMode = "fail";
@UriParam(label = LABEL_NAME, defaultValue = "true")
private boolean enableTimeAdjuster = true;
- @UriParam(label = LABEL_NAME, defaultValue = "latest")
- private String gtidNewChannelPosition = "latest";
+ @UriParam(label = LABEL_NAME, defaultValue = "earliest")
+ private String gtidNewChannelPosition = "earliest";
@UriParam(label = LABEL_NAME)
@Metadata(required = true)
private String databasePassword;
@@ -114,6 +114,8 @@ public class MySqlConnectorEmbeddedDebeziumConfiguration
@UriParam(label = LABEL_NAME)
@Metadata(required = true)
private String databaseServerName;
+ @UriParam(label = LABEL_NAME, defaultValue = "fail")
+ private String eventProcessingFailureHandlingMode = "fail";
@UriParam(label = LABEL_NAME, defaultValue = "3306")
private int databasePort = 3306;
@UriParam(label = LABEL_NAME)
@@ -437,7 +439,7 @@ public class MySqlConnectorEmbeddedDebeziumConfiguration
* exception indicating the problematic event and its binlog position is
* raised, causing the connector to be stopped; 'warn' the problematic event
* and its binlog position will be logged and the event will be
- * skipped;'ignore' the problematic event will be skipped.
+ * skipped;'skip' the problematic event will be skipped.
*/
public void setInconsistentSchemaHandlingMode(
String inconsistentSchemaHandlingMode) {
@@ -465,8 +467,8 @@ public class MySqlConnectorEmbeddedDebeziumConfiguration
/**
* If set to 'latest', when connector sees new GTID, it will start consuming
* gtid channel from the server latest executed gtid position. If 'earliest'
- * connector starts reading channel from first available (not purged) gtid
- * position on the server.
+ * (the default) connector starts reading channel from first available (not
+ * purged) gtid position on the server.
*/
public void setGtidNewChannelPosition(String gtidNewChannelPosition) {
this.gtidNewChannelPosition = gtidNewChannelPosition;
@@ -840,6 +842,23 @@ public class MySqlConnectorEmbeddedDebeziumConfiguration
}
/**
+ * Specify how failures during processing of events (i.e. when encountering
+ * a corrupted event) should be handled, including:'fail' (the default) an
+ * exception indicating the problematic event and its position is raised,
+ * causing the connector to be stopped; 'warn' the problematic event and its
+ * position will be logged and the event will be skipped;'ignore' the
+ * problematic event will be skipped.
+ */
+ public void setEventProcessingFailureHandlingMode(
+ String eventProcessingFailureHandlingMode) {
+ this.eventProcessingFailureHandlingMode = eventProcessingFailureHandlingMode;
+ }
+
+ public String getEventProcessingFailureHandlingMode() {
+ return eventProcessingFailureHandlingMode;
+ }
+
+ /**
* Port of the MySQL database server.
*/
public void setDatabasePort(int databasePort) {
@@ -1007,6 +1026,7 @@ public class MySqlConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "event.deserialization.failure.handling.mode", eventDeserializationFailureHandlingMode);
addPropertyIfNotNull(configBuilder, "time.precision.mode", timePrecisionMode);
addPropertyIfNotNull(configBuilder, "database.server.name", databaseServerName);
+ addPropertyIfNotNull(configBuilder, "event.processing.failure.handling.mode", eventProcessingFailureHandlingMode);
addPropertyIfNotNull(configBuilder, "database.port", databasePort);
addPropertyIfNotNull(configBuilder, "database.ssl.truststore", databaseSslTruststore);
addPropertyIfNotNull(configBuilder, "database.ssl.mode", databaseSslMode);
diff --git a/components/camel-debezium-mysql/src/generated/resources/org/apache/camel/component/debezium/debezium-mysql.json b/components/camel-debezium-mysql/src/generated/resources/org/apache/camel/component/debezium/debezium-mysql.json
index 88ff00b..6b25180 100644
--- a/components/camel-debezium-mysql/src/generated/resources/org/apache/camel/component/debezium/debezium-mysql.json
+++ b/components/camel-debezium-mysql/src/generated/resources/org/apache/camel/component/debezium/debezium-mysql.json
@@ -66,7 +66,8 @@
"decimalHandlingMode": { "kind": "property", "displayName": "Decimal Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "precise", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how DECIMAL and NUMERIC columns should be [...]
"enableTimeAdjuster": { "kind": "property", "displayName": "Enable Time Adjuster", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "MySQL allows user to insert year value as either 2-digit or 4-d [...]
"eventDeserializationFailureHandlingMode": { "kind": "property", "displayName": "Event Deserialization Failure Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how [...]
- "gtidNewChannelPosition": { "kind": "property", "displayName": "Gtid New Channel Position", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "latest", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "If set to 'latest', when connector sees new [...]
+ "eventProcessingFailureHandlingMode": { "kind": "property", "displayName": "Event Processing Failure Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how failures [...]
+ "gtidNewChannelPosition": { "kind": "property", "displayName": "Gtid New Channel Position", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "earliest", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "If set to 'latest', when connector sees ne [...]
"gtidSourceExcludes": { "kind": "property", "displayName": "Gtid Source Excludes", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The source UUIDs used to exclude GTID ranges when determine the starting positi [...]
"gtidSourceFilterDmlEvents": { "kind": "property", "displayName": "Gtid Source Filter Dml Events", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "If set to true, we will only produce DML events [...]
"gtidSourceIncludes": { "kind": "property", "displayName": "Gtid Source Includes", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The source UUIDs used to include GTID ranges when determine the starting positi [...]
@@ -74,7 +75,7 @@
"heartbeatTopicsPrefix": { "kind": "property", "displayName": "Heartbeat Topics Prefix", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is used to name h [...]
"includeQuery": { "kind": "property", "displayName": "Include Query", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Whether the connector should include the original SQL query that generated [...]
"includeSchemaChanges": { "kind": "property", "displayName": "Include Schema Changes", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Whether the connector should publish changes in the databas [...]
- "inconsistentSchemaHandlingMode": { "kind": "property", "displayName": "Inconsistent Schema Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how binlog events that [...]
+ "inconsistentSchemaHandlingMode": { "kind": "property", "displayName": "Inconsistent Schema Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how binlog events that [...]
"maxBatchSize": { "kind": "property", "displayName": "Max Batch Size", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
"maxQueueSize": { "kind": "property", "displayName": "Max Queue Size", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "8192", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of the queue for change events read from the database log but not [...]
"messageKeyColumns": { "kind": "property", "displayName": "Message Key Columns", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "A semicolon-separated list of expressions that match fully-qualified tables and c [...]
@@ -143,7 +144,8 @@
"decimalHandlingMode": { "kind": "parameter", "displayName": "Decimal Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "precise", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how DECIMAL and NUMERIC columns should be [...]
"enableTimeAdjuster": { "kind": "parameter", "displayName": "Enable Time Adjuster", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "MySQL allows user to insert year value as either 2-digit or 4- [...]
"eventDeserializationFailureHandlingMode": { "kind": "parameter", "displayName": "Event Deserialization Failure Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify ho [...]
- "gtidNewChannelPosition": { "kind": "parameter", "displayName": "Gtid New Channel Position", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "latest", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "If set to 'latest', when connector sees new [...]
+ "eventProcessingFailureHandlingMode": { "kind": "parameter", "displayName": "Event Processing Failure Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how failures [...]
+ "gtidNewChannelPosition": { "kind": "parameter", "displayName": "Gtid New Channel Position", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "earliest", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "If set to 'latest', when connector sees n [...]
"gtidSourceExcludes": { "kind": "parameter", "displayName": "Gtid Source Excludes", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The source UUIDs used to exclude GTID ranges when determine the starting posit [...]
"gtidSourceFilterDmlEvents": { "kind": "parameter", "displayName": "Gtid Source Filter Dml Events", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "If set to true, we will only produce DML event [...]
"gtidSourceIncludes": { "kind": "parameter", "displayName": "Gtid Source Includes", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The source UUIDs used to include GTID ranges when determine the starting posit [...]
@@ -151,7 +153,7 @@
"heartbeatTopicsPrefix": { "kind": "parameter", "displayName": "Heartbeat Topics Prefix", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is used to name [...]
"includeQuery": { "kind": "parameter", "displayName": "Include Query", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Whether the connector should include the original SQL query that generated [...]
"includeSchemaChanges": { "kind": "parameter", "displayName": "Include Schema Changes", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Whether the connector should publish changes in the databa [...]
- "inconsistentSchemaHandlingMode": { "kind": "parameter", "displayName": "Inconsistent Schema Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how binlog events tha [...]
+ "inconsistentSchemaHandlingMode": { "kind": "parameter", "displayName": "Inconsistent Schema Handling Mode", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how binlog events tha [...]
"maxBatchSize": { "kind": "parameter", "displayName": "Max Batch Size", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
"maxQueueSize": { "kind": "parameter", "displayName": "Max Queue Size", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "8192", "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of the queue for change events read from the database log but not [...]
"messageKeyColumns": { "kind": "parameter", "displayName": "Message Key Columns", "group": "mysql", "label": "consumer,mysql", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.MySqlConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "A semicolon-separated list of expressions that match fully-qualified tables and [...]
diff --git a/components/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc b/components/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
index 9d986c1..53df7d1 100644
--- a/components/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
+++ b/components/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
@@ -49,7 +49,7 @@ debezium-mysql:name[?options]
// component options: START
-The Debezium MySQL Connector component supports 72 options, which are listed below.
+The Debezium MySQL Connector component supports 73 options, which are listed below.
@@ -103,7 +103,8 @@ The Debezium MySQL Connector component supports 72 options, which are listed bel
| *decimalHandlingMode* (mysql) | Specify how DECIMAL and NUMERIC columns should be represented in change events, including:'precise' (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect's 'org.apache.kafka.connect.data.Decimal' type; 'string' uses string to represent values; 'double' represents values using Java's 'double', which may not offer the precision but will be far easier to use in con [...]
| *enableTimeAdjuster* (mysql) | MySQL allows user to insert year value as either 2-digit or 4-digit. In case of two digit the value is automatically mapped into 1970 - 2069.false - delegates the implicit conversion to the databasetrue - (the default) Debezium makes the conversion | true | boolean
| *eventDeserializationFailure HandlingMode* (mysql) | Specify how failures during deserialization of binlog events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its binlog position is raised, causing the connector to be stopped; 'warn' the problematic event and its binlog position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
-| *gtidNewChannelPosition* (mysql) | If set to 'latest', when connector sees new GTID, it will start consuming gtid channel from the server latest executed gtid position. If 'earliest' connector starts reading channel from first available (not purged) gtid position on the server. | latest | String
+| *eventProcessingFailureHandling Mode* (mysql) | Specify how failures during processing of events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its position is raised, causing the connector to be stopped; 'warn' the problematic event and its position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
+| *gtidNewChannelPosition* (mysql) | If set to 'latest', when connector sees new GTID, it will start consuming gtid channel from the server latest executed gtid position. If 'earliest' (the default) connector starts reading channel from first available (not purged) gtid position on the server. | earliest | String
| *gtidSourceExcludes* (mysql) | The source UUIDs used to exclude GTID ranges when determine the starting position in the MySQL server's binlog. | | String
| *gtidSourceFilterDmlEvents* (mysql) | If set to true, we will only produce DML events into Kafka for transactions that were written on mysql servers with UUIDs matching the filters defined by the gtid.source.includes or gtid.source.excludes configuration options, if they are specified. | true | boolean
| *gtidSourceIncludes* (mysql) | The source UUIDs used to include GTID ranges when determine the starting position in the MySQL server's binlog. | | String
@@ -111,7 +112,7 @@ The Debezium MySQL Connector component supports 72 options, which are listed bel
| *heartbeatTopicsPrefix* (mysql) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
| *includeQuery* (mysql) | Whether the connector should include the original SQL query that generated the change event. Note: This option requires MySQL be configured with the binlog_rows_query_log_events option set to ON. Query will not be present for events generated from snapshot. WARNING: Enabling this option may expose tables or fields explicitly blacklisted or masked by including the original SQL statement in the change event. For this reason the default value is 'false'. | false | [...]
| *includeSchemaChanges* (mysql) | Whether the connector should publish changes in the database schema to a Kafka topic with the same name as the database server ID. Each schema change will be recorded using a key that contains the database name and whose value includes the DDL statement(s).The default is 'true'. This is independent of how the connector internally records database history. | true | boolean
-| *inconsistentSchemaHandlingMode* (mysql) | Specify how binlog events that belong to a table missing from internal schema representation (i.e. internal representation is not consistent with database) should be handled, including:'fail' (the default) an exception indicating the problematic event and its binlog position is raised, causing the connector to be stopped; 'warn' the problematic event and its binlog position will be logged and the event will be skipped;'ignore' the problematic [...]
+| *inconsistentSchemaHandlingMode* (mysql) | Specify how binlog events that belong to a table missing from internal schema representation (i.e. internal representation is not consistent with database) should be handled, including:'fail' (the default) an exception indicating the problematic event and its binlog position is raised, causing the connector to be stopped; 'warn' the problematic event and its binlog position will be logged and the event will be skipped;'skip' the problematic ev [...]
| *maxBatchSize* (mysql) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
| *maxQueueSize* (mysql) | Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size. | 8192 | int
| *messageKeyColumns* (mysql) | A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ':',where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector,and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table's primary key column(s) [...]
@@ -151,7 +152,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (74 parameters):
+=== Query Parameters (75 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -206,7 +207,8 @@ with the following path and query parameters:
| *decimalHandlingMode* (mysql) | Specify how DECIMAL and NUMERIC columns should be represented in change events, including:'precise' (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect's 'org.apache.kafka.connect.data.Decimal' type; 'string' uses string to represent values; 'double' represents values using Java's 'double', which may not offer the precision but will be far easier to use in con [...]
| *enableTimeAdjuster* (mysql) | MySQL allows user to insert year value as either 2-digit or 4-digit. In case of two digit the value is automatically mapped into 1970 - 2069.false - delegates the implicit conversion to the databasetrue - (the default) Debezium makes the conversion | true | boolean
| *eventDeserializationFailure HandlingMode* (mysql) | Specify how failures during deserialization of binlog events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its binlog position is raised, causing the connector to be stopped; 'warn' the problematic event and its binlog position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
-| *gtidNewChannelPosition* (mysql) | If set to 'latest', when connector sees new GTID, it will start consuming gtid channel from the server latest executed gtid position. If 'earliest' connector starts reading channel from first available (not purged) gtid position on the server. | latest | String
+| *eventProcessingFailureHandling Mode* (mysql) | Specify how failures during processing of events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its position is raised, causing the connector to be stopped; 'warn' the problematic event and its position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
+| *gtidNewChannelPosition* (mysql) | If set to 'latest', when connector sees new GTID, it will start consuming gtid channel from the server latest executed gtid position. If 'earliest' (the default) connector starts reading channel from first available (not purged) gtid position on the server. | earliest | String
| *gtidSourceExcludes* (mysql) | The source UUIDs used to exclude GTID ranges when determine the starting position in the MySQL server's binlog. | | String
| *gtidSourceFilterDmlEvents* (mysql) | If set to true, we will only produce DML events into Kafka for transactions that were written on mysql servers with UUIDs matching the filters defined by the gtid.source.includes or gtid.source.excludes configuration options, if they are specified. | true | boolean
| *gtidSourceIncludes* (mysql) | The source UUIDs used to include GTID ranges when determine the starting position in the MySQL server's binlog. | | String
@@ -214,7 +216,7 @@ with the following path and query parameters:
| *heartbeatTopicsPrefix* (mysql) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
| *includeQuery* (mysql) | Whether the connector should include the original SQL query that generated the change event. Note: This option requires MySQL be configured with the binlog_rows_query_log_events option set to ON. Query will not be present for events generated from snapshot. WARNING: Enabling this option may expose tables or fields explicitly blacklisted or masked by including the original SQL statement in the change event. For this reason the default value is 'false'. | false | [...]
| *includeSchemaChanges* (mysql) | Whether the connector should publish changes in the database schema to a Kafka topic with the same name as the database server ID. Each schema change will be recorded using a key that contains the database name and whose value includes the DDL statement(s).The default is 'true'. This is independent of how the connector internally records database history. | true | boolean
-| *inconsistentSchemaHandlingMode* (mysql) | Specify how binlog events that belong to a table missing from internal schema representation (i.e. internal representation is not consistent with database) should be handled, including:'fail' (the default) an exception indicating the problematic event and its binlog position is raised, causing the connector to be stopped; 'warn' the problematic event and its binlog position will be logged and the event will be skipped;'ignore' the problematic [...]
+| *inconsistentSchemaHandlingMode* (mysql) | Specify how binlog events that belong to a table missing from internal schema representation (i.e. internal representation is not consistent with database) should be handled, including:'fail' (the default) an exception indicating the problematic event and its binlog position is raised, causing the connector to be stopped; 'warn' the problematic event and its binlog position will be logged and the event will be skipped;'skip' the problematic ev [...]
| *maxBatchSize* (mysql) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
| *maxQueueSize* (mysql) | Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size. | 8192 | int
| *messageKeyColumns* (mysql) | A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ':',where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector,and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table's primary key column(s) [...]
diff --git a/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresComponentConfigurer.java b/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresComponentConfigurer.java
index 346adb7..9a43bf5 100644
--- a/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresComponentConfigurer.java
+++ b/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresComponentConfigurer.java
@@ -67,6 +67,10 @@ public class DebeziumPostgresComponentConfigurer extends PropertyConfigurerSuppo
case "databaseUser": getOrCreateConfiguration(target).setDatabaseUser(property(camelContext, java.lang.String.class, value)); return true;
case "decimalhandlingmode":
case "decimalHandlingMode": getOrCreateConfiguration(target).setDecimalHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": getOrCreateConfiguration(target).setEventProcessingFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "heartbeatactionquery":
+ case "heartbeatActionQuery": getOrCreateConfiguration(target).setHeartbeatActionQuery(property(camelContext, java.lang.String.class, value)); return true;
case "heartbeatintervalms":
case "heartbeatIntervalMs": getOrCreateConfiguration(target).setHeartbeatIntervalMs(property(camelContext, int.class, value)); return true;
case "heartbeattopicsprefix":
@@ -107,6 +111,8 @@ public class DebeziumPostgresComponentConfigurer extends PropertyConfigurerSuppo
case "pluginName": getOrCreateConfiguration(target).setPluginName(property(camelContext, java.lang.String.class, value)); return true;
case "pollintervalms":
case "pollIntervalMs": getOrCreateConfiguration(target).setPollIntervalMs(property(camelContext, long.class, value)); return true;
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": getOrCreateConfiguration(target).setProvideTransactionMetadata(property(camelContext, boolean.class, value)); return true;
case "publicationname":
case "publicationName": getOrCreateConfiguration(target).setPublicationName(property(camelContext, java.lang.String.class, value)); return true;
case "schemablacklist":
@@ -181,6 +187,8 @@ public class DebeziumPostgresComponentConfigurer extends PropertyConfigurerSuppo
answer.put("databaseTcpkeepalive", boolean.class);
answer.put("databaseUser", java.lang.String.class);
answer.put("decimalHandlingMode", java.lang.String.class);
+ answer.put("eventProcessingFailureHandlingMode", java.lang.String.class);
+ answer.put("heartbeatActionQuery", java.lang.String.class);
answer.put("heartbeatIntervalMs", int.class);
answer.put("heartbeatTopicsPrefix", java.lang.String.class);
answer.put("hstoreHandlingMode", java.lang.String.class);
@@ -201,6 +209,7 @@ public class DebeziumPostgresComponentConfigurer extends PropertyConfigurerSuppo
answer.put("offsetStorageTopic", java.lang.String.class);
answer.put("pluginName", java.lang.String.class);
answer.put("pollIntervalMs", long.class);
+ answer.put("provideTransactionMetadata", boolean.class);
answer.put("publicationName", java.lang.String.class);
answer.put("schemaBlacklist", java.lang.String.class);
answer.put("schemaRefreshMode", java.lang.String.class);
@@ -272,6 +281,10 @@ public class DebeziumPostgresComponentConfigurer extends PropertyConfigurerSuppo
case "databaseUser": return getOrCreateConfiguration(target).getDatabaseUser();
case "decimalhandlingmode":
case "decimalHandlingMode": return getOrCreateConfiguration(target).getDecimalHandlingMode();
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": return getOrCreateConfiguration(target).getEventProcessingFailureHandlingMode();
+ case "heartbeatactionquery":
+ case "heartbeatActionQuery": return getOrCreateConfiguration(target).getHeartbeatActionQuery();
case "heartbeatintervalms":
case "heartbeatIntervalMs": return getOrCreateConfiguration(target).getHeartbeatIntervalMs();
case "heartbeattopicsprefix":
@@ -312,6 +325,8 @@ public class DebeziumPostgresComponentConfigurer extends PropertyConfigurerSuppo
case "pluginName": return getOrCreateConfiguration(target).getPluginName();
case "pollintervalms":
case "pollIntervalMs": return getOrCreateConfiguration(target).getPollIntervalMs();
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": return getOrCreateConfiguration(target).isProvideTransactionMetadata();
case "publicationname":
case "publicationName": return getOrCreateConfiguration(target).getPublicationName();
case "schemablacklist":
diff --git a/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresEndpointConfigurer.java b/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresEndpointConfigurer.java
index 1bd86be..51fc9ca 100644
--- a/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresEndpointConfigurer.java
+++ b/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/DebeziumPostgresEndpointConfigurer.java
@@ -59,10 +59,14 @@ public class DebeziumPostgresEndpointConfigurer extends PropertyConfigurerSuppor
case "databaseUser": target.getConfiguration().setDatabaseUser(property(camelContext, java.lang.String.class, value)); return true;
case "decimalhandlingmode":
case "decimalHandlingMode": target.getConfiguration().setDecimalHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": target.getConfiguration().setEventProcessingFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
case "exceptionhandler":
case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true;
case "exchangepattern":
case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
+ case "heartbeatactionquery":
+ case "heartbeatActionQuery": target.getConfiguration().setHeartbeatActionQuery(property(camelContext, java.lang.String.class, value)); return true;
case "heartbeatintervalms":
case "heartbeatIntervalMs": target.getConfiguration().setHeartbeatIntervalMs(property(camelContext, int.class, value)); return true;
case "heartbeattopicsprefix":
@@ -103,6 +107,8 @@ public class DebeziumPostgresEndpointConfigurer extends PropertyConfigurerSuppor
case "pluginName": target.getConfiguration().setPluginName(property(camelContext, java.lang.String.class, value)); return true;
case "pollintervalms":
case "pollIntervalMs": target.getConfiguration().setPollIntervalMs(property(camelContext, long.class, value)); return true;
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": target.getConfiguration().setProvideTransactionMetadata(property(camelContext, boolean.class, value)); return true;
case "publicationname":
case "publicationName": target.getConfiguration().setPublicationName(property(camelContext, java.lang.String.class, value)); return true;
case "schemablacklist":
@@ -177,8 +183,10 @@ public class DebeziumPostgresEndpointConfigurer extends PropertyConfigurerSuppor
answer.put("databaseTcpkeepalive", boolean.class);
answer.put("databaseUser", java.lang.String.class);
answer.put("decimalHandlingMode", java.lang.String.class);
+ answer.put("eventProcessingFailureHandlingMode", java.lang.String.class);
answer.put("exceptionHandler", org.apache.camel.spi.ExceptionHandler.class);
answer.put("exchangePattern", org.apache.camel.ExchangePattern.class);
+ answer.put("heartbeatActionQuery", java.lang.String.class);
answer.put("heartbeatIntervalMs", int.class);
answer.put("heartbeatTopicsPrefix", java.lang.String.class);
answer.put("hstoreHandlingMode", java.lang.String.class);
@@ -199,6 +207,7 @@ public class DebeziumPostgresEndpointConfigurer extends PropertyConfigurerSuppor
answer.put("offsetStorageTopic", java.lang.String.class);
answer.put("pluginName", java.lang.String.class);
answer.put("pollIntervalMs", long.class);
+ answer.put("provideTransactionMetadata", boolean.class);
answer.put("publicationName", java.lang.String.class);
answer.put("schemaBlacklist", java.lang.String.class);
answer.put("schemaRefreshMode", java.lang.String.class);
@@ -270,10 +279,14 @@ public class DebeziumPostgresEndpointConfigurer extends PropertyConfigurerSuppor
case "databaseUser": return target.getConfiguration().getDatabaseUser();
case "decimalhandlingmode":
case "decimalHandlingMode": return target.getConfiguration().getDecimalHandlingMode();
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": return target.getConfiguration().getEventProcessingFailureHandlingMode();
case "exceptionhandler":
case "exceptionHandler": return target.getExceptionHandler();
case "exchangepattern":
case "exchangePattern": return target.getExchangePattern();
+ case "heartbeatactionquery":
+ case "heartbeatActionQuery": return target.getConfiguration().getHeartbeatActionQuery();
case "heartbeatintervalms":
case "heartbeatIntervalMs": return target.getConfiguration().getHeartbeatIntervalMs();
case "heartbeattopicsprefix":
@@ -314,6 +327,8 @@ public class DebeziumPostgresEndpointConfigurer extends PropertyConfigurerSuppor
case "pluginName": return target.getConfiguration().getPluginName();
case "pollintervalms":
case "pollIntervalMs": return target.getConfiguration().getPollIntervalMs();
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": return target.getConfiguration().isProvideTransactionMetadata();
case "publicationname":
case "publicationName": return target.getConfiguration().getPublicationName();
case "schemablacklist":
diff --git a/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/configuration/PostgresConnectorEmbeddedDebeziumConfiguration.java b/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/configuration/PostgresConnectorEmbeddedDebeziumConfiguration.java
index f5dca2e..00848dd 100644
--- a/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/configuration/PostgresConnectorEmbeddedDebeziumConfiguration.java
+++ b/components/camel-debezium-postgres/src/generated/java/org/apache/camel/component/debezium/configuration/PostgresConnectorEmbeddedDebeziumConfiguration.java
@@ -29,6 +29,8 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
@UriParam(label = LABEL_NAME, defaultValue = "disable")
private String databaseSslmode = "disable";
@UriParam(label = LABEL_NAME)
+ private String heartbeatActionQuery;
+ @UriParam(label = LABEL_NAME)
private String databaseSslcert;
@UriParam(label = LABEL_NAME, defaultValue = "500")
private long pollIntervalMs = 500;
@@ -85,6 +87,8 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
private String hstoreHandlingMode = "json";
@UriParam(label = LABEL_NAME, defaultValue = "0")
private long snapshotDelayMs = 0;
+ @UriParam(label = LABEL_NAME, defaultValue = "false")
+ private boolean provideTransactionMetadata = false;
@UriParam(label = LABEL_NAME)
private String tableWhitelist;
@UriParam(label = LABEL_NAME, defaultValue = "false")
@@ -106,6 +110,8 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
@UriParam(label = LABEL_NAME)
@Metadata(required = true)
private String databaseServerName;
+ @UriParam(label = LABEL_NAME, defaultValue = "fail")
+ private String eventProcessingFailureHandlingMode = "fail";
@UriParam(label = LABEL_NAME, defaultValue = "5432")
private int databasePort = 5432;
@UriParam(label = LABEL_NAME, defaultValue = "false")
@@ -233,6 +239,17 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
}
/**
+ * The query executed with every heartbeat. Defaults to an empty string.
+ */
+ public void setHeartbeatActionQuery(String heartbeatActionQuery) {
+ this.heartbeatActionQuery = heartbeatActionQuery;
+ }
+
+ public String getHeartbeatActionQuery() {
+ return heartbeatActionQuery;
+ }
+
+ /**
* File containing the SSL Certificate for the client. See the Postgres SSL
* docs for further information
*/
@@ -581,8 +598,8 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
/**
* Specify how HSTORE columns should be represented in change events,
- * including:'json' represents values as json string'map' (default)
- * represents values using java.util.Map
+ * including:'json' represents values as string-ified JSON (default)'map'
+ * represents values as a key/value map
*/
public void setHstoreHandlingMode(String hstoreHandlingMode) {
this.hstoreHandlingMode = hstoreHandlingMode;
@@ -604,6 +621,17 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
}
/**
+ * Enables transaction metadata extraction together with event counting
+ */
+ public void setProvideTransactionMetadata(boolean provideTransactionMetadata) {
+ this.provideTransactionMetadata = provideTransactionMetadata;
+ }
+
+ public boolean isProvideTransactionMetadata() {
+ return provideTransactionMetadata;
+ }
+
+ /**
* The tables for which changes are to be captured
*/
public void setTableWhitelist(String tableWhitelist) {
@@ -744,6 +772,23 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
}
/**
+ * Specify how failures during processing of events (i.e. when encountering
+ * a corrupted event) should be handled, including:'fail' (the default) an
+ * exception indicating the problematic event and its position is raised,
+ * causing the connector to be stopped; 'warn' the problematic event and its
+ * position will be logged and the event will be skipped;'ignore' the
+ * problematic event will be skipped.
+ */
+ public void setEventProcessingFailureHandlingMode(
+ String eventProcessingFailureHandlingMode) {
+ this.eventProcessingFailureHandlingMode = eventProcessingFailureHandlingMode;
+ }
+
+ public String getEventProcessingFailureHandlingMode() {
+ return eventProcessingFailureHandlingMode;
+ }
+
+ /**
* Port of the Postgres database server.
*/
public void setDatabasePort(int databasePort) {
@@ -802,6 +847,7 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "slot.max.retries", slotMaxRetries);
addPropertyIfNotNull(configBuilder, "schema.refresh.mode", schemaRefreshMode);
addPropertyIfNotNull(configBuilder, "database.sslmode", databaseSslmode);
+ addPropertyIfNotNull(configBuilder, "heartbeat.action.query", heartbeatActionQuery);
addPropertyIfNotNull(configBuilder, "database.sslcert", databaseSslcert);
addPropertyIfNotNull(configBuilder, "poll.interval.ms", pollIntervalMs);
addPropertyIfNotNull(configBuilder, "database.initial.statements", databaseInitialStatements);
@@ -830,6 +876,7 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "slot.name", slotName);
addPropertyIfNotNull(configBuilder, "hstore.handling.mode", hstoreHandlingMode);
addPropertyIfNotNull(configBuilder, "snapshot.delay.ms", snapshotDelayMs);
+ addPropertyIfNotNull(configBuilder, "provide.transaction.metadata", provideTransactionMetadata);
addPropertyIfNotNull(configBuilder, "table.whitelist", tableWhitelist);
addPropertyIfNotNull(configBuilder, "tombstones.on.delete", tombstonesOnDelete);
addPropertyIfNotNull(configBuilder, "slot.retry.delay.ms", slotRetryDelayMs);
@@ -840,6 +887,7 @@ public class PostgresConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "xmin.fetch.interval.ms", xminFetchIntervalMs);
addPropertyIfNotNull(configBuilder, "time.precision.mode", timePrecisionMode);
addPropertyIfNotNull(configBuilder, "database.server.name", databaseServerName);
+ addPropertyIfNotNull(configBuilder, "event.processing.failure.handling.mode", eventProcessingFailureHandlingMode);
addPropertyIfNotNull(configBuilder, "database.port", databasePort);
addPropertyIfNotNull(configBuilder, "include.unknown.datatypes", includeUnknownDatatypes);
addPropertyIfNotNull(configBuilder, "database.hostname", databaseHostname);
diff --git a/components/camel-debezium-postgres/src/generated/resources/org/apache/camel/component/debezium/debezium-postgres.json b/components/camel-debezium-postgres/src/generated/resources/org/apache/camel/component/debezium/debezium-postgres.json
index 7b617a9..49fba66 100644
--- a/components/camel-debezium-postgres/src/generated/resources/org/apache/camel/component/debezium/debezium-postgres.json
+++ b/components/camel-debezium-postgres/src/generated/resources/org/apache/camel/component/debezium/debezium-postgres.json
@@ -50,9 +50,11 @@
"databaseTcpkeepalive": { "kind": "property", "displayName": "Database Tcpkeepalive", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Enable or disable TCP keep-alive probe to avoid dro [...]
"databaseUser": { "kind": "property", "displayName": "Database User", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Name of the Postgres database user to be used when connecting to the database." },
"decimalHandlingMode": { "kind": "property", "displayName": "Decimal Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "precise", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how DECIMAL and NUMERIC columns s [...]
+ "eventProcessingFailureHandlingMode": { "kind": "property", "displayName": "Event Processing Failure Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how [...]
+ "heartbeatActionQuery": { "kind": "property", "displayName": "Heartbeat Action Query", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The query executed with every heartbeat. Defaults to an empty string." },
"heartbeatIntervalMs": { "kind": "property", "displayName": "Heartbeat Interval Ms", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Length of an interval in milli-seconds in in which the conn [...]
"heartbeatTopicsPrefix": { "kind": "property", "displayName": "Heartbeat Topics Prefix", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is used [...]
- "hstoreHandlingMode": { "kind": "property", "displayName": "Hstore Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "json", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how HSTORE columns should be represent [...]
+ "hstoreHandlingMode": { "kind": "property", "displayName": "Hstore Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "json", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how HSTORE columns should be represent [...]
"includeUnknownDatatypes": { "kind": "property", "displayName": "Include Unknown Datatypes", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify whether the fields of data type not [...]
"intervalHandlingMode": { "kind": "property", "displayName": "Interval Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "numeric", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how INTERVAL columns should be [...]
"maxBatchSize": { "kind": "property", "displayName": "Max Batch Size", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
@@ -60,6 +62,7 @@
"messageKeyColumns": { "kind": "property", "displayName": "Message Key Columns", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "A semicolon-separated list of expressions that match fully-qualified tab [...]
"pluginName": { "kind": "property", "displayName": "Plugin Name", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "decoderbufs", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The name of the Postgres logical decoding plugin install [...]
"pollIntervalMs": { "kind": "property", "displayName": "Poll Interval Ms", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Frequency in milliseconds to wait for new change events to appear [...]
+ "provideTransactionMetadata": { "kind": "property", "displayName": "Provide Transaction Metadata", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Enables transaction metadata extracti [...]
"publicationName": { "kind": "property", "displayName": "Publication Name", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "dbz_publication", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The name of the Postgres 10 publication us [...]
"schemaBlacklist": { "kind": "property", "displayName": "Schema Blacklist", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The schemas for which events must not be captured" },
"schemaRefreshMode": { "kind": "property", "displayName": "Schema Refresh Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "columns_diff", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify the conditions that trigger a re [...]
@@ -119,9 +122,11 @@
"databaseTcpkeepalive": { "kind": "parameter", "displayName": "Database Tcpkeepalive", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Enable or disable TCP keep-alive probe to avoid dr [...]
"databaseUser": { "kind": "parameter", "displayName": "Database User", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Name of the Postgres database user to be used when connecting to the database." },
"decimalHandlingMode": { "kind": "parameter", "displayName": "Decimal Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "precise", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how DECIMAL and NUMERIC columns [...]
+ "eventProcessingFailureHandlingMode": { "kind": "parameter", "displayName": "Event Processing Failure Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how [...]
+ "heartbeatActionQuery": { "kind": "parameter", "displayName": "Heartbeat Action Query", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The query executed with every heartbeat. Defaults to an empty str [...]
"heartbeatIntervalMs": { "kind": "parameter", "displayName": "Heartbeat Interval Ms", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Length of an interval in milli-seconds in in which the con [...]
"heartbeatTopicsPrefix": { "kind": "parameter", "displayName": "Heartbeat Topics Prefix", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is used [...]
- "hstoreHandlingMode": { "kind": "parameter", "displayName": "Hstore Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "json", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how HSTORE columns should be represen [...]
+ "hstoreHandlingMode": { "kind": "parameter", "displayName": "Hstore Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "json", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how HSTORE columns should be represen [...]
"includeUnknownDatatypes": { "kind": "parameter", "displayName": "Include Unknown Datatypes", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify whether the fields of data type no [...]
"intervalHandlingMode": { "kind": "parameter", "displayName": "Interval Handling Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "numeric", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how INTERVAL columns should be [...]
"maxBatchSize": { "kind": "parameter", "displayName": "Max Batch Size", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
@@ -129,6 +134,7 @@
"messageKeyColumns": { "kind": "parameter", "displayName": "Message Key Columns", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "A semicolon-separated list of expressions that match fully-qualified ta [...]
"pluginName": { "kind": "parameter", "displayName": "Plugin Name", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "decoderbufs", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The name of the Postgres logical decoding plugin instal [...]
"pollIntervalMs": { "kind": "parameter", "displayName": "Poll Interval Ms", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Frequency in milliseconds to wait for new change events to appear [...]
+ "provideTransactionMetadata": { "kind": "parameter", "displayName": "Provide Transaction Metadata", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Enables transaction metadata extract [...]
"publicationName": { "kind": "parameter", "displayName": "Publication Name", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "dbz_publication", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The name of the Postgres 10 publication u [...]
"schemaBlacklist": { "kind": "parameter", "displayName": "Schema Blacklist", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The schemas for which events must not be captured" },
"schemaRefreshMode": { "kind": "parameter", "displayName": "Schema Refresh Mode", "group": "postgres", "label": "consumer,postgres", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "columns_diff", "configurationClass": "org.apache.camel.component.debezium.configuration.PostgresConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify the conditions that trigger a r [...]
diff --git a/components/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc b/components/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
index cfcc52c..5845ef6 100644
--- a/components/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
+++ b/components/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
@@ -40,7 +40,7 @@ debezium-postgres:name[?options]
// component options: START
-The Debezium PostgresSQL Connector component supports 64 options, which are listed below.
+The Debezium PostgresSQL Connector component supports 67 options, which are listed below.
@@ -78,9 +78,11 @@ The Debezium PostgresSQL Connector component supports 64 options, which are list
| *databaseTcpkeepalive* (postgres) | Enable or disable TCP keep-alive probe to avoid dropping TCP connection | true | boolean
| *databaseUser* (postgres) | Name of the Postgres database user to be used when connecting to the database. | | String
| *decimalHandlingMode* (postgres) | Specify how DECIMAL and NUMERIC columns should be represented in change events, including:'precise' (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect's 'org.apache.kafka.connect.data.Decimal' type; 'string' uses string to represent values; 'double' represents values using Java's 'double', which may not offer the precision but will be far easier to use in [...]
+| *eventProcessingFailureHandling Mode* (postgres) | Specify how failures during processing of events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its position is raised, causing the connector to be stopped; 'warn' the problematic event and its position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
+| *heartbeatActionQuery* (postgres) | The query executed with every heartbeat. Defaults to an empty string. | | String
| *heartbeatIntervalMs* (postgres) | Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default. | 0 | int
| *heartbeatTopicsPrefix* (postgres) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
-| *hstoreHandlingMode* (postgres) | Specify how HSTORE columns should be represented in change events, including:'json' represents values as json string'map' (default) represents values using java.util.Map | json | String
+| *hstoreHandlingMode* (postgres) | Specify how HSTORE columns should be represented in change events, including:'json' represents values as string-ified JSON (default)'map' represents values as a key/value map | json | String
| *includeUnknownDatatypes* (postgres) | Specify whether the fields of data type not supported by Debezium should be processed:'false' (the default) omits the fields; 'true' converts the field into an implementation dependent binary representation. | false | boolean
| *intervalHandlingMode* (postgres) | Specify how INTERVAL columns should be represented in change events, including:'string' represents values as an exact ISO formatted string'numeric' (default) represents values using the inexact conversion into microseconds | numeric | String
| *maxBatchSize* (postgres) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
@@ -88,6 +90,7 @@ The Debezium PostgresSQL Connector component supports 64 options, which are list
| *messageKeyColumns* (postgres) | A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ':',where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector,and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table's primary key column [...]
| *pluginName* (postgres) | The name of the Postgres logical decoding plugin installed on the server. Supported values are 'decoderbufs' and 'wal2json'. Defaults to 'decoderbufs'. | decoderbufs | String
| *pollIntervalMs* (postgres) | Frequency in milliseconds to wait for new change events to appear after receiving no events. Defaults to 500ms. | 500 | long
+| *provideTransactionMetadata* (postgres) | Enables transaction metadata extraction together with event counting | false | boolean
| *publicationName* (postgres) | The name of the Postgres 10 publication used for streaming changes from a plugin.Defaults to 'dbz_publication' | dbz_publication | String
| *schemaBlacklist* (postgres) | The schemas for which events must not be captured | | String
| *schemaRefreshMode* (postgres) | Specify the conditions that trigger a refresh of the in-memory schema for a table. 'columns_diff' (the default) is the safest mode, ensuring the in-memory schema stays in-sync with the database table's schema at all times. 'columns_diff_exclude_unchanged_toast' instructs the connector to refresh the in-memory schema cache if there is a discrepancy between it and the schema derived from the incoming message, unless unchanged TOASTable data fully accounts [...]
@@ -134,7 +137,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (66 parameters):
+=== Query Parameters (69 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -173,9 +176,11 @@ with the following path and query parameters:
| *databaseTcpkeepalive* (postgres) | Enable or disable TCP keep-alive probe to avoid dropping TCP connection | true | boolean
| *databaseUser* (postgres) | Name of the Postgres database user to be used when connecting to the database. | | String
| *decimalHandlingMode* (postgres) | Specify how DECIMAL and NUMERIC columns should be represented in change events, including:'precise' (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect's 'org.apache.kafka.connect.data.Decimal' type; 'string' uses string to represent values; 'double' represents values using Java's 'double', which may not offer the precision but will be far easier to use in [...]
+| *eventProcessingFailureHandling Mode* (postgres) | Specify how failures during processing of events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its position is raised, causing the connector to be stopped; 'warn' the problematic event and its position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
+| *heartbeatActionQuery* (postgres) | The query executed with every heartbeat. Defaults to an empty string. | | String
| *heartbeatIntervalMs* (postgres) | Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default. | 0 | int
| *heartbeatTopicsPrefix* (postgres) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
-| *hstoreHandlingMode* (postgres) | Specify how HSTORE columns should be represented in change events, including:'json' represents values as json string'map' (default) represents values using java.util.Map | json | String
+| *hstoreHandlingMode* (postgres) | Specify how HSTORE columns should be represented in change events, including:'json' represents values as string-ified JSON (default)'map' represents values as a key/value map | json | String
| *includeUnknownDatatypes* (postgres) | Specify whether the fields of data type not supported by Debezium should be processed:'false' (the default) omits the fields; 'true' converts the field into an implementation dependent binary representation. | false | boolean
| *intervalHandlingMode* (postgres) | Specify how INTERVAL columns should be represented in change events, including:'string' represents values as an exact ISO formatted string'numeric' (default) represents values using the inexact conversion into microseconds | numeric | String
| *maxBatchSize* (postgres) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
@@ -183,6 +188,7 @@ with the following path and query parameters:
| *messageKeyColumns* (postgres) | A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ':',where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector,and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table's primary key column [...]
| *pluginName* (postgres) | The name of the Postgres logical decoding plugin installed on the server. Supported values are 'decoderbufs' and 'wal2json'. Defaults to 'decoderbufs'. | decoderbufs | String
| *pollIntervalMs* (postgres) | Frequency in milliseconds to wait for new change events to appear after receiving no events. Defaults to 500ms. | 500 | long
+| *provideTransactionMetadata* (postgres) | Enables transaction metadata extraction together with event counting | false | boolean
| *publicationName* (postgres) | The name of the Postgres 10 publication used for streaming changes from a plugin.Defaults to 'dbz_publication' | dbz_publication | String
| *schemaBlacklist* (postgres) | The schemas for which events must not be captured | | String
| *schemaRefreshMode* (postgres) | Specify the conditions that trigger a refresh of the in-memory schema for a table. 'columns_diff' (the default) is the safest mode, ensuring the in-memory schema stays in-sync with the database table's schema at all times. 'columns_diff_exclude_unchanged_toast' instructs the connector to refresh the in-memory schema cache if there is a discrepancy between it and the schema derived from the incoming message, unless unchanged TOASTable data fully accounts [...]
diff --git a/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverComponentConfigurer.java b/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverComponentConfigurer.java
index d6a7990..76e77a3 100644
--- a/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverComponentConfigurer.java
+++ b/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverComponentConfigurer.java
@@ -63,6 +63,8 @@ public class DebeziumSqlserverComponentConfigurer extends PropertyConfigurerSupp
case "databaseUser": getOrCreateConfiguration(target).setDatabaseUser(property(camelContext, java.lang.String.class, value)); return true;
case "decimalhandlingmode":
case "decimalHandlingMode": getOrCreateConfiguration(target).setDecimalHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": getOrCreateConfiguration(target).setEventProcessingFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
case "heartbeatintervalms":
case "heartbeatIntervalMs": getOrCreateConfiguration(target).setHeartbeatIntervalMs(property(camelContext, int.class, value)); return true;
case "heartbeattopicsprefix":
@@ -95,6 +97,8 @@ public class DebeziumSqlserverComponentConfigurer extends PropertyConfigurerSupp
case "offsetStorageTopic": getOrCreateConfiguration(target).setOffsetStorageTopic(property(camelContext, java.lang.String.class, value)); return true;
case "pollintervalms":
case "pollIntervalMs": getOrCreateConfiguration(target).setPollIntervalMs(property(camelContext, long.class, value)); return true;
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": getOrCreateConfiguration(target).setProvideTransactionMetadata(property(camelContext, boolean.class, value)); return true;
case "snapshotdelayms":
case "snapshotDelayMs": getOrCreateConfiguration(target).setSnapshotDelayMs(property(camelContext, long.class, value)); return true;
case "snapshotfetchsize":
@@ -143,6 +147,7 @@ public class DebeziumSqlserverComponentConfigurer extends PropertyConfigurerSupp
answer.put("databaseServerTimezone", java.lang.String.class);
answer.put("databaseUser", java.lang.String.class);
answer.put("decimalHandlingMode", java.lang.String.class);
+ answer.put("eventProcessingFailureHandlingMode", java.lang.String.class);
answer.put("heartbeatIntervalMs", int.class);
answer.put("heartbeatTopicsPrefix", java.lang.String.class);
answer.put("internalKeyConverter", java.lang.String.class);
@@ -159,6 +164,7 @@ public class DebeziumSqlserverComponentConfigurer extends PropertyConfigurerSupp
answer.put("offsetStorageReplicationFactor", int.class);
answer.put("offsetStorageTopic", java.lang.String.class);
answer.put("pollIntervalMs", long.class);
+ answer.put("provideTransactionMetadata", boolean.class);
answer.put("snapshotDelayMs", long.class);
answer.put("snapshotFetchSize", int.class);
answer.put("snapshotLockTimeoutMs", long.class);
@@ -214,6 +220,8 @@ public class DebeziumSqlserverComponentConfigurer extends PropertyConfigurerSupp
case "databaseUser": return getOrCreateConfiguration(target).getDatabaseUser();
case "decimalhandlingmode":
case "decimalHandlingMode": return getOrCreateConfiguration(target).getDecimalHandlingMode();
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": return getOrCreateConfiguration(target).getEventProcessingFailureHandlingMode();
case "heartbeatintervalms":
case "heartbeatIntervalMs": return getOrCreateConfiguration(target).getHeartbeatIntervalMs();
case "heartbeattopicsprefix":
@@ -246,6 +254,8 @@ public class DebeziumSqlserverComponentConfigurer extends PropertyConfigurerSupp
case "offsetStorageTopic": return getOrCreateConfiguration(target).getOffsetStorageTopic();
case "pollintervalms":
case "pollIntervalMs": return getOrCreateConfiguration(target).getPollIntervalMs();
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": return getOrCreateConfiguration(target).isProvideTransactionMetadata();
case "snapshotdelayms":
case "snapshotDelayMs": return getOrCreateConfiguration(target).getSnapshotDelayMs();
case "snapshotfetchsize":
diff --git a/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverEndpointConfigurer.java b/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverEndpointConfigurer.java
index 99898db..440027a 100644
--- a/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverEndpointConfigurer.java
+++ b/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/DebeziumSqlserverEndpointConfigurer.java
@@ -55,6 +55,8 @@ public class DebeziumSqlserverEndpointConfigurer extends PropertyConfigurerSuppo
case "databaseUser": target.getConfiguration().setDatabaseUser(property(camelContext, java.lang.String.class, value)); return true;
case "decimalhandlingmode":
case "decimalHandlingMode": target.getConfiguration().setDecimalHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": target.getConfiguration().setEventProcessingFailureHandlingMode(property(camelContext, java.lang.String.class, value)); return true;
case "exceptionhandler":
case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true;
case "exchangepattern":
@@ -91,6 +93,8 @@ public class DebeziumSqlserverEndpointConfigurer extends PropertyConfigurerSuppo
case "offsetStorageTopic": target.getConfiguration().setOffsetStorageTopic(property(camelContext, java.lang.String.class, value)); return true;
case "pollintervalms":
case "pollIntervalMs": target.getConfiguration().setPollIntervalMs(property(camelContext, long.class, value)); return true;
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": target.getConfiguration().setProvideTransactionMetadata(property(camelContext, boolean.class, value)); return true;
case "snapshotdelayms":
case "snapshotDelayMs": target.getConfiguration().setSnapshotDelayMs(property(camelContext, long.class, value)); return true;
case "snapshotfetchsize":
@@ -139,6 +143,7 @@ public class DebeziumSqlserverEndpointConfigurer extends PropertyConfigurerSuppo
answer.put("databaseServerTimezone", java.lang.String.class);
answer.put("databaseUser", java.lang.String.class);
answer.put("decimalHandlingMode", java.lang.String.class);
+ answer.put("eventProcessingFailureHandlingMode", java.lang.String.class);
answer.put("exceptionHandler", org.apache.camel.spi.ExceptionHandler.class);
answer.put("exchangePattern", org.apache.camel.ExchangePattern.class);
answer.put("heartbeatIntervalMs", int.class);
@@ -157,6 +162,7 @@ public class DebeziumSqlserverEndpointConfigurer extends PropertyConfigurerSuppo
answer.put("offsetStorageReplicationFactor", int.class);
answer.put("offsetStorageTopic", java.lang.String.class);
answer.put("pollIntervalMs", long.class);
+ answer.put("provideTransactionMetadata", boolean.class);
answer.put("snapshotDelayMs", long.class);
answer.put("snapshotFetchSize", int.class);
answer.put("snapshotLockTimeoutMs", long.class);
@@ -212,6 +218,8 @@ public class DebeziumSqlserverEndpointConfigurer extends PropertyConfigurerSuppo
case "databaseUser": return target.getConfiguration().getDatabaseUser();
case "decimalhandlingmode":
case "decimalHandlingMode": return target.getConfiguration().getDecimalHandlingMode();
+ case "eventprocessingfailurehandlingmode":
+ case "eventProcessingFailureHandlingMode": return target.getConfiguration().getEventProcessingFailureHandlingMode();
case "exceptionhandler":
case "exceptionHandler": return target.getExceptionHandler();
case "exchangepattern":
@@ -248,6 +256,8 @@ public class DebeziumSqlserverEndpointConfigurer extends PropertyConfigurerSuppo
case "offsetStorageTopic": return target.getConfiguration().getOffsetStorageTopic();
case "pollintervalms":
case "pollIntervalMs": return target.getConfiguration().getPollIntervalMs();
+ case "providetransactionmetadata":
+ case "provideTransactionMetadata": return target.getConfiguration().isProvideTransactionMetadata();
case "snapshotdelayms":
case "snapshotDelayMs": return target.getConfiguration().getSnapshotDelayMs();
case "snapshotfetchsize":
diff --git a/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/configuration/SqlServerConnectorEmbeddedDebeziumConfiguration.java b/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/configuration/SqlServerConnectorEmbeddedDebeziumConfiguration.java
index cf066a6..ea5ae1b 100644
--- a/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/configuration/SqlServerConnectorEmbeddedDebeziumConfiguration.java
+++ b/components/camel-debezium-sqlserver/src/generated/java/org/apache/camel/component/debezium/configuration/SqlServerConnectorEmbeddedDebeziumConfiguration.java
@@ -26,6 +26,8 @@ public class SqlServerConnectorEmbeddedDebeziumConfiguration
private int databaseHistoryKafkaRecoveryAttempts = 100;
@UriParam(label = LABEL_NAME)
private String tableBlacklist;
+ @UriParam(label = LABEL_NAME, defaultValue = "false")
+ private boolean provideTransactionMetadata = false;
@UriParam(label = LABEL_NAME)
private String tableWhitelist;
@UriParam(label = LABEL_NAME)
@@ -65,6 +67,8 @@ public class SqlServerConnectorEmbeddedDebeziumConfiguration
private int heartbeatIntervalMs = 0;
@UriParam(label = LABEL_NAME, defaultValue = "v2")
private String sourceStructVersion = "v2";
+ @UriParam(label = LABEL_NAME, defaultValue = "fail")
+ private String eventProcessingFailureHandlingMode = "fail";
@UriParam(label = LABEL_NAME, defaultValue = "1433")
private int databasePort = 1433;
@UriParam(label = LABEL_NAME)
@@ -172,6 +176,17 @@ public class SqlServerConnectorEmbeddedDebeziumConfiguration
}
/**
+ * Enables transaction metadata extraction together with event counting
+ */
+ public void setProvideTransactionMetadata(boolean provideTransactionMetadata) {
+ this.provideTransactionMetadata = provideTransactionMetadata;
+ }
+
+ public boolean isProvideTransactionMetadata() {
+ return provideTransactionMetadata;
+ }
+
+ /**
* The tables for which changes are to be captured
*/
public void setTableWhitelist(String tableWhitelist) {
@@ -431,6 +446,23 @@ public class SqlServerConnectorEmbeddedDebeziumConfiguration
}
/**
+ * Specify how failures during processing of events (i.e. when encountering
+ * a corrupted event) should be handled, including:'fail' (the default) an
+ * exception indicating the problematic event and its position is raised,
+ * causing the connector to be stopped; 'warn' the problematic event and its
+ * position will be logged and the event will be skipped;'ignore' the
+ * problematic event will be skipped.
+ */
+ public void setEventProcessingFailureHandlingMode(
+ String eventProcessingFailureHandlingMode) {
+ this.eventProcessingFailureHandlingMode = eventProcessingFailureHandlingMode;
+ }
+
+ public String getEventProcessingFailureHandlingMode() {
+ return eventProcessingFailureHandlingMode;
+ }
+
+ /**
* Port of the SQL Server database server.
*/
public void setDatabasePort(int databasePort) {
@@ -514,6 +546,7 @@ public class SqlServerConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "snapshot.delay.ms", snapshotDelayMs);
addPropertyIfNotNull(configBuilder, "database.history.kafka.recovery.attempts", databaseHistoryKafkaRecoveryAttempts);
addPropertyIfNotNull(configBuilder, "table.blacklist", tableBlacklist);
+ addPropertyIfNotNull(configBuilder, "provide.transaction.metadata", provideTransactionMetadata);
addPropertyIfNotNull(configBuilder, "table.whitelist", tableWhitelist);
addPropertyIfNotNull(configBuilder, "database.server.timezone", databaseServerTimezone);
addPropertyIfNotNull(configBuilder, "tombstones.on.delete", tombstonesOnDelete);
@@ -533,6 +566,7 @@ public class SqlServerConnectorEmbeddedDebeziumConfiguration
addPropertyIfNotNull(configBuilder, "database.server.name", databaseServerName);
addPropertyIfNotNull(configBuilder, "heartbeat.interval.ms", heartbeatIntervalMs);
addPropertyIfNotNull(configBuilder, "source.struct.version", sourceStructVersion);
+ addPropertyIfNotNull(configBuilder, "event.processing.failure.handling.mode", eventProcessingFailureHandlingMode);
addPropertyIfNotNull(configBuilder, "database.port", databasePort);
addPropertyIfNotNull(configBuilder, "database.hostname", databaseHostname);
addPropertyIfNotNull(configBuilder, "database.password", databasePassword);
diff --git a/components/camel-debezium-sqlserver/src/generated/resources/org/apache/camel/component/debezium/debezium-sqlserver.json b/components/camel-debezium-sqlserver/src/generated/resources/org/apache/camel/component/debezium/debezium-sqlserver.json
index 98c00bf..8b0bcde 100644
--- a/components/camel-debezium-sqlserver/src/generated/resources/org/apache/camel/component/debezium/debezium-sqlserver.json
+++ b/components/camel-debezium-sqlserver/src/generated/resources/org/apache/camel/component/debezium/debezium-sqlserver.json
@@ -48,12 +48,14 @@
"databaseServerTimezone": { "kind": "property", "displayName": "Database Server Timezone", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The timezone of the server used to correctly shift the comm [...]
"databaseUser": { "kind": "property", "displayName": "Database User", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Name of the SQL Server database user to be used when connecting to the database." },
"decimalHandlingMode": { "kind": "property", "displayName": "Decimal Handling Mode", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "precise", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how DECIMAL and NUMERIC column [...]
+ "eventProcessingFailureHandlingMode": { "kind": "property", "displayName": "Event Processing Failure Handling Mode", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify h [...]
"heartbeatIntervalMs": { "kind": "property", "displayName": "Heartbeat Interval Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Length of an interval in milli-seconds in in which the c [...]
"heartbeatTopicsPrefix": { "kind": "property", "displayName": "Heartbeat Topics Prefix", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is us [...]
"maxBatchSize": { "kind": "property", "displayName": "Max Batch Size", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
"maxQueueSize": { "kind": "property", "displayName": "Max Queue Size", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "8192", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of the queue for change events read from the database [...]
"messageKeyColumns": { "kind": "property", "displayName": "Message Key Columns", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "A semicolon-separated list of expressions that match fully-qualified [...]
"pollIntervalMs": { "kind": "property", "displayName": "Poll Interval Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Frequency in milliseconds to wait for new change events to appe [...]
+ "provideTransactionMetadata": { "kind": "property", "displayName": "Provide Transaction Metadata", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Enables transaction metadata extra [...]
"snapshotDelayMs": { "kind": "property", "displayName": "Snapshot Delay Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The number of milliseconds to delay before a snapshot will begin." },
"snapshotFetchSize": { "kind": "property", "displayName": "Snapshot Fetch Size", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The maximum number of records that should be loaded into memory while performing [...]
"snapshotLockTimeoutMs": { "kind": "property", "displayName": "Snapshot Lock Timeout Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "10000", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The maximum number of millis to wait for table [...]
@@ -99,12 +101,14 @@
"databaseServerTimezone": { "kind": "parameter", "displayName": "Database Server Timezone", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The timezone of the server used to correctly shift the com [...]
"databaseUser": { "kind": "parameter", "displayName": "Database User", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Name of the SQL Server database user to be used when connecting to the database." },
"decimalHandlingMode": { "kind": "parameter", "displayName": "Decimal Handling Mode", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "precise", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify how DECIMAL and NUMERIC colum [...]
+ "eventProcessingFailureHandlingMode": { "kind": "parameter", "displayName": "Event Processing Failure Handling Mode", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "fail", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Specify [...]
"heartbeatIntervalMs": { "kind": "parameter", "displayName": "Heartbeat Interval Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Length of an interval in milli-seconds in in which the [...]
"heartbeatTopicsPrefix": { "kind": "parameter", "displayName": "Heartbeat Topics Prefix", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "__debezium-heartbeat", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The prefix that is u [...]
"maxBatchSize": { "kind": "parameter", "displayName": "Max Batch Size", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "2048", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of each batch of source records. Defaults to 2048." },
"maxQueueSize": { "kind": "parameter", "displayName": "Max Queue Size", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "8192", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Maximum size of the queue for change events read from the database [...]
"messageKeyColumns": { "kind": "parameter", "displayName": "Message Key Columns", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "A semicolon-separated list of expressions that match fully-qualified [...]
"pollIntervalMs": { "kind": "parameter", "displayName": "Poll Interval Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Frequency in milliseconds to wait for new change events to app [...]
+ "provideTransactionMetadata": { "kind": "parameter", "displayName": "Provide Transaction Metadata", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "Enables transaction metadata extr [...]
"snapshotDelayMs": { "kind": "parameter", "displayName": "Snapshot Delay Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The number of milliseconds to delay before a snapshot will begin." },
"snapshotFetchSize": { "kind": "parameter", "displayName": "Snapshot Fetch Size", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The maximum number of records that should be loaded into memory while performing [...]
"snapshotLockTimeoutMs": { "kind": "parameter", "displayName": "Snapshot Lock Timeout Ms", "group": "sqlserver", "label": "consumer,sqlserver", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "10000", "configurationClass": "org.apache.camel.component.debezium.configuration.SqlServerConnectorEmbeddedDebeziumConfiguration", "configurationField": "configuration", "description": "The maximum number of millis to wait for tabl [...]
diff --git a/components/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc b/components/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
index e289f7d..016a22a 100644
--- a/components/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
+++ b/components/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
@@ -39,7 +39,7 @@ debezium-sqlserver:name[?options]
// component options: START
-The Debezium SQL Server Connector component supports 46 options, which are listed below.
+The Debezium SQL Server Connector component supports 48 options, which are listed below.
@@ -75,12 +75,14 @@ The Debezium SQL Server Connector component supports 46 options, which are liste
| *databaseServerTimezone* (sqlserver) | The timezone of the server used to correctly shift the commit transaction timestamp on the client sideOptions include: Any valid Java ZoneId | | String
| *databaseUser* (sqlserver) | Name of the SQL Server database user to be used when connecting to the database. | | String
| *decimalHandlingMode* (sqlserver) | Specify how DECIMAL and NUMERIC columns should be represented in change events, including:'precise' (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect's 'org.apache.kafka.connect.data.Decimal' type; 'string' uses string to represent values; 'double' represents values using Java's 'double', which may not offer the precision but will be far easier to use in [...]
+| *eventProcessingFailureHandling Mode* (sqlserver) | Specify how failures during processing of events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its position is raised, causing the connector to be stopped; 'warn' the problematic event and its position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
| *heartbeatIntervalMs* (sqlserver) | Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default. | 0 | int
| *heartbeatTopicsPrefix* (sqlserver) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
| *maxBatchSize* (sqlserver) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
| *maxQueueSize* (sqlserver) | Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size. | 8192 | int
| *messageKeyColumns* (sqlserver) | A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ':',where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector,and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table's primary key colum [...]
| *pollIntervalMs* (sqlserver) | Frequency in milliseconds to wait for new change events to appear after receiving no events. Defaults to 500ms. | 500 | long
+| *provideTransactionMetadata* (sqlserver) | Enables transaction metadata extraction together with event counting | false | boolean
| *snapshotDelayMs* (sqlserver) | The number of milliseconds to delay before a snapshot will begin. | 0 | long
| *snapshotFetchSize* (sqlserver) | The maximum number of records that should be loaded into memory while performing a snapshot | | int
| *snapshotLockTimeoutMs* (sqlserver) | The maximum number of millis to wait for table locks at the beginning of a snapshot. If locks cannot be acquired in this time frame, the snapshot will be aborted. Defaults to 10 seconds | 10000 | long
@@ -115,7 +117,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (48 parameters):
+=== Query Parameters (50 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -152,12 +154,14 @@ with the following path and query parameters:
| *databaseServerTimezone* (sqlserver) | The timezone of the server used to correctly shift the commit transaction timestamp on the client sideOptions include: Any valid Java ZoneId | | String
| *databaseUser* (sqlserver) | Name of the SQL Server database user to be used when connecting to the database. | | String
| *decimalHandlingMode* (sqlserver) | Specify how DECIMAL and NUMERIC columns should be represented in change events, including:'precise' (the default) uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect's 'org.apache.kafka.connect.data.Decimal' type; 'string' uses string to represent values; 'double' represents values using Java's 'double', which may not offer the precision but will be far easier to use in [...]
+| *eventProcessingFailureHandling Mode* (sqlserver) | Specify how failures during processing of events (i.e. when encountering a corrupted event) should be handled, including:'fail' (the default) an exception indicating the problematic event and its position is raised, causing the connector to be stopped; 'warn' the problematic event and its position will be logged and the event will be skipped;'ignore' the problematic event will be skipped. | fail | String
| *heartbeatIntervalMs* (sqlserver) | Length of an interval in milli-seconds in in which the connector periodically sends heartbeat messages to a heartbeat topic. Use 0 to disable heartbeat messages. Disabled by default. | 0 | int
| *heartbeatTopicsPrefix* (sqlserver) | The prefix that is used to name heartbeat topics.Defaults to __debezium-heartbeat. | __debezium-heartbeat | String
| *maxBatchSize* (sqlserver) | Maximum size of each batch of source records. Defaults to 2048. | 2048 | int
| *maxQueueSize* (sqlserver) | Maximum size of the queue for change events read from the database log but not yet recorded or forwarded. Defaults to 8192, and should always be larger than the maximum batch size. | 8192 | int
| *messageKeyColumns* (sqlserver) | A semicolon-separated list of expressions that match fully-qualified tables and column(s) to be used as message key. Each expression must match the pattern ':',where the table names could be defined as (DB_NAME.TABLE_NAME) or (SCHEMA_NAME.TABLE_NAME), depending on the specific connector,and the key columns are a comma-separated list of columns representing the custom key. For any table without an explicit key configuration the table's primary key colum [...]
| *pollIntervalMs* (sqlserver) | Frequency in milliseconds to wait for new change events to appear after receiving no events. Defaults to 500ms. | 500 | long
+| *provideTransactionMetadata* (sqlserver) | Enables transaction metadata extraction together with event counting | false | boolean
| *snapshotDelayMs* (sqlserver) | The number of milliseconds to delay before a snapshot will begin. | 0 | long
| *snapshotFetchSize* (sqlserver) | The maximum number of records that should be loaded into memory while performing a snapshot | | int
| *snapshotLockTimeoutMs* (sqlserver) | The maximum number of millis to wait for table locks at the beginning of a snapshot. If locks cannot be acquired in this time frame, the snapshot will be aborted. Defaults to 10 seconds | 10000 | long
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMongodbComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMongodbComponentBuilderFactory.java
index 35a63b8..ea3f326 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMongodbComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMongodbComponentBuilderFactory.java
@@ -422,7 +422,7 @@ public interface DebeziumMongodbComponentBuilderFactory {
return this;
}
/**
- * Maximum number of threads used to perform an intial sync of the
+ * Maximum number of threads used to perform an initial sync of the
* collections in a replica set. Defaults to 1.
*
* The option is a: <code>int</code> type.
@@ -571,6 +571,20 @@ public interface DebeziumMongodbComponentBuilderFactory {
return this;
}
/**
+ * The comma-separated list of operations to skip during streaming,
+ * defined as: 'i' for inserts; 'u' for updates; 'd' for deletes. By
+ * default, no operations will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: mongodb
+ */
+ default DebeziumMongodbComponentBuilder skippedOperations(
+ java.lang.String skippedOperations) {
+ doSetProperty("skippedOperations", skippedOperations);
+ return this;
+ }
+ /**
* The number of milliseconds to delay before a snapshot will begin.
*
* The option is a: <code>long</code> type.
@@ -704,6 +718,7 @@ public interface DebeziumMongodbComponentBuilderFactory {
case "mongodbSslInvalidHostnameAllowed": getOrCreateConfiguration((DebeziumMongodbComponent) component).setMongodbSslInvalidHostnameAllowed((boolean) value); return true;
case "mongodbUser": getOrCreateConfiguration((DebeziumMongodbComponent) component).setMongodbUser((java.lang.String) value); return true;
case "pollIntervalMs": getOrCreateConfiguration((DebeziumMongodbComponent) component).setPollIntervalMs((long) value); return true;
+ case "skippedOperations": getOrCreateConfiguration((DebeziumMongodbComponent) component).setSkippedOperations((java.lang.String) value); return true;
case "snapshotDelayMs": getOrCreateConfiguration((DebeziumMongodbComponent) component).setSnapshotDelayMs((long) value); return true;
case "snapshotFetchSize": getOrCreateConfiguration((DebeziumMongodbComponent) component).setSnapshotFetchSize((int) value); return true;
case "snapshotMode": getOrCreateConfiguration((DebeziumMongodbComponent) component).setSnapshotMode((java.lang.String) value); return true;
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMysqlComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMysqlComponentBuilderFactory.java
index 7d77dde..2348c48 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMysqlComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumMysqlComponentBuilderFactory.java
@@ -753,14 +753,32 @@ public interface DebeziumMysqlComponentBuilderFactory {
return this;
}
/**
+ * Specify how failures during processing of events (i.e. when
+ * encountering a corrupted event) should be handled, including:'fail'
+ * (the default) an exception indicating the problematic event and its
+ * position is raised, causing the connector to be stopped; 'warn' the
+ * problematic event and its position will be logged and the event will
+ * be skipped;'ignore' the problematic event will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: fail
+ * Group: mysql
+ */
+ default DebeziumMysqlComponentBuilder eventProcessingFailureHandlingMode(
+ java.lang.String eventProcessingFailureHandlingMode) {
+ doSetProperty("eventProcessingFailureHandlingMode", eventProcessingFailureHandlingMode);
+ return this;
+ }
+ /**
* If set to 'latest', when connector sees new GTID, it will start
* consuming gtid channel from the server latest executed gtid position.
- * If 'earliest' connector starts reading channel from first available
- * (not purged) gtid position on the server.
+ * If 'earliest' (the default) connector starts reading channel from
+ * first available (not purged) gtid position on the server.
*
* The option is a: <code>java.lang.String</code> type.
*
- * Default: latest
+ * Default: earliest
* Group: mysql
*/
default DebeziumMysqlComponentBuilder gtidNewChannelPosition(
@@ -882,7 +900,7 @@ public interface DebeziumMysqlComponentBuilderFactory {
* default) an exception indicating the problematic event and its binlog
* position is raised, causing the connector to be stopped; 'warn' the
* problematic event and its binlog position will be logged and the
- * event will be skipped;'ignore' the problematic event will be skipped.
+ * event will be skipped;'skip' the problematic event will be skipped.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -1236,6 +1254,7 @@ public interface DebeziumMysqlComponentBuilderFactory {
case "decimalHandlingMode": getOrCreateConfiguration((DebeziumMySqlComponent) component).setDecimalHandlingMode((java.lang.String) value); return true;
case "enableTimeAdjuster": getOrCreateConfiguration((DebeziumMySqlComponent) component).setEnableTimeAdjuster((boolean) value); return true;
case "eventDeserializationFailureHandlingMode": getOrCreateConfiguration((DebeziumMySqlComponent) component).setEventDeserializationFailureHandlingMode((java.lang.String) value); return true;
+ case "eventProcessingFailureHandlingMode": getOrCreateConfiguration((DebeziumMySqlComponent) component).setEventProcessingFailureHandlingMode((java.lang.String) value); return true;
case "gtidNewChannelPosition": getOrCreateConfiguration((DebeziumMySqlComponent) component).setGtidNewChannelPosition((java.lang.String) value); return true;
case "gtidSourceExcludes": getOrCreateConfiguration((DebeziumMySqlComponent) component).setGtidSourceExcludes((java.lang.String) value); return true;
case "gtidSourceFilterDmlEvents": getOrCreateConfiguration((DebeziumMySqlComponent) component).setGtidSourceFilterDmlEvents((boolean) value); return true;
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumPostgresComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumPostgresComponentBuilderFactory.java
index 3407117..1f54fa0 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumPostgresComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumPostgresComponentBuilderFactory.java
@@ -499,6 +499,36 @@ public interface DebeziumPostgresComponentBuilderFactory {
return this;
}
/**
+ * Specify how failures during processing of events (i.e. when
+ * encountering a corrupted event) should be handled, including:'fail'
+ * (the default) an exception indicating the problematic event and its
+ * position is raised, causing the connector to be stopped; 'warn' the
+ * problematic event and its position will be logged and the event will
+ * be skipped;'ignore' the problematic event will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: fail
+ * Group: postgres
+ */
+ default DebeziumPostgresComponentBuilder eventProcessingFailureHandlingMode(
+ java.lang.String eventProcessingFailureHandlingMode) {
+ doSetProperty("eventProcessingFailureHandlingMode", eventProcessingFailureHandlingMode);
+ return this;
+ }
+ /**
+ * The query executed with every heartbeat. Defaults to an empty string.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: postgres
+ */
+ default DebeziumPostgresComponentBuilder heartbeatActionQuery(
+ java.lang.String heartbeatActionQuery) {
+ doSetProperty("heartbeatActionQuery", heartbeatActionQuery);
+ return this;
+ }
+ /**
* Length of an interval in milli-seconds in in which the connector
* periodically sends heartbeat messages to a heartbeat topic. Use 0 to
* disable heartbeat messages. Disabled by default.
@@ -529,8 +559,8 @@ public interface DebeziumPostgresComponentBuilderFactory {
}
/**
* Specify how HSTORE columns should be represented in change events,
- * including:'json' represents values as json string'map' (default)
- * represents values using java.util.Map.
+ * including:'json' represents values as string-ified JSON
+ * (default)'map' represents values as a key/value map.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -650,6 +680,19 @@ public interface DebeziumPostgresComponentBuilderFactory {
return this;
}
/**
+ * Enables transaction metadata extraction together with event counting.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: postgres
+ */
+ default DebeziumPostgresComponentBuilder provideTransactionMetadata(
+ boolean provideTransactionMetadata) {
+ doSetProperty("provideTransactionMetadata", provideTransactionMetadata);
+ return this;
+ }
+ /**
* The name of the Postgres 10 publication used for streaming changes
* from a plugin.Defaults to 'dbz_publication'.
*
@@ -1068,6 +1111,8 @@ public interface DebeziumPostgresComponentBuilderFactory {
case "databaseTcpkeepalive": getOrCreateConfiguration((DebeziumPostgresComponent) component).setDatabaseTcpkeepalive((boolean) value); return true;
case "databaseUser": getOrCreateConfiguration((DebeziumPostgresComponent) component).setDatabaseUser((java.lang.String) value); return true;
case "decimalHandlingMode": getOrCreateConfiguration((DebeziumPostgresComponent) component).setDecimalHandlingMode((java.lang.String) value); return true;
+ case "eventProcessingFailureHandlingMode": getOrCreateConfiguration((DebeziumPostgresComponent) component).setEventProcessingFailureHandlingMode((java.lang.String) value); return true;
+ case "heartbeatActionQuery": getOrCreateConfiguration((DebeziumPostgresComponent) component).setHeartbeatActionQuery((java.lang.String) value); return true;
case "heartbeatIntervalMs": getOrCreateConfiguration((DebeziumPostgresComponent) component).setHeartbeatIntervalMs((int) value); return true;
case "heartbeatTopicsPrefix": getOrCreateConfiguration((DebeziumPostgresComponent) component).setHeartbeatTopicsPrefix((java.lang.String) value); return true;
case "hstoreHandlingMode": getOrCreateConfiguration((DebeziumPostgresComponent) component).setHstoreHandlingMode((java.lang.String) value); return true;
@@ -1078,6 +1123,7 @@ public interface DebeziumPostgresComponentBuilderFactory {
case "messageKeyColumns": getOrCreateConfiguration((DebeziumPostgresComponent) component).setMessageKeyColumns((java.lang.String) value); return true;
case "pluginName": getOrCreateConfiguration((DebeziumPostgresComponent) component).setPluginName((java.lang.String) value); return true;
case "pollIntervalMs": getOrCreateConfiguration((DebeziumPostgresComponent) component).setPollIntervalMs((long) value); return true;
+ case "provideTransactionMetadata": getOrCreateConfiguration((DebeziumPostgresComponent) component).setProvideTransactionMetadata((boolean) value); return true;
case "publicationName": getOrCreateConfiguration((DebeziumPostgresComponent) component).setPublicationName((java.lang.String) value); return true;
case "schemaBlacklist": getOrCreateConfiguration((DebeziumPostgresComponent) component).setSchemaBlacklist((java.lang.String) value); return true;
case "schemaRefreshMode": getOrCreateConfiguration((DebeziumPostgresComponent) component).setSchemaRefreshMode((java.lang.String) value); return true;
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumSqlserverComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumSqlserverComponentBuilderFactory.java
index a59bb09..a21a007 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumSqlserverComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/DebeziumSqlserverComponentBuilderFactory.java
@@ -469,6 +469,24 @@ public interface DebeziumSqlserverComponentBuilderFactory {
return this;
}
/**
+ * Specify how failures during processing of events (i.e. when
+ * encountering a corrupted event) should be handled, including:'fail'
+ * (the default) an exception indicating the problematic event and its
+ * position is raised, causing the connector to be stopped; 'warn' the
+ * problematic event and its position will be logged and the event will
+ * be skipped;'ignore' the problematic event will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: fail
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverComponentBuilder eventProcessingFailureHandlingMode(
+ java.lang.String eventProcessingFailureHandlingMode) {
+ doSetProperty("eventProcessingFailureHandlingMode", eventProcessingFailureHandlingMode);
+ return this;
+ }
+ /**
* Length of an interval in milli-seconds in in which the connector
* periodically sends heartbeat messages to a heartbeat topic. Use 0 to
* disable heartbeat messages. Disabled by default.
@@ -558,6 +576,19 @@ public interface DebeziumSqlserverComponentBuilderFactory {
return this;
}
/**
+ * Enables transaction metadata extraction together with event counting.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverComponentBuilder provideTransactionMetadata(
+ boolean provideTransactionMetadata) {
+ doSetProperty("provideTransactionMetadata", provideTransactionMetadata);
+ return this;
+ }
+ /**
* The number of milliseconds to delay before a snapshot will begin.
*
* The option is a: <code>long</code> type.
@@ -781,12 +812,14 @@ public interface DebeziumSqlserverComponentBuilderFactory {
case "databaseServerTimezone": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setDatabaseServerTimezone((java.lang.String) value); return true;
case "databaseUser": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setDatabaseUser((java.lang.String) value); return true;
case "decimalHandlingMode": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setDecimalHandlingMode((java.lang.String) value); return true;
+ case "eventProcessingFailureHandlingMode": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setEventProcessingFailureHandlingMode((java.lang.String) value); return true;
case "heartbeatIntervalMs": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setHeartbeatIntervalMs((int) value); return true;
case "heartbeatTopicsPrefix": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setHeartbeatTopicsPrefix((java.lang.String) value); return true;
case "maxBatchSize": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setMaxBatchSize((int) value); return true;
case "maxQueueSize": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setMaxQueueSize((int) value); return true;
case "messageKeyColumns": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setMessageKeyColumns((java.lang.String) value); return true;
case "pollIntervalMs": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setPollIntervalMs((long) value); return true;
+ case "provideTransactionMetadata": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setProvideTransactionMetadata((boolean) value); return true;
case "snapshotDelayMs": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setSnapshotDelayMs((long) value); return true;
case "snapshotFetchSize": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setSnapshotFetchSize((int) value); return true;
case "snapshotLockTimeoutMs": getOrCreateConfiguration((DebeziumSqlserverComponent) component).setSnapshotLockTimeoutMs((long) value); return true;
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
index 8e533f0..5a9df0d 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
@@ -541,7 +541,7 @@ public interface DebeziumMongodbEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of threads used to perform an intial sync of the
+ * Maximum number of threads used to perform an initial sync of the
* collections in a replica set. Defaults to 1.
*
* The option is a: <code>int</code> type.
@@ -555,7 +555,7 @@ public interface DebeziumMongodbEndpointBuilderFactory {
return this;
}
/**
- * Maximum number of threads used to perform an intial sync of the
+ * Maximum number of threads used to perform an initial sync of the
* collections in a replica set. Defaults to 1.
*
* The option will be converted to a <code>int</code> type.
@@ -786,6 +786,20 @@ public interface DebeziumMongodbEndpointBuilderFactory {
return this;
}
/**
+ * The comma-separated list of operations to skip during streaming,
+ * defined as: 'i' for inserts; 'u' for updates; 'd' for deletes. By
+ * default, no operations will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: mongodb
+ */
+ default DebeziumMongodbEndpointBuilder skippedOperations(
+ String skippedOperations) {
+ doSetProperty("skippedOperations", skippedOperations);
+ return this;
+ }
+ /**
* The number of milliseconds to delay before a snapshot will begin.
*
* The option is a: <code>long</code> type.
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMySqlEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMySqlEndpointBuilderFactory.java
index 86b58de..e6bdf75 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMySqlEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMySqlEndpointBuilderFactory.java
@@ -991,14 +991,32 @@ public interface DebeziumMySqlEndpointBuilderFactory {
return this;
}
/**
+ * Specify how failures during processing of events (i.e. when
+ * encountering a corrupted event) should be handled, including:'fail'
+ * (the default) an exception indicating the problematic event and its
+ * position is raised, causing the connector to be stopped; 'warn' the
+ * problematic event and its position will be logged and the event will
+ * be skipped;'ignore' the problematic event will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: fail
+ * Group: mysql
+ */
+ default DebeziumMySqlEndpointBuilder eventProcessingFailureHandlingMode(
+ String eventProcessingFailureHandlingMode) {
+ doSetProperty("eventProcessingFailureHandlingMode", eventProcessingFailureHandlingMode);
+ return this;
+ }
+ /**
* If set to 'latest', when connector sees new GTID, it will start
* consuming gtid channel from the server latest executed gtid position.
- * If 'earliest' connector starts reading channel from first available
- * (not purged) gtid position on the server.
+ * If 'earliest' (the default) connector starts reading channel from
+ * first available (not purged) gtid position on the server.
*
* The option is a: <code>java.lang.String</code> type.
*
- * Default: latest
+ * Default: earliest
* Group: mysql
*/
default DebeziumMySqlEndpointBuilder gtidNewChannelPosition(
@@ -1187,7 +1205,7 @@ public interface DebeziumMySqlEndpointBuilderFactory {
* default) an exception indicating the problematic event and its binlog
* position is raised, causing the connector to be stopped; 'warn' the
* problematic event and its binlog position will be logged and the
- * event will be skipped;'ignore' the problematic event will be skipped.
+ * event will be skipped;'skip' the problematic event will be skipped.
*
* The option is a: <code>java.lang.String</code> type.
*
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
index 929fcf5..3b13c9c 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
@@ -584,6 +584,36 @@ public interface DebeziumPostgresEndpointBuilderFactory {
return this;
}
/**
+ * Specify how failures during processing of events (i.e. when
+ * encountering a corrupted event) should be handled, including:'fail'
+ * (the default) an exception indicating the problematic event and its
+ * position is raised, causing the connector to be stopped; 'warn' the
+ * problematic event and its position will be logged and the event will
+ * be skipped;'ignore' the problematic event will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: fail
+ * Group: postgres
+ */
+ default DebeziumPostgresEndpointBuilder eventProcessingFailureHandlingMode(
+ String eventProcessingFailureHandlingMode) {
+ doSetProperty("eventProcessingFailureHandlingMode", eventProcessingFailureHandlingMode);
+ return this;
+ }
+ /**
+ * The query executed with every heartbeat. Defaults to an empty string.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: postgres
+ */
+ default DebeziumPostgresEndpointBuilder heartbeatActionQuery(
+ String heartbeatActionQuery) {
+ doSetProperty("heartbeatActionQuery", heartbeatActionQuery);
+ return this;
+ }
+ /**
* Length of an interval in milli-seconds in in which the connector
* periodically sends heartbeat messages to a heartbeat topic. Use 0 to
* disable heartbeat messages. Disabled by default.
@@ -629,8 +659,8 @@ public interface DebeziumPostgresEndpointBuilderFactory {
}
/**
* Specify how HSTORE columns should be represented in change events,
- * including:'json' represents values as json string'map' (default)
- * represents values using java.util.Map.
+ * including:'json' represents values as string-ified JSON
+ * (default)'map' represents values as a key/value map.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -805,6 +835,32 @@ public interface DebeziumPostgresEndpointBuilderFactory {
return this;
}
/**
+ * Enables transaction metadata extraction together with event counting.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: postgres
+ */
+ default DebeziumPostgresEndpointBuilder provideTransactionMetadata(
+ boolean provideTransactionMetadata) {
+ doSetProperty("provideTransactionMetadata", provideTransactionMetadata);
+ return this;
+ }
+ /**
+ * Enables transaction metadata extraction together with event counting.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: postgres
+ */
+ default DebeziumPostgresEndpointBuilder provideTransactionMetadata(
+ String provideTransactionMetadata) {
+ doSetProperty("provideTransactionMetadata", provideTransactionMetadata);
+ return this;
+ }
+ /**
* The name of the Postgres 10 publication used for streaming changes
* from a plugin.Defaults to 'dbz_publication'.
*
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
index 55aac48..c2227f4 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
@@ -572,6 +572,24 @@ public interface DebeziumSqlserverEndpointBuilderFactory {
return this;
}
/**
+ * Specify how failures during processing of events (i.e. when
+ * encountering a corrupted event) should be handled, including:'fail'
+ * (the default) an exception indicating the problematic event and its
+ * position is raised, causing the connector to be stopped; 'warn' the
+ * problematic event and its position will be logged and the event will
+ * be skipped;'ignore' the problematic event will be skipped.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: fail
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverEndpointBuilder eventProcessingFailureHandlingMode(
+ String eventProcessingFailureHandlingMode) {
+ doSetProperty("eventProcessingFailureHandlingMode", eventProcessingFailureHandlingMode);
+ return this;
+ }
+ /**
* Length of an interval in milli-seconds in in which the connector
* periodically sends heartbeat messages to a heartbeat topic. Use 0 to
* disable heartbeat messages. Disabled by default.
@@ -718,6 +736,32 @@ public interface DebeziumSqlserverEndpointBuilderFactory {
return this;
}
/**
+ * Enables transaction metadata extraction together with event counting.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverEndpointBuilder provideTransactionMetadata(
+ boolean provideTransactionMetadata) {
+ doSetProperty("provideTransactionMetadata", provideTransactionMetadata);
+ return this;
+ }
+ /**
+ * Enables transaction metadata extraction together with event counting.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverEndpointBuilder provideTransactionMetadata(
+ String provideTransactionMetadata) {
+ doSetProperty("provideTransactionMetadata", provideTransactionMetadata);
+ return this;
+ }
+ /**
* The number of milliseconds to delay before a snapshot will begin.
*
* The option is a: <code>long</code> type.
diff --git a/parent/pom.xml b/parent/pom.xml
index de0295b..872b410 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -176,7 +176,7 @@
<dozer-version>6.5.0</dozer-version>
<drools-version>7.34.0.Final</drools-version>
<dropbox-version>3.1.3</dropbox-version>
- <debezium-version>1.0.0.Final</debezium-version>
+ <debezium-version>1.1.0.Final</debezium-version>
<debezium-mysql-connector-version>8.0.16</debezium-mysql-connector-version>
<egit-github-core-version>2.1.5</egit-github-core-version>
<egit-github-core-bundle-version>2.1.5_1</egit-github-core-bundle-version>