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/01/29 14:07:40 UTC
[camel] branch master updated: CAMEL-14420: Support additional
properties on Debezium components
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 2ee5d19 CAMEL-14420: Support additional properties on Debezium components
new 06cfe88 Merge pull request #3530 from omarsmak/CAMEL-14420
2ee5d19 is described below
commit 2ee5d192018ce24acf56c156802d0fd8bc697ee7
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Wed Jan 29 14:48:27 2020 +0100
CAMEL-14420: Support additional properties on Debezium components
---
.../component/debezium/DebeziumComponent.java | 9 ++++
.../EmbeddedDebeziumConfiguration.java | 28 ++++++++++++
.../component/debezium/DebeziumComponentTest.java | 53 ++++++++++++++++++++++
.../EmbeddedDebeziumConfigurationTest.java | 15 ++++++
.../src/main/docs/debezium-mongodb-component.adoc | 3 +-
.../src/main/docs/debezium-mysql-component.adoc | 3 +-
.../src/main/docs/debezium-postgres-component.adoc | 3 +-
.../main/docs/debezium-sqlserver-component.adoc | 3 +-
8 files changed, 113 insertions(+), 4 deletions(-)
diff --git a/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumComponent.java b/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumComponent.java
index 3b48d0a..18919df 100644
--- a/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumComponent.java
+++ b/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumComponent.java
@@ -23,6 +23,7 @@ import org.apache.camel.component.debezium.configuration.ConfigurationValidation
import org.apache.camel.component.debezium.configuration.EmbeddedDebeziumConfiguration;
import org.apache.camel.support.DefaultComponent;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.PropertiesHelper;
/**
* Base class for all debezium components
@@ -53,6 +54,14 @@ public abstract class DebeziumComponent<C extends EmbeddedDebeziumConfiguration>
DebeziumEndpoint endpoint = initializeDebeziumEndpoint(uri, configuration);
setProperties(endpoint, parameters);
+ // extract the additional properties map
+ if (PropertiesHelper.hasProperties(parameters, "additionalProperties.")) {
+ final Map<String, Object> additionalProperties = endpoint.getConfiguration().getAdditionalProperties();
+
+ // add and overwrite additional properties from endpoint to pre-configured properties
+ additionalProperties.putAll(PropertiesHelper.extractProperties(parameters, "additionalProperties."));
+ }
+
// validate configurations
final ConfigurationValidation configurationValidation = configuration.validateConfiguration();
diff --git a/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfiguration.java b/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfiguration.java
index 9ea17c5..18600d4 100644
--- a/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfiguration.java
+++ b/components/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfiguration.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.debezium.configuration;
+import java.util.HashMap;
+import java.util.Map;
+
import io.debezium.config.Configuration;
import io.debezium.config.Field;
import io.debezium.embedded.EmbeddedEngine;
@@ -68,6 +71,9 @@ public abstract class EmbeddedDebeziumConfiguration {
// internal.value.converter
@UriParam(label = LABEL_NAME, defaultValue = "org.apache.kafka.connect.json.JsonConverter")
private String internalValueConverter = JsonConverter.class.getName();
+ // Additional properties
+ @UriParam(label = "common", prefix = "additionalProperties.", multiValue = true)
+ private Map<String, Object> additionalProperties = new HashMap<>();
public EmbeddedDebeziumConfiguration() {
ObjectHelper.notNull(configureConnectorClass(), "connectorClass");
@@ -139,6 +145,9 @@ public abstract class EmbeddedDebeziumConfiguration {
configBuilder.with("internal.value.converter", internalValueConverter);
}
+ // additional properties
+ applyAdditionalProperties(configBuilder, getAdditionalProperties());
+
return configBuilder.build();
}
@@ -156,6 +165,12 @@ public abstract class EmbeddedDebeziumConfiguration {
}
}
+ private void applyAdditionalProperties(final Configuration.Builder configBuilder, final Map<String, Object> additionalProperties) {
+ if (!ObjectHelper.isEmpty(getAdditionalProperties())) {
+ additionalProperties.forEach((property, value) -> addPropertyIfNotNull(configBuilder, property, value));
+ }
+ }
+
/**
* Validate all configurations defined and return
* {@link ConfigurationValidation} instance which contains the validation
@@ -340,4 +355,17 @@ public abstract class EmbeddedDebeziumConfiguration {
public void setInternalValueConverter(String internalValueConverter) {
this.internalValueConverter = internalValueConverter;
}
+
+ /**
+ * Sets additional properties for debezium components in case they can't be set directly on the camel configurations
+ * (e.g: setting Kafka Connect properties needed by Debezium engine, for example setting KafkaOffsetBackingStore), the properties have to be prefixed with
+ * `additionalProperties.`. E.g: `additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro`
+ */
+ public void setAdditionalProperties(Map<String, Object> additionalProperties) {
+ this.additionalProperties = additionalProperties;
+ }
+
+ public Map<String, Object> getAdditionalProperties() {
+ return additionalProperties;
+ }
}
diff --git a/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumComponentTest.java b/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumComponentTest.java
new file mode 100644
index 0000000..ae0b35d
--- /dev/null
+++ b/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumComponentTest.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.debezium;
+
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.component.debezium.configuration.FileConnectorEmbeddedDebeziumConfiguration;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class DebeziumComponentTest extends CamelTestSupport {
+
+ @Test
+ public void testIfSetsAdditionalProperties() throws Exception {
+ final DebeziumTestComponent component = new DebeziumTestComponent(context);
+ final FileConnectorEmbeddedDebeziumConfiguration configuration = new FileConnectorEmbeddedDebeziumConfiguration();
+
+ final Map<String, Object> params = new HashMap<>();
+ params.put("extra.1", 789);
+ params.put("extra.3", "test.extra.3");
+
+ configuration.setOffsetStorageFileName("/test");
+ configuration.setTestFilePath(Paths.get("."));
+ configuration.setTopicConfig("test_conf");
+ configuration.setAdditionalProperties(params);
+
+ component.setConfiguration(configuration);
+
+ final DebeziumTestEndpoint endpoint = (DebeziumTestEndpoint) component.createEndpoint("debezium-dummy:test_name?additionalProperties.extra.1=123&additionalProperties.extra.2=test");
+
+ assertEquals("test_name", endpoint.getConfiguration().getName());
+ assertEquals("test_conf", endpoint.getConfiguration().getTopicConfig());
+ assertEquals("123", endpoint.getConfiguration().getAdditionalProperties().get("extra.1"));
+ assertEquals("test", endpoint.getConfiguration().getAdditionalProperties().get("extra.2"));
+ assertEquals("test.extra.3", endpoint.getConfiguration().getAdditionalProperties().get("extra.3"));
+ }
+}
diff --git a/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfigurationTest.java b/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfigurationTest.java
index 30d061d..6406504 100644
--- a/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfigurationTest.java
+++ b/components/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/configuration/EmbeddedDebeziumConfigurationTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.debezium.configuration;
+import java.util.Collections;
+
import io.debezium.config.Configuration;
import io.debezium.embedded.EmbeddedEngine;
import org.apache.camel.component.debezium.DebeziumConstants;
@@ -60,4 +62,17 @@ public class EmbeddedDebeziumConfigurationTest {
assertTrue(configuration.validateConfiguration().isValid());
}
+ @Test
+ public void testIfCreatesAdditionalProperties() {
+ final TestEmbeddedDebeziumConfiguration configuration = new TestEmbeddedDebeziumConfiguration();
+ configuration.setName("test_config");
+ configuration.setAdditionalProperties(Collections.singletonMap("test.additional", "test_additional"));
+ configuration.setTestField("test_field");
+
+ final Configuration dbzEmbeddedConfiguration = configuration.createDebeziumConfiguration();
+
+ assertEquals("test_config", dbzEmbeddedConfiguration.getString("name"));
+ assertEquals("test_additional", dbzEmbeddedConfiguration.getString("test.additional"));
+ }
+
}
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 e22e36e..2825a4d 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
@@ -75,12 +75,13 @@ with the following path and query parameters:
|===
-=== Query Parameters (43 parameters):
+=== Query Parameters (44 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *additionalProperties* (common) | Sets additional properties for debezium components in case they can't be set directly on the camel configurations (e.g: setting Kafka Connect properties needed by Debezium engine, for example setting KafkaOffsetBackingStore), the properties have to be prefixed with additionalProperties.. E.g: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=\http://localhost:8811/avro | | Map
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *internalKeyConverter* (consumer) | The Converter class that should be used to serialize and deserialize key data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
| *internalValueConverter* (consumer) | The Converter class that should be used to serialize and deserialize value data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
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 43de3a7..912887e 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
@@ -82,12 +82,13 @@ with the following path and query parameters:
|===
-=== Query Parameters (73 parameters):
+=== Query Parameters (74 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *additionalProperties* (common) | Sets additional properties for debezium components in case they can't be set directly on the camel configurations (e.g: setting Kafka Connect properties needed by Debezium engine, for example setting KafkaOffsetBackingStore), the properties have to be prefixed with additionalProperties.. E.g: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=\http://localhost:8811/avro | | Map
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *internalKeyConverter* (consumer) | The Converter class that should be used to serialize and deserialize key data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
| *internalValueConverter* (consumer) | The Converter class that should be used to serialize and deserialize value data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
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 1f84788..a868321 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,12 +73,13 @@ with the following path and query parameters:
|===
-=== Query Parameters (65 parameters):
+=== Query Parameters (66 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *additionalProperties* (common) | Sets additional properties for debezium components in case they can't be set directly on the camel configurations (e.g: setting Kafka Connect properties needed by Debezium engine, for example setting KafkaOffsetBackingStore), the properties have to be prefixed with additionalProperties.. E.g: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=\http://localhost:8811/avro | | Map
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *internalKeyConverter* (consumer) | The Converter class that should be used to serialize and deserialize key data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
| *internalValueConverter* (consumer) | The Converter class that should be used to serialize and deserialize value data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
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 2188cd6..80087a4 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,12 +72,13 @@ with the following path and query parameters:
|===
-=== Query Parameters (47 parameters):
+=== Query Parameters (48 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *additionalProperties* (common) | Sets additional properties for debezium components in case they can't be set directly on the camel configurations (e.g: setting Kafka Connect properties needed by Debezium engine, for example setting KafkaOffsetBackingStore), the properties have to be prefixed with additionalProperties.. E.g: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=\http://localhost:8811/avro | | Map
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *internalKeyConverter* (consumer) | The Converter class that should be used to serialize and deserialize key data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String
| *internalValueConverter* (consumer) | The Converter class that should be used to serialize and deserialize value data for offsets. The default is JSON converter. | org.apache.kafka.connect.json.JsonConverter | String