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 faa3182 CAMEL-14188: Upgrade Debezium to 1.0
new d60dfcb Merge pull request #3487 from omarsmak/CAMEL-14188
faa3182 is described below
commit faa31827ad8e347f49649571f3cb4ce154a85cc5
Author: Omar Al-Safi <omars...@gmail.com>
AuthorDate: Thu Jan 16 12:50:20 2020 +0100
CAMEL-14188: Upgrade Debezium to 1.0
---
.../component/debezium/DebeziumEndpointTest.java | 7 +--
.../src/main/docs/debezium-postgres-component.adoc | 3 +-
components/camel-debezium-sqlserver/pom.xml | 4 +-
.../main/docs/debezium-sqlserver-component.adoc | 6 ++-
.../DebeziumPostgresEndpointBuilderFactory.java | 16 +++++++
.../DebeziumSqlserverEndpointBuilderFactory.java | 54 ++++++++++++++++++++--
parent/pom.xml | 4 +-
7 files changed, 80 insertions(+), 14 deletions(-)
diff --git
a/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
b/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
index 19b5613..05d71a3 100644
---
a/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
+++
b/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.debezium;
+import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
@@ -205,7 +206,7 @@ public class DebeziumEndpointTest {
after.put("id", (byte)1);
source.put("lsn", 1234);
- final Struct payload = envelope.create(after, source,
System.nanoTime());
+ final Struct payload = envelope.create(after, source, Instant.now());
return new SourceRecord(new HashMap<>(), createSourceOffset(),
"dummy", createKeySchema(),
createKeyRecord(), envelope.schema(), payload);
}
@@ -216,7 +217,7 @@ public class DebeziumEndpointTest {
.withSource(SchemaBuilder.struct().build()).build();
final Struct before = new Struct(recordSchema);
before.put("id", (byte)1);
- final Struct payload = envelope.delete(before, null,
System.nanoTime());
+ final Struct payload = envelope.delete(before, null, Instant.now());
return new SourceRecord(new HashMap<>(), createSourceOffset(),
"dummy", createKeySchema(),
createKeyRecord(), envelope.schema(), payload);
}
@@ -243,7 +244,7 @@ public class DebeziumEndpointTest {
before.put("id", (byte)1);
after.put("id", (byte)2);
source.put("lsn", 1234);
- final Struct payload = envelope.update(before, after, source,
System.nanoTime());
+ final Struct payload = envelope.update(before, after, source,
Instant.now());
return new SourceRecord(new HashMap<>(), createSourceOffset(),
"dummy", createKeySchema(),
createKeyRecord(), envelope.schema(), payload);
}
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 6a0c357..1f84788 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
@@ -73,7 +73,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (64 parameters):
+=== Query Parameters (65 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -115,6 +115,7 @@ with the following path and query parameters:
| *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
| *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
| *maxQueueSize* (postgres) | 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* (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 [...]
diff --git a/components/camel-debezium-sqlserver/pom.xml
b/components/camel-debezium-sqlserver/pom.xml
index 89a1700..8354b97 100644
--- a/components/camel-debezium-sqlserver/pom.xml
+++ b/components/camel-debezium-sqlserver/pom.xml
@@ -91,9 +91,7 @@
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-sqlserver</artifactId>
- <!-- Only for the maven plugin, we will use
1.0.0.Beta2 to generate the config due to missing configs on 0.10 -->
- <!-- Once we have 1.0.0.Final, we will change this
back to ${debezium-version} -->
- <version>1.0.0.Beta2</version>
+ <version>${debezium-version}</version>
<scope>runtime</scope>
</dependency>
</dependencies>
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 5411351..2188cd6 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
@@ -72,7 +72,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (45 parameters):
+=== Query Parameters (47 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -105,6 +105,7 @@ with the following path and query parameters:
| *databasePassword* (sqlserver) | *Required* Password of the SQL Server
database user to be used when connecting to the database. | | String
| *databasePort* (sqlserver) | Port of the SQL Server database server. | 1433
| int
| *databaseServerName* (sqlserver) | *Required* Unique name that identifies
the database server and all recorded offsets, and that is used as a prefix for
all schemas and topics. Each distinct installation should have a separate
namespace and be monitored by at most one Debezium connector. | | String
+| *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 [...]
| *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
@@ -116,13 +117,14 @@ with the following path and query parameters:
| *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
-| *snapshotMode* (sqlserver) | The criteria for running a snapshot upon
startup of the connector. Options include: 'initial' (the default) to specify
the connector should run a snapshot only when no offsets are available for the
logical server name; 'initial_schema_only' to specify the connector should run
a snapshot of the schema when no offsets are available for the logical server
name. | initial | String
+| *snapshotMode* (sqlserver) | The criteria for running a snapshot upon
startup of the connector. Options include: 'initial' (the default) to specify
the connector should run a snapshot only when no offsets are available for the
logical server name; 'schema_only' to specify the connector should run a
snapshot of the schema when no offsets are available for the logical server
name. | initial | String
| *snapshotSelectStatement Overrides* (sqlserver) | This property contains a
comma-separated list of fully-qualified tables (DB_NAME.TABLE_NAME) or
(SCHEMA_NAME.TABLE_NAME), depending on thespecific connectors . Select
statements for the individual tables are specified in further configuration
properties, one for each table, identified by the id
'snapshot.select.statement.overrides.DB_NAME.TABLE_NAME' or
'snapshot.select.statement.overrides.SCHEMA_NAME.TABLE_NAME', respectively. The
valu [...]
| *sourceStructVersion* (sqlserver) | A version of the format of the publicly
visible source part in the message | v2 | String
| *tableBlacklist* (sqlserver) | Description is not available here, please
check Debezium website for corresponding key 'table.blacklist' description. |
| String
| *tableIgnoreBuiltin* (sqlserver) | Flag specifying whether built-in tables
should be ignored. | true | boolean
| *tableWhitelist* (sqlserver) | The tables for which changes are to be
captured | | String
| *timePrecisionMode* (sqlserver) | Time, date, and timestamps can be
represented with different kinds of precisions, including:'adaptive' (the
default) bases the precision of time, date, and timestamp values on the
database column's precision; 'adaptive_time_microseconds' like 'adaptive' mode,
but TIME fields always use microseconds precision;'connect' always represents
time, date, and timestamp values using Kafka Connect's built-in representations
for Time, Date, and Timestamp, which u [...]
+| *tombstonesOnDelete* (sqlserver) | Whether delete operations should be
represented by a delete event and a subsquenttombstone event (true) or only by
a delete event (false). Emitting the tombstone event (the default behavior)
allows Kafka to completely delete all events pertaining to the given key once
the source record got deleted. | false | boolean
|===
// endpoint options: END
// spring-boot-auto-configure options: START
diff --git
a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
index 936daf2..fa277ce 100644
---
a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
+++
b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumPostgresEndpointBuilderFactory.java
@@ -637,6 +637,22 @@ public interface DebeziumPostgresEndpointBuilderFactory {
return this;
}
/**
+ * 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.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: numeric
+ * Group: postgres
+ */
+ default DebeziumPostgresEndpointBuilder intervalHandlingMode(
+ String intervalHandlingMode) {
+ doSetProperty("intervalHandlingMode", intervalHandlingMode);
+ return this;
+ }
+ /**
* Maximum size of each batch of source records. Defaults to 2048.
*
* The option is a: <code>int</code> type.
diff --git
a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
index 07172b3..501e549 100644
---
a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
+++
b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/DebeziumSqlserverEndpointBuilderFactory.java
@@ -487,6 +487,20 @@ public interface DebeziumSqlserverEndpointBuilderFactory {
return this;
}
/**
+ * The timezone of the server used to correctly shift the commit
+ * transaction timestamp on the client sideOptions include: Any valid
+ * Java ZoneId.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverEndpointBuilder databaseServerTimezone(
+ String databaseServerTimezone) {
+ doSetProperty("databaseServerTimezone", databaseServerTimezone);
+ return this;
+ }
+ /**
* Name of the SQL Server database user to be used when connecting to
* the database.
*
@@ -751,9 +765,9 @@ public interface DebeziumSqlserverEndpointBuilderFactory {
* The criteria for running a snapshot upon startup of the connector.
* Options include: 'initial' (the default) to specify the connector
* should run a snapshot only when no offsets are available for the
- * logical server name; 'initial_schema_only' to specify the connector
- * should run a snapshot of the schema when no offsets are available
for
- * the logical server name.
+ * logical server name; 'schema_only' to specify the connector should
+ * run a snapshot of the schema when no offsets are available for the
+ * logical server name.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -873,6 +887,40 @@ public interface DebeziumSqlserverEndpointBuilderFactory {
doSetProperty("timePrecisionMode", timePrecisionMode);
return this;
}
+ /**
+ * Whether delete operations should be represented by a delete event
and
+ * a subsquenttombstone event (true) or only by a delete event (false).
+ * Emitting the tombstone event (the default behavior) allows Kafka to
+ * completely delete all events pertaining to the given key once the
+ * source record got deleted.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverEndpointBuilder tombstonesOnDelete(
+ boolean tombstonesOnDelete) {
+ doSetProperty("tombstonesOnDelete", tombstonesOnDelete);
+ return this;
+ }
+ /**
+ * Whether delete operations should be represented by a delete event
and
+ * a subsquenttombstone event (true) or only by a delete event (false).
+ * Emitting the tombstone event (the default behavior) allows Kafka to
+ * completely delete all events pertaining to the given key once the
+ * source record got deleted.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: sqlserver
+ */
+ default DebeziumSqlserverEndpointBuilder tombstonesOnDelete(
+ String tombstonesOnDelete) {
+ doSetProperty("tombstonesOnDelete", tombstonesOnDelete);
+ return this;
+ }
}
/**
diff --git a/parent/pom.xml b/parent/pom.xml
index e24b8fc..0492020 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -172,7 +172,7 @@
<dozer-version>6.5.0</dozer-version>
<drools-version>7.31.0.Final</drools-version>
<dropbox-version>3.1.3</dropbox-version>
- <debezium-version>0.10.0.Final</debezium-version>
+ <debezium-version>1.0.0.Final</debezium-version>
<debezium-mysql-connector-version>8.0.15</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>
@@ -3770,7 +3770,7 @@
<artifactId>pulsar-client-admin</artifactId>
<version>${pulsar-version}</version>
</dependency>
-
+
<!-- apicurio -->
<dependency>
<groupId>io.apicurio</groupId>
