You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2022/10/04 08:05:52 UTC
[camel] branch main updated: CAMEL-10173: camel-etcd3 - Add an implementation based on jetcd (#8452)
This is an automated email from the ASF dual-hosted git repository.
nfilotto pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new ddcb425a2b3 CAMEL-10173: camel-etcd3 - Add an implementation based on jetcd (#8452)
ddcb425a2b3 is described below
commit ddcb425a2b3a631b4e3c1641fe1898ac939db45e
Author: Nicolas Filotto <es...@users.noreply.github.com>
AuthorDate: Tue Oct 4 10:05:46 2022 +0200
CAMEL-10173: camel-etcd3 - Add an implementation based on jetcd (#8452)
## Motivation
The API v3 of etcd should be supported by Camel and for this, we would like to propose a new component based on https://github.com/etcd-io/jetcd.
## Modifications:
* Add a new test infra for Integration tests with etcd v3
* Convert the test for the AggregationRepository into an integration test
* Propose a consumer and a producer based on jetcd and the legacy code.
* Propose a ServiceDiscovery based on jetcd and the legacy code.
* Propose a RoutePolicy based on jetcd and the legacy code. The leadership management relies on a lease with TTL.
---
camel-dependencies/pom.xml | 1 +
components/camel-etcd3/pom.xml | 50 ++-
.../component/etcd3/Etcd3ComponentConfigurer.java | 189 +++++++++
.../component/etcd3/Etcd3EndpointConfigurer.java | 185 +++++++++
.../component/etcd3/Etcd3EndpointUriFactory.java | 99 +++++
.../Etcd3ServiceDiscoveryFactoryConfigurer.java | 174 ++++++++
.../org/apache/camel/cloud/etcd-service-discovery | 2 +
.../{other.properties => component.properties} | 4 +-
.../services/org/apache/camel/component/etcd3 | 2 +
.../org/apache/camel/configurer/etcd3-component | 2 +
.../org/apache/camel/configurer/etcd3-endpoint | 2 +
...ponent.etcd3.cloud.Etcd3ServiceDiscoveryFactory | 2 +
.../org/apache/camel/urifactory/etcd3-endpoint | 2 +
.../org/apache/camel/component/etcd3/etcd3.json | 86 ++++
components/camel-etcd3/src/main/docs/etcd3.adoc | 92 +++-
.../camel/component/etcd3/Etcd3Component.java | 61 +++
.../camel/component/etcd3/Etcd3Configuration.java | 360 ++++++++++++++++
.../camel/component/etcd3/Etcd3Constants.java | 50 +++
.../camel/component/etcd3/Etcd3Consumer.java | 150 +++++++
.../camel/component/etcd3/Etcd3Endpoint.java | 60 +++
.../apache/camel/component/etcd3/Etcd3Helper.java | 52 +++
.../camel/component/etcd3/Etcd3Producer.java | 229 ++++++++++
.../etcd3/cloud/Etcd3GetServicesResponse.java | 61 +++
.../etcd3/cloud/Etcd3OnDemandServiceDiscovery.java | 36 ++
.../etcd3/cloud/Etcd3ServiceDefinition.java | 39 ++
.../etcd3/cloud/Etcd3ServiceDiscovery.java | 159 +++++++
.../etcd3/cloud/Etcd3ServiceDiscoveryFactory.java | 285 +++++++++++++
.../etcd3/cloud/Etcd3WatchServiceDiscovery.java | 169 ++++++++
.../component/etcd3/policy/Etcd3RoutePolicy.java | 461 +++++++++++++++++++++
.../aggregate/Etcd3AggregationRepository.java | 2 +-
.../component/etcd3/AggregateEtcd3ManualTest.java | 14 +-
.../cloud/integration/Etcd3ServiceDiscoveryIT.java | 142 +++++++
.../etcd3/integration/Etcd3ConsumerIT.java | 149 +++++++
.../etcd3/integration/Etcd3ProducerIT.java | 344 +++++++++++++++
.../component/etcd3/policy/Etcd3RoutePolicyIT.java | 101 +++++
.../etcd3/policy/Etcd3RoutePolicyMain.java | 49 +++
.../component/etcd3/support/Etcd3TestSupport.java | 56 +++
parent/pom.xml | 3 +-
test-infra/camel-test-infra-etcd3/pom.xml | 64 +++
.../src/main/resources/META-INF/MANIFEST.MF | 0
.../test/infra/etcd3/common/Etcd3Properties.java | 26 ++
.../etcd3/services/Etcd3LocalContainerService.java | 86 ++++
.../infra/etcd3/services/Etcd3RemoteService.java | 42 ++
.../test/infra/etcd3/services/Etcd3Service.java | 40 ++
.../infra/etcd3/services/Etcd3ServiceFactory.java | 37 ++
test-infra/pom.xml | 1 +
46 files changed, 4200 insertions(+), 20 deletions(-)
diff --git a/camel-dependencies/pom.xml b/camel-dependencies/pom.xml
index f86e2e7405c..46426b597d4 100644
--- a/camel-dependencies/pom.xml
+++ b/camel-dependencies/pom.xml
@@ -314,6 +314,7 @@
<jcr-version>2.0</jcr-version>
<jedis-client-version>3.7.1</jedis-client-version>
<jetcd-grpc-version>1.47.0</jetcd-grpc-version>
+ <jetcd-guava-version>31.1-jre</jetcd-guava-version>
<jetcd-version>0.7.3</jetcd-version>
<jettison-version>1.5.1</jettison-version>
<jetty-plugin-version>${jetty-version}</jetty-plugin-version>
diff --git a/components/camel-etcd3/pom.xml b/components/camel-etcd3/pom.xml
index 7b8763d21d9..3dc2c5e0426 100644
--- a/components/camel-etcd3/pom.xml
+++ b/components/camel-etcd3/pom.xml
@@ -30,7 +30,7 @@
<artifactId>camel-etcd3</artifactId>
<packaging>jar</packaging>
<name>Camel :: Etcd3</name>
- <description>Aggregation repository using EtcD as datastore</description>
+ <description>Camel EtcD v3 component based on jetcd</description>
<properties>
<firstVersion>3.5.0</firstVersion>
@@ -40,11 +40,26 @@
<dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-cloud</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-support</artifactId>
</dependency>
-
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
<dependency>
<groupId>io.etcd</groupId>
<artifactId>jetcd-core</artifactId>
@@ -62,6 +77,18 @@
<artifactId>grpc-core</artifactId>
<version>${jetcd-grpc-version}</version>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${jetcd-guava-version}</version>
+ </dependency>
+
+ <!-- logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- testing -->
<dependency>
@@ -70,11 +97,24 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-slf4j-impl</artifactId>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-main</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- test infra -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-etcd3</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
<scope>test</scope>
</dependency>
-
</dependencies>
<build>
diff --git a/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3ComponentConfigurer.java b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3ComponentConfigurer.java
new file mode 100644
index 00000000000..fda0696cc57
--- /dev/null
+++ b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3ComponentConfigurer.java
@@ -0,0 +1,189 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.etcd3;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class Etcd3ComponentConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ private org.apache.camel.component.etcd3.Etcd3Configuration getOrCreateConfiguration(Etcd3Component target) {
+ if (target.getConfiguration() == null) {
+ target.setConfiguration(new org.apache.camel.component.etcd3.Etcd3Configuration());
+ }
+ return target.getConfiguration();
+ }
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ Etcd3Component target = (Etcd3Component) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": getOrCreateConfiguration(target).setAuthHeaders(property(camelContext, java.util.Map.class, value)); return true;
+ case "authority": getOrCreateConfiguration(target).setAuthority(property(camelContext, java.lang.String.class, value)); return true;
+ case "autowiredenabled":
+ case "autowiredEnabled": target.setAutowiredEnabled(property(camelContext, boolean.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.etcd3.Etcd3Configuration.class, value)); return true;
+ case "connectiontimeout":
+ case "connectionTimeout": getOrCreateConfiguration(target).setConnectionTimeout(property(camelContext, java.time.Duration.class, value)); return true;
+ case "endpoints": getOrCreateConfiguration(target).setEndpoints(property(camelContext, java.lang.String[].class, value)); return true;
+ case "fromindex":
+ case "fromIndex": getOrCreateConfiguration(target).setFromIndex(property(camelContext, long.class, value)); return true;
+ case "headers": getOrCreateConfiguration(target).setHeaders(property(camelContext, java.util.Map.class, value)); return true;
+ case "keepalivetime":
+ case "keepAliveTime": getOrCreateConfiguration(target).setKeepAliveTime(property(camelContext, java.time.Duration.class, value)); return true;
+ case "keepalivetimeout":
+ case "keepAliveTimeout": getOrCreateConfiguration(target).setKeepAliveTimeout(property(camelContext, java.time.Duration.class, value)); return true;
+ case "keycharset":
+ case "keyCharset": getOrCreateConfiguration(target).setKeyCharset(property(camelContext, java.lang.String.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "loadbalancerpolicy":
+ case "loadBalancerPolicy": getOrCreateConfiguration(target).setLoadBalancerPolicy(property(camelContext, java.lang.String.class, value)); return true;
+ case "maxinboundmessagesize":
+ case "maxInboundMessageSize": getOrCreateConfiguration(target).setMaxInboundMessageSize(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "namespace": getOrCreateConfiguration(target).setNamespace(property(camelContext, java.lang.String.class, value)); return true;
+ case "password": getOrCreateConfiguration(target).setPassword(property(camelContext, java.lang.String.class, value)); return true;
+ case "prefix": getOrCreateConfiguration(target).setPrefix(property(camelContext, boolean.class, value)); return true;
+ case "retrydelay":
+ case "retryDelay": getOrCreateConfiguration(target).setRetryDelay(property(camelContext, long.class, value)); return true;
+ case "retrymaxdelay":
+ case "retryMaxDelay": getOrCreateConfiguration(target).setRetryMaxDelay(property(camelContext, long.class, value)); return true;
+ case "retrymaxduration":
+ case "retryMaxDuration": getOrCreateConfiguration(target).setRetryMaxDuration(property(camelContext, java.time.Duration.class, value)); return true;
+ case "servicepath":
+ case "servicePath": getOrCreateConfiguration(target).setServicePath(property(camelContext, java.lang.String.class, value)); return true;
+ case "sslcontext":
+ case "sslContext": getOrCreateConfiguration(target).setSslContext(property(camelContext, io.netty.handler.ssl.SslContext.class, value)); return true;
+ case "username":
+ case "userName": getOrCreateConfiguration(target).setUserName(property(camelContext, java.lang.String.class, value)); return true;
+ case "valuecharset":
+ case "valueCharset": getOrCreateConfiguration(target).setValueCharset(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": return java.util.Map.class;
+ case "authority": return java.lang.String.class;
+ case "autowiredenabled":
+ case "autowiredEnabled": return boolean.class;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return boolean.class;
+ case "configuration": return org.apache.camel.component.etcd3.Etcd3Configuration.class;
+ case "connectiontimeout":
+ case "connectionTimeout": return java.time.Duration.class;
+ case "endpoints": return java.lang.String[].class;
+ case "fromindex":
+ case "fromIndex": return long.class;
+ case "headers": return java.util.Map.class;
+ case "keepalivetime":
+ case "keepAliveTime": return java.time.Duration.class;
+ case "keepalivetimeout":
+ case "keepAliveTimeout": return java.time.Duration.class;
+ case "keycharset":
+ case "keyCharset": return java.lang.String.class;
+ case "lazystartproducer":
+ case "lazyStartProducer": return boolean.class;
+ case "loadbalancerpolicy":
+ case "loadBalancerPolicy": return java.lang.String.class;
+ case "maxinboundmessagesize":
+ case "maxInboundMessageSize": return java.lang.Integer.class;
+ case "namespace": return java.lang.String.class;
+ case "password": return java.lang.String.class;
+ case "prefix": return boolean.class;
+ case "retrydelay":
+ case "retryDelay": return long.class;
+ case "retrymaxdelay":
+ case "retryMaxDelay": return long.class;
+ case "retrymaxduration":
+ case "retryMaxDuration": return java.time.Duration.class;
+ case "servicepath":
+ case "servicePath": return java.lang.String.class;
+ case "sslcontext":
+ case "sslContext": return io.netty.handler.ssl.SslContext.class;
+ case "username":
+ case "userName": return java.lang.String.class;
+ case "valuecharset":
+ case "valueCharset": return java.lang.String.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ Etcd3Component target = (Etcd3Component) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": return getOrCreateConfiguration(target).getAuthHeaders();
+ case "authority": return getOrCreateConfiguration(target).getAuthority();
+ case "autowiredenabled":
+ case "autowiredEnabled": return target.isAutowiredEnabled();
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+ case "configuration": return target.getConfiguration();
+ case "connectiontimeout":
+ case "connectionTimeout": return getOrCreateConfiguration(target).getConnectionTimeout();
+ case "endpoints": return getOrCreateConfiguration(target).getEndpoints();
+ case "fromindex":
+ case "fromIndex": return getOrCreateConfiguration(target).getFromIndex();
+ case "headers": return getOrCreateConfiguration(target).getHeaders();
+ case "keepalivetime":
+ case "keepAliveTime": return getOrCreateConfiguration(target).getKeepAliveTime();
+ case "keepalivetimeout":
+ case "keepAliveTimeout": return getOrCreateConfiguration(target).getKeepAliveTimeout();
+ case "keycharset":
+ case "keyCharset": return getOrCreateConfiguration(target).getKeyCharset();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "loadbalancerpolicy":
+ case "loadBalancerPolicy": return getOrCreateConfiguration(target).getLoadBalancerPolicy();
+ case "maxinboundmessagesize":
+ case "maxInboundMessageSize": return getOrCreateConfiguration(target).getMaxInboundMessageSize();
+ case "namespace": return getOrCreateConfiguration(target).getNamespace();
+ case "password": return getOrCreateConfiguration(target).getPassword();
+ case "prefix": return getOrCreateConfiguration(target).isPrefix();
+ case "retrydelay":
+ case "retryDelay": return getOrCreateConfiguration(target).getRetryDelay();
+ case "retrymaxdelay":
+ case "retryMaxDelay": return getOrCreateConfiguration(target).getRetryMaxDelay();
+ case "retrymaxduration":
+ case "retryMaxDuration": return getOrCreateConfiguration(target).getRetryMaxDuration();
+ case "servicepath":
+ case "servicePath": return getOrCreateConfiguration(target).getServicePath();
+ case "sslcontext":
+ case "sslContext": return getOrCreateConfiguration(target).getSslContext();
+ case "username":
+ case "userName": return getOrCreateConfiguration(target).getUserName();
+ case "valuecharset":
+ case "valueCharset": return getOrCreateConfiguration(target).getValueCharset();
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": return java.lang.String.class;
+ case "headers": return java.lang.String.class;
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3EndpointConfigurer.java b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3EndpointConfigurer.java
new file mode 100644
index 00000000000..5b6ec200ac1
--- /dev/null
+++ b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3EndpointConfigurer.java
@@ -0,0 +1,185 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.etcd3;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class Etcd3EndpointConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ Etcd3Endpoint target = (Etcd3Endpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": target.getConfiguration().setAuthHeaders(property(camelContext, java.util.Map.class, value)); return true;
+ case "authority": target.getConfiguration().setAuthority(property(camelContext, java.lang.String.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "connectiontimeout":
+ case "connectionTimeout": target.getConfiguration().setConnectionTimeout(property(camelContext, java.time.Duration.class, value)); return true;
+ case "endpoints": target.getConfiguration().setEndpoints(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 "fromindex":
+ case "fromIndex": target.getConfiguration().setFromIndex(property(camelContext, long.class, value)); return true;
+ case "headers": target.getConfiguration().setHeaders(property(camelContext, java.util.Map.class, value)); return true;
+ case "keepalivetime":
+ case "keepAliveTime": target.getConfiguration().setKeepAliveTime(property(camelContext, java.time.Duration.class, value)); return true;
+ case "keepalivetimeout":
+ case "keepAliveTimeout": target.getConfiguration().setKeepAliveTimeout(property(camelContext, java.time.Duration.class, value)); return true;
+ case "keycharset":
+ case "keyCharset": target.getConfiguration().setKeyCharset(property(camelContext, java.lang.String.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "loadbalancerpolicy":
+ case "loadBalancerPolicy": target.getConfiguration().setLoadBalancerPolicy(property(camelContext, java.lang.String.class, value)); return true;
+ case "maxinboundmessagesize":
+ case "maxInboundMessageSize": target.getConfiguration().setMaxInboundMessageSize(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "namespace": target.getConfiguration().setNamespace(property(camelContext, java.lang.String.class, value)); return true;
+ case "password": target.getConfiguration().setPassword(property(camelContext, java.lang.String.class, value)); return true;
+ case "prefix": target.getConfiguration().setPrefix(property(camelContext, boolean.class, value)); return true;
+ case "retrydelay":
+ case "retryDelay": target.getConfiguration().setRetryDelay(property(camelContext, long.class, value)); return true;
+ case "retrymaxdelay":
+ case "retryMaxDelay": target.getConfiguration().setRetryMaxDelay(property(camelContext, long.class, value)); return true;
+ case "retrymaxduration":
+ case "retryMaxDuration": target.getConfiguration().setRetryMaxDuration(property(camelContext, java.time.Duration.class, value)); return true;
+ case "servicepath":
+ case "servicePath": target.getConfiguration().setServicePath(property(camelContext, java.lang.String.class, value)); return true;
+ case "sslcontext":
+ case "sslContext": target.getConfiguration().setSslContext(property(camelContext, io.netty.handler.ssl.SslContext.class, value)); return true;
+ case "username":
+ case "userName": target.getConfiguration().setUserName(property(camelContext, java.lang.String.class, value)); return true;
+ case "valuecharset":
+ case "valueCharset": target.getConfiguration().setValueCharset(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": return java.util.Map.class;
+ case "authority": return java.lang.String.class;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return boolean.class;
+ case "connectiontimeout":
+ case "connectionTimeout": return java.time.Duration.class;
+ case "endpoints": return java.lang.String[].class;
+ case "exceptionhandler":
+ case "exceptionHandler": return org.apache.camel.spi.ExceptionHandler.class;
+ case "exchangepattern":
+ case "exchangePattern": return org.apache.camel.ExchangePattern.class;
+ case "fromindex":
+ case "fromIndex": return long.class;
+ case "headers": return java.util.Map.class;
+ case "keepalivetime":
+ case "keepAliveTime": return java.time.Duration.class;
+ case "keepalivetimeout":
+ case "keepAliveTimeout": return java.time.Duration.class;
+ case "keycharset":
+ case "keyCharset": return java.lang.String.class;
+ case "lazystartproducer":
+ case "lazyStartProducer": return boolean.class;
+ case "loadbalancerpolicy":
+ case "loadBalancerPolicy": return java.lang.String.class;
+ case "maxinboundmessagesize":
+ case "maxInboundMessageSize": return java.lang.Integer.class;
+ case "namespace": return java.lang.String.class;
+ case "password": return java.lang.String.class;
+ case "prefix": return boolean.class;
+ case "retrydelay":
+ case "retryDelay": return long.class;
+ case "retrymaxdelay":
+ case "retryMaxDelay": return long.class;
+ case "retrymaxduration":
+ case "retryMaxDuration": return java.time.Duration.class;
+ case "servicepath":
+ case "servicePath": return java.lang.String.class;
+ case "sslcontext":
+ case "sslContext": return io.netty.handler.ssl.SslContext.class;
+ case "username":
+ case "userName": return java.lang.String.class;
+ case "valuecharset":
+ case "valueCharset": return java.lang.String.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ Etcd3Endpoint target = (Etcd3Endpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": return target.getConfiguration().getAuthHeaders();
+ case "authority": return target.getConfiguration().getAuthority();
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+ case "connectiontimeout":
+ case "connectionTimeout": return target.getConfiguration().getConnectionTimeout();
+ case "endpoints": return target.getConfiguration().getEndpoints();
+ case "exceptionhandler":
+ case "exceptionHandler": return target.getExceptionHandler();
+ case "exchangepattern":
+ case "exchangePattern": return target.getExchangePattern();
+ case "fromindex":
+ case "fromIndex": return target.getConfiguration().getFromIndex();
+ case "headers": return target.getConfiguration().getHeaders();
+ case "keepalivetime":
+ case "keepAliveTime": return target.getConfiguration().getKeepAliveTime();
+ case "keepalivetimeout":
+ case "keepAliveTimeout": return target.getConfiguration().getKeepAliveTimeout();
+ case "keycharset":
+ case "keyCharset": return target.getConfiguration().getKeyCharset();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "loadbalancerpolicy":
+ case "loadBalancerPolicy": return target.getConfiguration().getLoadBalancerPolicy();
+ case "maxinboundmessagesize":
+ case "maxInboundMessageSize": return target.getConfiguration().getMaxInboundMessageSize();
+ case "namespace": return target.getConfiguration().getNamespace();
+ case "password": return target.getConfiguration().getPassword();
+ case "prefix": return target.getConfiguration().isPrefix();
+ case "retrydelay":
+ case "retryDelay": return target.getConfiguration().getRetryDelay();
+ case "retrymaxdelay":
+ case "retryMaxDelay": return target.getConfiguration().getRetryMaxDelay();
+ case "retrymaxduration":
+ case "retryMaxDuration": return target.getConfiguration().getRetryMaxDuration();
+ case "servicepath":
+ case "servicePath": return target.getConfiguration().getServicePath();
+ case "sslcontext":
+ case "sslContext": return target.getConfiguration().getSslContext();
+ case "username":
+ case "userName": return target.getConfiguration().getUserName();
+ case "valuecharset":
+ case "valueCharset": return target.getConfiguration().getValueCharset();
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "authHeaders": return java.lang.String.class;
+ case "headers": return java.lang.String.class;
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3EndpointUriFactory.java b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3EndpointUriFactory.java
new file mode 100644
index 00000000000..6b5512eff07
--- /dev/null
+++ b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/Etcd3EndpointUriFactory.java
@@ -0,0 +1,99 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.etcd3;
+
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.spi.EndpointUriFactory;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+public class Etcd3EndpointUriFactory extends org.apache.camel.support.component.EndpointUriFactorySupport implements EndpointUriFactory {
+
+ private static final String BASE = ":path";
+
+ private static final Set<String> PROPERTY_NAMES;
+ private static final Set<String> SECRET_PROPERTY_NAMES;
+ private static final Set<String> MULTI_VALUE_PREFIXES;
+ static {
+ Set<String> props = new HashSet<>(26);
+ props.add("authHeaders");
+ props.add("authority");
+ props.add("bridgeErrorHandler");
+ props.add("connectionTimeout");
+ props.add("endpoints");
+ props.add("exceptionHandler");
+ props.add("exchangePattern");
+ props.add("fromIndex");
+ props.add("headers");
+ props.add("keepAliveTime");
+ props.add("keepAliveTimeout");
+ props.add("keyCharset");
+ props.add("lazyStartProducer");
+ props.add("loadBalancerPolicy");
+ props.add("maxInboundMessageSize");
+ props.add("namespace");
+ props.add("password");
+ props.add("path");
+ props.add("prefix");
+ props.add("retryDelay");
+ props.add("retryMaxDelay");
+ props.add("retryMaxDuration");
+ props.add("servicePath");
+ props.add("sslContext");
+ props.add("userName");
+ props.add("valueCharset");
+ PROPERTY_NAMES = Collections.unmodifiableSet(props);
+ Set<String> secretProps = new HashSet<>(2);
+ secretProps.add("password");
+ secretProps.add("userName");
+ SECRET_PROPERTY_NAMES = Collections.unmodifiableSet(secretProps);
+ Set<String> prefixes = new HashSet<>(2);
+ prefixes.add("authHeaders.");
+ prefixes.add("headers.");
+ MULTI_VALUE_PREFIXES = Collections.unmodifiableSet(prefixes);
+ }
+
+ @Override
+ public boolean isEnabled(String scheme) {
+ return "etcd3".equals(scheme);
+ }
+
+ @Override
+ public String buildUri(String scheme, Map<String, Object> properties, boolean encode) throws URISyntaxException {
+ String syntax = scheme + BASE;
+ String uri = syntax;
+
+ Map<String, Object> copy = new HashMap<>(properties);
+
+ uri = buildPathParameter(syntax, uri, "path", null, false, copy);
+ uri = buildQueryParameters(uri, copy, encode);
+ return uri;
+ }
+
+ @Override
+ public Set<String> propertyNames() {
+ return PROPERTY_NAMES;
+ }
+
+ @Override
+ public Set<String> secretPropertyNames() {
+ return SECRET_PROPERTY_NAMES;
+ }
+
+ @Override
+ public Set<String> multiValuePrefixes() {
+ return MULTI_VALUE_PREFIXES;
+ }
+
+ @Override
+ public boolean isLenientProperties() {
+ return false;
+ }
+}
+
diff --git a/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscoveryFactoryConfigurer.java b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscoveryFactoryConfigurer.java
new file mode 100644
index 00000000000..03ed7922e8d
--- /dev/null
+++ b/components/camel-etcd3/src/generated/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscoveryFactoryConfigurer.java
@@ -0,0 +1,174 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.etcd3.cloud;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class Etcd3ServiceDiscoveryFactoryConfigurer extends org.apache.camel.support.component.PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory target = (org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "AuthHeaders": target.setAuthHeaders(property(camelContext, java.util.Map.class, value)); return true;
+ case "authority":
+ case "Authority": target.setAuthority(property(camelContext, java.lang.String.class, value)); return true;
+ case "connectiontimeout":
+ case "ConnectionTimeout": target.setConnectionTimeout(property(camelContext, java.time.Duration.class, value)); return true;
+ case "endpoints":
+ case "Endpoints": target.setEndpoints(property(camelContext, java.lang.String[].class, value)); return true;
+ case "headers":
+ case "Headers": target.setHeaders(property(camelContext, java.util.Map.class, value)); return true;
+ case "keepalivetime":
+ case "KeepAliveTime": target.setKeepAliveTime(property(camelContext, java.time.Duration.class, value)); return true;
+ case "keepalivetimeout":
+ case "KeepAliveTimeout": target.setKeepAliveTimeout(property(camelContext, java.time.Duration.class, value)); return true;
+ case "keycharset":
+ case "KeyCharset": target.setKeyCharset(property(camelContext, java.lang.String.class, value)); return true;
+ case "loadbalancerpolicy":
+ case "LoadBalancerPolicy": target.setLoadBalancerPolicy(property(camelContext, java.lang.String.class, value)); return true;
+ case "maxinboundmessagesize":
+ case "MaxInboundMessageSize": target.setMaxInboundMessageSize(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "namespace":
+ case "Namespace": target.setNamespace(property(camelContext, java.lang.String.class, value)); return true;
+ case "password":
+ case "Password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true;
+ case "retrydelay":
+ case "RetryDelay": target.setRetryDelay(property(camelContext, long.class, value)); return true;
+ case "retrymaxdelay":
+ case "RetryMaxDelay": target.setRetryMaxDelay(property(camelContext, long.class, value)); return true;
+ case "retrymaxduration":
+ case "RetryMaxDuration": target.setRetryMaxDuration(property(camelContext, java.time.Duration.class, value)); return true;
+ case "servicepath":
+ case "ServicePath": target.setServicePath(property(camelContext, java.lang.String.class, value)); return true;
+ case "sslcontext":
+ case "SslContext": target.setSslContext(property(camelContext, io.netty.handler.ssl.SslContext.class, value)); return true;
+ case "type":
+ case "Type": target.setType(property(camelContext, java.lang.String.class, value)); return true;
+ case "username":
+ case "UserName": target.setUserName(property(camelContext, java.lang.String.class, value)); return true;
+ case "valuecharset":
+ case "ValueCharset": target.setValueCharset(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "AuthHeaders": return java.util.Map.class;
+ case "authority":
+ case "Authority": return java.lang.String.class;
+ case "connectiontimeout":
+ case "ConnectionTimeout": return java.time.Duration.class;
+ case "endpoints":
+ case "Endpoints": return java.lang.String[].class;
+ case "headers":
+ case "Headers": return java.util.Map.class;
+ case "keepalivetime":
+ case "KeepAliveTime": return java.time.Duration.class;
+ case "keepalivetimeout":
+ case "KeepAliveTimeout": return java.time.Duration.class;
+ case "keycharset":
+ case "KeyCharset": return java.lang.String.class;
+ case "loadbalancerpolicy":
+ case "LoadBalancerPolicy": return java.lang.String.class;
+ case "maxinboundmessagesize":
+ case "MaxInboundMessageSize": return java.lang.Integer.class;
+ case "namespace":
+ case "Namespace": return java.lang.String.class;
+ case "password":
+ case "Password": return java.lang.String.class;
+ case "retrydelay":
+ case "RetryDelay": return long.class;
+ case "retrymaxdelay":
+ case "RetryMaxDelay": return long.class;
+ case "retrymaxduration":
+ case "RetryMaxDuration": return java.time.Duration.class;
+ case "servicepath":
+ case "ServicePath": return java.lang.String.class;
+ case "sslcontext":
+ case "SslContext": return io.netty.handler.ssl.SslContext.class;
+ case "type":
+ case "Type": return java.lang.String.class;
+ case "username":
+ case "UserName": return java.lang.String.class;
+ case "valuecharset":
+ case "ValueCharset": return java.lang.String.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory target = (org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "AuthHeaders": return target.getAuthHeaders();
+ case "authority":
+ case "Authority": return target.getAuthority();
+ case "connectiontimeout":
+ case "ConnectionTimeout": return target.getConnectionTimeout();
+ case "endpoints":
+ case "Endpoints": return target.getEndpoints();
+ case "headers":
+ case "Headers": return target.getHeaders();
+ case "keepalivetime":
+ case "KeepAliveTime": return target.getKeepAliveTime();
+ case "keepalivetimeout":
+ case "KeepAliveTimeout": return target.getKeepAliveTimeout();
+ case "keycharset":
+ case "KeyCharset": return target.getKeyCharset();
+ case "loadbalancerpolicy":
+ case "LoadBalancerPolicy": return target.getLoadBalancerPolicy();
+ case "maxinboundmessagesize":
+ case "MaxInboundMessageSize": return target.getMaxInboundMessageSize();
+ case "namespace":
+ case "Namespace": return target.getNamespace();
+ case "password":
+ case "Password": return target.getPassword();
+ case "retrydelay":
+ case "RetryDelay": return target.getRetryDelay();
+ case "retrymaxdelay":
+ case "RetryMaxDelay": return target.getRetryMaxDelay();
+ case "retrymaxduration":
+ case "RetryMaxDuration": return target.getRetryMaxDuration();
+ case "servicepath":
+ case "ServicePath": return target.getServicePath();
+ case "sslcontext":
+ case "SslContext": return target.getSslContext();
+ case "type":
+ case "Type": return target.getType();
+ case "username":
+ case "UserName": return target.getUserName();
+ case "valuecharset":
+ case "ValueCharset": return target.getValueCharset();
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "authheaders":
+ case "AuthHeaders": return java.lang.String.class;
+ case "headers":
+ case "Headers": return java.lang.String.class;
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/cloud/etcd-service-discovery b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/cloud/etcd-service-discovery
new file mode 100644
index 00000000000..afa38d07cf3
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/cloud/etcd-service-discovery
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/other.properties b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/component.properties
similarity index 67%
rename from components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/other.properties
rename to components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/component.properties
index 307624feba8..55882e4ce56 100644
--- a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/other.properties
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/component.properties
@@ -1,7 +1,7 @@
# Generated by camel build tools - do NOT edit this file!
-name=etcd3
+components=etcd3
groupId=org.apache.camel
artifactId=camel-etcd3
version=3.20.0-SNAPSHOT
projectName=Camel :: Etcd3
-projectDescription=Aggregation repository using EtcD as datastore
+projectDescription=Camel EtcD v3 component based on jetcd
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/component/etcd3 b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/component/etcd3
new file mode 100644
index 00000000000..f56078fb24a
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/component/etcd3
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.etcd3.Etcd3Component
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/etcd3-component b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/etcd3-component
new file mode 100644
index 00000000000..0a0786bbb50
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/etcd3-component
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.etcd3.Etcd3ComponentConfigurer
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/etcd3-endpoint b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/etcd3-endpoint
new file mode 100644
index 00000000000..6a3fb564b84
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/etcd3-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.etcd3.Etcd3EndpointConfigurer
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory
new file mode 100644
index 00000000000..d92b6551351
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactory
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.etcd3.cloud.Etcd3ServiceDiscoveryFactoryConfigurer
diff --git a/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/urifactory/etcd3-endpoint b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/urifactory/etcd3-endpoint
new file mode 100644
index 00000000000..0e8faa50adf
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/META-INF/services/org/apache/camel/urifactory/etcd3-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.etcd3.Etcd3EndpointUriFactory
diff --git a/components/camel-etcd3/src/generated/resources/org/apache/camel/component/etcd3/etcd3.json b/components/camel-etcd3/src/generated/resources/org/apache/camel/component/etcd3/etcd3.json
new file mode 100644
index 00000000000..a2a1bdf5f48
--- /dev/null
+++ b/components/camel-etcd3/src/generated/resources/org/apache/camel/component/etcd3/etcd3.json
@@ -0,0 +1,86 @@
+{
+ "component": {
+ "kind": "component",
+ "name": "etcd3",
+ "title": "Etcd v3",
+ "description": "Get, set, delete or watch keys in etcd key-value store.",
+ "deprecated": false,
+ "firstVersion": "3.19.0",
+ "label": "clustering,database",
+ "javaType": "org.apache.camel.component.etcd3.Etcd3Component",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-etcd3",
+ "version": "3.19.0-SNAPSHOT",
+ "scheme": "etcd3",
+ "extendsScheme": "",
+ "syntax": "etcd3:path",
+ "async": false,
+ "api": false,
+ "consumerOnly": false,
+ "producerOnly": false,
+ "lenientProperties": false
+ },
+ "componentProperties": {
+ "configuration": { "kind": "property", "displayName": "Configuration", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.etcd3.Etcd3Configuration", "deprecated": false, "autowired": false, "secret": false, "description": "Component configuration." },
+ "endpoints": { "kind": "property", "displayName": "Endpoints", "group": "common", "label": "common", "required": false, "type": "array", "javaType": "java.lang.String[]", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "Etcd3Constants.ETCD_DEFAULT_ENDPOINTS", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd server endpoints using the IPNameResolver." },
+ "keyCharset": { "kind": "property", "displayName": "Key Charset", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "UTF-8", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the charset to use for the keys." },
+ "namespace": { "kind": "property", "displayName": "Namespace", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the namespace of keys used. \/ will be treated as no namespace." },
+ "prefix": { "kind": "property", "displayName": "Prefix", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "To apply an action on all the key-value pairs whose key that starts with the target path." },
+ "bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a me [...]
+ "fromIndex": { "kind": "property", "displayName": "From Index", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "The index to watch from" },
+ "lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during star [...]
+ "valueCharset": { "kind": "property", "displayName": "Value Charset", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "UTF-8", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the charset to use for the values." },
+ "authHeaders": { "kind": "property", "displayName": "Auth Headers", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.String>", "prefix": "authHeaders.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the headers to be added to aut [...]
+ "authority": { "kind": "property", "displayName": "Authority", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the authority used to authenticate connections to servers." },
+ "autowiredEnabled": { "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which t [...]
+ "connectionTimeout": { "kind": "property", "displayName": "Connection Timeout", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the connection timeout." },
+ "headers": { "kind": "property", "displayName": "Headers", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.String>", "prefix": "headers.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the headers to be added to http request hea [...]
+ "keepAliveTime": { "kind": "property", "displayName": "Keep Alive Time", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "30 seconds", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the interval for gRPC keepalives. The current minimum allowed by gRPC is [...]
+ "keepAliveTimeout": { "kind": "property", "displayName": "Keep Alive Timeout", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "10 seconds", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the timeout for gRPC keepalives." },
+ "loadBalancerPolicy": { "kind": "property", "displayName": "Load Balancer Policy", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd load balancer policy." },
+ "maxInboundMessageSize": { "kind": "property", "displayName": "Max Inbound Message Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the maximum message size allowed for a single gRPC frame." },
+ "retryDelay": { "kind": "property", "displayName": "Retry Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the delay between retries in milliseconds." },
+ "retryMaxDelay": { "kind": "property", "displayName": "Retry Max Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 2500, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the max backing off delay between retries in milliseconds." },
+ "retryMaxDuration": { "kind": "property", "displayName": "Retry Max Duration", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the retries max duration." },
+ "servicePath": { "kind": "property", "displayName": "Service Path", "group": "cloud", "label": "cloud", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "\/services\/", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "The path to look for service discovery." },
+ "password": { "kind": "property", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd auth password." },
+ "sslContext": { "kind": "property", "displayName": "Ssl Context", "group": "security", "label": "advanced,security", "required": false, "type": "object", "javaType": "io.netty.handler.ssl.SslContext", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure SSL\/TLS context to use instead of the system default." },
+ "userName": { "kind": "property", "displayName": "User Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd auth user." }
+ },
+ "headers": {
+ "CamelEtcdAction": { "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The action to perform. Supported values: set get delete", "constantName": "org.apache.camel.component.etcd3.Etcd3Constants#ETCD_ACTION" },
+ "CamelEtcdPath": { "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The target path", "constantName": "org.apache.camel.component.etcd3.Etcd3Constants#ETCD_PATH" },
+ "CamelEtcdIsPrefix": { "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "To apply an action on all the key-value pairs whose key that starts with the target path.", "constantName": "org.apache.camel.component.etcd3.Etcd3Constants#ETCD_IS_PREFIX" },
+ "CamelEtcdKeyCharset": { "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The charset to use for the keys.", "constantName": "org.apache.camel.component.etcd3.Etcd3Constants#ETCD_KEY_CHARSET" },
+ "CamelEtcdValueCharset": { "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The charset to use for the values.", "constantName": "org.apache.camel.component.etcd3.Etcd3Constants#ETCD_VALUE_CHARSET" }
+ },
+ "properties": {
+ "path": { "kind": "path", "displayName": "Path", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The path the endpoint refers to" },
+ "endpoints": { "kind": "parameter", "displayName": "Endpoints", "group": "common", "label": "common", "required": false, "type": "array", "javaType": "java.lang.String[]", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "Etcd3Constants.ETCD_DEFAULT_ENDPOINTS", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd server endpoints using the IPNameResolver." },
+ "keyCharset": { "kind": "parameter", "displayName": "Key Charset", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "UTF-8", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the charset to use for the keys." },
+ "namespace": { "kind": "parameter", "displayName": "Namespace", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the namespace of keys used. \/ will be treated as no namespace." },
+ "prefix": { "kind": "parameter", "displayName": "Prefix", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "To apply an action on all the key-value pairs whose key that starts with the target path." },
+ "bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now [...]
+ "exceptionHandler": { "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the con [...]
+ "exchangePattern": { "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut", "InOptionalOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
+ "fromIndex": { "kind": "parameter", "displayName": "From Index", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "The index to watch from" },
+ "valueCharset": { "kind": "parameter", "displayName": "Value Charset", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "UTF-8", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the charset to use for the values." },
+ "lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may other [...]
+ "authHeaders": { "kind": "parameter", "displayName": "Auth Headers", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.String>", "prefix": "authHeaders.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the headers to be added to au [...]
+ "authority": { "kind": "parameter", "displayName": "Authority", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the authority used to authenticate connections to servers." },
+ "connectionTimeout": { "kind": "parameter", "displayName": "Connection Timeout", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the connection timeout." },
+ "headers": { "kind": "parameter", "displayName": "Headers", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.String>", "prefix": "headers.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the headers to be added to http request he [...]
+ "keepAliveTime": { "kind": "parameter", "displayName": "Keep Alive Time", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "30 seconds", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the interval for gRPC keepalives. The current minimum allowed by gRPC i [...]
+ "keepAliveTimeout": { "kind": "parameter", "displayName": "Keep Alive Timeout", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "10 seconds", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the timeout for gRPC keepalives." },
+ "loadBalancerPolicy": { "kind": "parameter", "displayName": "Load Balancer Policy", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd load balancer policy." },
+ "maxInboundMessageSize": { "kind": "parameter", "displayName": "Max Inbound Message Size", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the maximum message size allowed for a single gRPC frame." },
+ "retryDelay": { "kind": "parameter", "displayName": "Retry Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the delay between retries in milliseconds." },
+ "retryMaxDelay": { "kind": "parameter", "displayName": "Retry Max Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 2500, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the max backing off delay between retries in milliseconds." },
+ "retryMaxDuration": { "kind": "parameter", "displayName": "Retry Max Duration", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure the retries max duration." },
+ "servicePath": { "kind": "parameter", "displayName": "Service Path", "group": "cloud", "label": "cloud", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "\/services\/", "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "The path to look for service discovery." },
+ "password": { "kind": "parameter", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd auth password." },
+ "sslContext": { "kind": "parameter", "displayName": "Ssl Context", "group": "security", "label": "advanced,security", "required": false, "type": "object", "javaType": "io.netty.handler.ssl.SslContext", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure SSL\/TLS context to use instead of the system default." },
+ "userName": { "kind": "parameter", "displayName": "User Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.etcd3.Etcd3Configuration", "configurationField": "configuration", "description": "Configure etcd auth user." }
+ }
+}
diff --git a/components/camel-etcd3/src/main/docs/etcd3.adoc b/components/camel-etcd3/src/main/docs/etcd3.adoc
index 5463f2f63c8..a1061092ef0 100644
--- a/components/camel-etcd3/src/main/docs/etcd3.adoc
+++ b/components/camel-etcd3/src/main/docs/etcd3.adoc
@@ -10,6 +10,96 @@
*Since Camel {since}*
-The Etcd3 component provides an `AggregationStrategy` to use Etcd3 as the backend datastore.
+*{component-header}*
+
+The camel Etcd component allows you to work with Etcd, a distributed reliable key-value store.
+
+Maven users will need to add the following dependency to their `pom.xml`
+for this component:
+
+[source,xml]
+----
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-etcd3</artifactId>
+ <version>x.x.x</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+----
+
+== URI Format
+
+----------------------------
+etcd3:path[?options]
+----------------------------
+
+// component-configure options: START
+
+// component-configure options: END
+
+// component options: START
+include::partial$component-configure-options.adoc[]
+include::partial$component-endpoint-options.adoc[]
+// component options: END
+
+// endpoint options: START
+
+// endpoint options: END
+
+// component headers: START
+include::partial$component-endpoint-headers.adoc[]
+// component headers: END
+
+== Producer Operations (Since 3.20)
+
+The following ETCD operations are currently supported. Simply
+set the exchange header with a key of "CamelEtcdAction"
+and a value set to one of the following.
+
+[width="100%",cols="10%,10%,10%,70%",options="header",]
+|===
+|operation |input message body | output message body |description
+
+|set |*String* value of the key-value pair to put |*PutResponse* result of a put operation| Puts a new key-value pair into etcd where the option "path" or the exchange header "CamelEtcdPath" is the key.
+You can set the key charset by setting the exchange header with
+the key "CamelEtcdKeyCharset".
+You can set the value charset by setting the exchange header with
+the key "CamelEtcdValueCharset".
+
+|get |None |*GetResponse* result of the get operation | Retrieves the key-value pair(s) that match with the key corresponding to
+the option "path" or the exchange header "CamelEtcdPath".
+You can set the key charset by setting the exchange header with
+the key "CamelEtcdKeyCharset".
+You indicate if the key is a prefix by setting the exchange header with
+the key "CamelEtcdIsPrefix" to true.
+
+|delete |None |*DeleteResponse* result of the delete operation |Deletes the key-value pair(s) that match with the key corresponding to
+the option "path" or the exchange header "CamelEtcdPath".
+You can set the key charset by setting the exchange header with
+the key "CamelEtcdKeyCharset".
+You indicate if the key is a prefix by setting the exchange header with
+the key "CamelEtcdIsPrefix" to true.
+
+== Consumer (Since 3.20)
+
+The consumer of the etcd components allows to watch changes on the matching key-value pair(s). One exchange is created per event with the header "CamelEtcdPath" set to the path of the corresponding key-value pair and the body
+of type *WatchEvent*.
+
+You can set the key charset by setting the exchange header with
+the key "CamelEtcdKeyCharset".
+You indicate if the key is a prefix by setting the exchange header with
+the key "CamelEtcdIsPrefix" to true.
+
+By default, the consumer receives only the latest changes, but it is also possible to start watching events from a specific revision by setting the option "fromIndex" to the expected starting index.
+
+== AggregationRepository
+
+The Etcd3 component provides an `AggregationStrategy` to use Etcd as the backend datastore.
+
+== RoutePolicy (Since 3.20)
+
+The Etcd3 component provides a `RoutePolicy` to use Etcd as clustered lock.
+
+|===
include::spring-boot:partial$starter.adoc[]
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Component.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Component.java
new file mode 100644
index 00000000000..3925691a7d6
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Component.java
@@ -0,0 +1,61 @@
+/*
+ * 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.etcd3;
+
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.ObjectHelper;
+
+@Component("etcd3")
+public class Etcd3Component extends DefaultComponent {
+
+ @Metadata
+ private Etcd3Configuration configuration = new Etcd3Configuration();
+
+ public Etcd3Configuration getConfiguration() {
+ return configuration;
+ }
+
+ /**
+ * Component configuration.
+ */
+ public void setConfiguration(Etcd3Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ // path must start with leading slash
+ String path = ObjectHelper.isEmpty(remaining) ? "/" : remaining;
+ if (!path.startsWith("/")) {
+ path = String.format("/%s", path);
+ }
+
+ Endpoint endpoint = new Etcd3Endpoint(uri, this, loadConfiguration(), path);
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+
+ private Etcd3Configuration loadConfiguration() {
+ return Optional.ofNullable(this.configuration).orElseGet(Etcd3Configuration::new).copy();
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Configuration.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Configuration.java
new file mode 100644
index 00000000000..fcf7940fbe8
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Configuration.java
@@ -0,0 +1,360 @@
+/*
+ * 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.etcd3;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.ClientBuilder;
+import io.netty.handler.ssl.SslContext;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+
+import static org.apache.camel.component.etcd3.Etcd3Constants.ETCD_DEFAULT_ENDPOINTS;
+
+@UriParams
+public class Etcd3Configuration implements Cloneable {
+
+ @UriParam(label = "common", defaultValue = "Etcd3Constants.ETCD_DEFAULT_ENDPOINTS")
+ private String[] endpoints = ETCD_DEFAULT_ENDPOINTS;
+ @UriParam(label = "security", secret = true)
+ private String userName;
+ @UriParam(label = "security", secret = true)
+ private String password;
+ @UriParam(label = "advanced,security")
+ private SslContext sslContext;
+ @UriParam(label = "common")
+ private String namespace;
+ @UriParam(label = "advanced")
+ private String loadBalancerPolicy;
+ @UriParam(label = "advanced")
+ private String authority;
+ @UriParam(label = "advanced")
+ private Integer maxInboundMessageSize;
+ @UriParam(label = "advanced", prefix = "headers.", multiValue = true)
+ private Map<String, String> headers = new HashMap<>();
+ @UriParam(label = "advanced", prefix = "authHeaders.", multiValue = true)
+ private Map<String, String> authHeaders = new HashMap<>();
+ @UriParam(label = "advanced", defaultValue = "500")
+ private long retryDelay = 500L;
+ @UriParam(label = "advanced", defaultValue = "2500")
+ private long retryMaxDelay = 2500L;
+ @UriParam(label = "advanced", defaultValue = "30 seconds")
+ private Duration keepAliveTime = Duration.ofSeconds(30L);
+ @UriParam(label = "advanced", defaultValue = "10 seconds")
+ private Duration keepAliveTimeout = Duration.ofSeconds(10L);
+ @UriParam(label = "advanced")
+ private Duration retryMaxDuration;
+ @UriParam(label = "advanced")
+ private Duration connectionTimeout;
+ @UriParam(label = "common", defaultValue = "false")
+ private boolean prefix;
+ @UriParam(label = "consumer,advanced", defaultValue = "0", description = "The index to watch from")
+ private long fromIndex;
+ @UriParam(label = "cloud", defaultValue = "/services/")
+ private String servicePath = "/services/";
+ @UriParam(label = "common", defaultValue = "UTF-8")
+ private String keyCharset = "UTF-8";
+ @UriParam(label = "producer", defaultValue = "UTF-8")
+ private String valueCharset = "UTF-8";
+
+ public String[] getEndpoints() {
+ return endpoints;
+ }
+
+ /**
+ * Configure etcd server endpoints using the IPNameResolver.
+ */
+ public void setEndpoints(String... endpoints) {
+ this.endpoints = endpoints;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ /**
+ * Configure etcd auth user.
+ */
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * Configure etcd auth password.
+ */
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public String getNamespace() {
+ return namespace;
+ }
+
+ /**
+ * Configure the namespace of keys used. "/" will be treated as no namespace.
+ */
+ public void setNamespace(String namespace) {
+ this.namespace = namespace;
+ }
+
+ public SslContext getSslContext() {
+ return sslContext;
+ }
+
+ /**
+ * Configure SSL/TLS context to use instead of the system default.
+ */
+ public void setSslContext(SslContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
+ public String getLoadBalancerPolicy() {
+ return loadBalancerPolicy;
+ }
+
+ /**
+ * Configure etcd load balancer policy.
+ */
+ public void setLoadBalancerPolicy(String loadBalancerPolicy) {
+ this.loadBalancerPolicy = loadBalancerPolicy;
+ }
+
+ public String getAuthority() {
+ return authority;
+ }
+
+ /**
+ * Configure the authority used to authenticate connections to servers.
+ */
+ public void setAuthority(String authority) {
+ this.authority = authority;
+ }
+
+ public Integer getMaxInboundMessageSize() {
+ return maxInboundMessageSize;
+ }
+
+ /**
+ * Configure the maximum message size allowed for a single gRPC frame.
+ */
+ public void setMaxInboundMessageSize(Integer maxInboundMessageSize) {
+ this.maxInboundMessageSize = maxInboundMessageSize;
+ }
+
+ public Map<String, String> getHeaders() {
+ return headers;
+ }
+
+ /**
+ * Configure the headers to be added to http request headers.
+ */
+ public void setHeaders(Map<String, String> headers) {
+ this.headers = headers;
+ }
+
+ public Map<String, String> getAuthHeaders() {
+ return authHeaders;
+ }
+
+ /**
+ * Configure the headers to be added to auth request headers.
+ */
+ public void setAuthHeaders(Map<String, String> authHeaders) {
+ this.authHeaders = authHeaders;
+ }
+
+ public long getRetryDelay() {
+ return retryDelay;
+ }
+
+ /**
+ * Configure the delay between retries in milliseconds.
+ */
+ public void setRetryDelay(long retryDelay) {
+ this.retryDelay = retryDelay;
+ }
+
+ public long getRetryMaxDelay() {
+ return retryMaxDelay;
+ }
+
+ /**
+ * Configure the max backing off delay between retries in milliseconds.
+ */
+ public void setRetryMaxDelay(long retryMaxDelay) {
+ this.retryMaxDelay = retryMaxDelay;
+ }
+
+ public Duration getKeepAliveTime() {
+ return keepAliveTime;
+ }
+
+ /**
+ * Configure the interval for gRPC keepalives. The current minimum allowed by gRPC is 10 seconds.
+ */
+ public void setKeepAliveTime(Duration keepAliveTime) {
+ this.keepAliveTime = keepAliveTime;
+ }
+
+ public Duration getKeepAliveTimeout() {
+ return keepAliveTimeout;
+ }
+
+ /**
+ * Configure the timeout for gRPC keepalives.
+ */
+ public void setKeepAliveTimeout(Duration keepAliveTimeout) {
+ this.keepAliveTimeout = keepAliveTimeout;
+ }
+
+ public Duration getRetryMaxDuration() {
+ return retryMaxDuration;
+ }
+
+ /**
+ * Configure the retries max duration.
+ */
+ public void setRetryMaxDuration(Duration retryMaxDuration) {
+ this.retryMaxDuration = retryMaxDuration;
+ }
+
+ public Duration getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ /**
+ * Configure the connection timeout.
+ */
+ public void setConnectionTimeout(Duration connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ public boolean isPrefix() {
+ return prefix;
+ }
+
+ /**
+ * To apply an action on all the key-value pairs whose key that starts with the target path.
+ */
+ public void setPrefix(boolean prefix) {
+ this.prefix = prefix;
+ }
+
+ public long getFromIndex() {
+ return fromIndex;
+ }
+
+ /**
+ * The index to watch from.
+ */
+ public void setFromIndex(long fromIndex) {
+ this.fromIndex = fromIndex;
+ }
+
+ public String getServicePath() {
+ return servicePath;
+ }
+
+ /**
+ * The path to look for service discovery.
+ */
+ public void setServicePath(String servicePath) {
+ this.servicePath = servicePath;
+ }
+
+ public String getKeyCharset() {
+ return keyCharset;
+ }
+
+ /**
+ * Configure the charset to use for the keys.
+ */
+ public void setKeyCharset(String keyCharset) {
+ this.keyCharset = keyCharset;
+ }
+
+ public String getValueCharset() {
+ return valueCharset;
+ }
+
+ /**
+ * Configure the charset to use for the values.
+ */
+ public void setValueCharset(String valueCharset) {
+ this.valueCharset = valueCharset;
+ }
+
+ /**
+ * @return a {@link Client} instance configured with all parameters set.
+ */
+ public Client createClient() {
+ final ClientBuilder builder = Client.builder()
+ .endpoints(endpoints)
+ .sslContext(sslContext)
+ .authority(authority)
+ .maxInboundMessageSize(maxInboundMessageSize)
+ .retryDelay(retryDelay)
+ .retryMaxDelay(retryMaxDelay)
+ .retryMaxDuration(retryMaxDuration)
+ .keepaliveTime(keepAliveTime)
+ .keepaliveTimeout(keepAliveTimeout)
+ .connectTimeout(connectionTimeout);
+ if (loadBalancerPolicy != null) {
+ builder.loadBalancerPolicy(loadBalancerPolicy);
+ }
+ if (userName != null) {
+ builder.user(ByteSequence.from(userName.getBytes()));
+ }
+ if (password != null) {
+ builder.password(ByteSequence.from(password.getBytes()));
+ }
+ if (namespace != null) {
+ builder.namespace(ByteSequence.from(namespace.getBytes()));
+ }
+ if (headers != null && !headers.isEmpty()) {
+ for (Map.Entry<String, String> entry : headers.entrySet()) {
+ builder.header(entry.getKey(), entry.getKey());
+ }
+ }
+ if (authHeaders != null && !authHeaders.isEmpty()) {
+ for (Map.Entry<String, String> entry : authHeaders.entrySet()) {
+ builder.authHeader(entry.getKey(), entry.getKey());
+ }
+ }
+ return builder.build();
+ }
+
+ Etcd3Configuration copy() {
+ try {
+ Etcd3Configuration configuration = (Etcd3Configuration) super.clone();
+ configuration.setHeaders(new HashMap<>(headers));
+ configuration.setAuthHeaders(new HashMap<>(authHeaders));
+ return configuration;
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Constants.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Constants.java
new file mode 100644
index 00000000000..694bc6b3308
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Constants.java
@@ -0,0 +1,50 @@
+/*
+ * 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.etcd3;
+
+import org.apache.camel.spi.Metadata;
+
+public abstract class Etcd3Constants {
+
+ @Metadata(label = "producer", description = "The action to perform.\n" +
+ "Supported values:\n" +
+ "\n" +
+ "* set\n" +
+ "* get\n" +
+ "* delete\n",
+ javaType = "String")
+ public static final String ETCD_ACTION = "CamelEtcdAction";
+ @Metadata(description = "The target path", javaType = "String")
+ public static final String ETCD_PATH = "CamelEtcdPath";
+ @Metadata(label = "producer",
+ description = "To apply an action on all the key-value pairs whose key that starts with the target path.",
+ javaType = "Boolean")
+ public static final String ETCD_IS_PREFIX = "CamelEtcdIsPrefix";
+ @Metadata(label = "producer", description = "The charset to use for the keys.", javaType = "String")
+ public static final String ETCD_KEY_CHARSET = "CamelEtcdKeyCharset";
+ @Metadata(label = "producer", description = "The charset to use for the values.", javaType = "String")
+ public static final String ETCD_VALUE_CHARSET = "CamelEtcdValueCharset";
+
+ public static final String ETCD_KEYS_ACTION_SET = "set";
+ public static final String ETCD_KEYS_ACTION_DELETE = "delete";
+ public static final String ETCD_KEYS_ACTION_GET = "get";
+
+ public static final String[] ETCD_DEFAULT_ENDPOINTS = new String[] { "http://localhost:2379" };
+
+ private Etcd3Constants() {
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Consumer.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Consumer.java
new file mode 100644
index 00000000000..246a38df01d
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Consumer.java
@@ -0,0 +1,150 @@
+/*
+ * 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.etcd3;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.Watch;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchEvent;
+import io.etcd.jetcd.watch.WatchResponse;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.util.StringHelper;
+
+/**
+ * A consumer allowing to watch key-value pairs stored into etcd v3.
+ */
+class Etcd3Consumer extends DefaultConsumer implements Watch.Listener {
+
+ /**
+ * The configuration of the consumer.
+ */
+ private final Etcd3Configuration configuration;
+ /**
+ * The path of the key-value pair to watch.
+ */
+ private final String path;
+ /**
+ * The client to access to etcd.
+ */
+ private final Client client;
+ /**
+ * The client to watch key-value pairs stored into etcd.
+ */
+ private final Watch watch;
+ /**
+ * The revision from which the changes of the key-value pair are watched.
+ */
+ private final AtomicLong revision;
+ /**
+ * The charset to use for the keys.
+ */
+ private final Charset keyCharset;
+ /**
+ * The current watcher used to watch the changes of the target key-value pair.
+ */
+ private final AtomicReference<Watch.Watcher> watcher = new AtomicReference<>();
+
+ /**
+ * Construct a {@code Etcd3Consumer} with the given parameters.
+ *
+ * @param endpoint the endpoint corresponding to the consumer.
+ * @param processor the processor corresponding to the consumer.
+ * @param configuration the configuration of the consumer.
+ * @param path the path of the key-value pair to watch.
+ */
+ Etcd3Consumer(Etcd3Endpoint endpoint, Processor processor, Etcd3Configuration configuration,
+ String path) {
+ super(endpoint, processor);
+
+ this.configuration = configuration;
+ this.path = StringHelper.notEmpty(path, "path");
+ this.client = configuration.createClient();
+ this.watch = client.getWatchClient();
+ this.revision = new AtomicLong(configuration.getFromIndex());
+ this.keyCharset = Charset.forName(configuration.getKeyCharset());
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ doWatch();
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ try {
+ client.close();
+ } finally {
+ super.doStop();
+ }
+ }
+
+ /**
+ * If allowed, starts to watch the changes on the target key-value pair.
+ */
+ private void doWatch() {
+ if (!isRunAllowed()) {
+ return;
+ }
+ watcher.getAndUpdate(w -> {
+ if (w != null) {
+ w.close();
+ }
+ return watch.watch(
+ ByteSequence.from(path, keyCharset),
+ WatchOption.newBuilder().isPrefix(configuration.isPrefix()).withRevision(revision.get()).build(),
+ this);
+ });
+ }
+
+ @Override
+ public void onNext(WatchResponse response) {
+ for (WatchEvent event : response.getEvents()) {
+ final Exchange exchange = createExchange(false);
+ final KeyValue keyValue = event.getKeyValue();
+ exchange.getIn().setHeader(Etcd3Constants.ETCD_PATH, keyValue.getKey().toString(keyCharset));
+ exchange.getIn().setBody(event);
+ // Watch from the revision + 1 of the node we got for ensuring
+ // no events are missed between watch commands
+ revision.getAndUpdate(r -> Math.max(r, keyValue.getModRevision() + 1));
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Error processing exchange", exchange, e);
+ }
+ releaseExchange(exchange, false);
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ handleException("Error processing etcd response", throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ doWatch();
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Endpoint.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Endpoint.java
new file mode 100644
index 00000000000..ab640c095ca
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Endpoint.java
@@ -0,0 +1,60 @@
+/*
+ * 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.etcd3;
+
+import org.apache.camel.Category;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.support.DefaultEndpoint;
+
+/**
+ * Get, set, delete or watch keys in etcd key-value store.
+ */
+@UriEndpoint(firstVersion = "3.19.0", scheme = "etcd3", title = "Etcd v3",
+ syntax = "etcd3:path", category = { Category.CLUSTERING, Category.DATABASE },
+ headersClass = Etcd3Constants.class)
+public class Etcd3Endpoint extends DefaultEndpoint {
+
+ @UriPath(label = "common", description = "The path the endpoint refers to")
+ private final String path;
+ @UriParam
+ private final Etcd3Configuration configuration;
+
+ public Etcd3Endpoint(String uri, Etcd3Component component, Etcd3Configuration configuration, String path) {
+ super(uri, component);
+ this.path = path;
+ this.configuration = configuration;
+ }
+
+ public Etcd3Configuration getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new Etcd3Producer(this, configuration, path);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ return new Etcd3Consumer(this, processor, configuration, path);
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Helper.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Helper.java
new file mode 100644
index 00000000000..85c8f1747c7
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Helper.java
@@ -0,0 +1,52 @@
+/*
+ * 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.etcd3;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * The helper class for the etcd component.
+ */
+public final class Etcd3Helper {
+
+ private Etcd3Helper() {
+ }
+
+ /**
+ * @return a specific {@code ObjectMapper} matching with the expectations..
+ */
+ public static ObjectMapper createObjectMapper() {
+ return new ObjectMapper()
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ .setSerializationInclusion(JsonInclude.Include.NON_NULL);
+ }
+
+ /**
+ * Converts the given path as path prefix.
+ *
+ * @param path the path to convert.
+ * @return the given path if it ends with slash, the given path with an appended slash otherwise.
+ */
+ public static String toPathPrefix(String path) {
+ if (path.endsWith("/")) {
+ return path;
+ }
+ return String.format("%s/", path);
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Producer.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Producer.java
new file mode 100644
index 00000000000..2dcb5e49194
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/Etcd3Producer.java
@@ -0,0 +1,229 @@
+/*
+ * 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.etcd3;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.CompletableFuture;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.options.DeleteOption;
+import io.etcd.jetcd.options.GetOption;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.util.StringHelper;
+
+/**
+ * A producer allowing to get, delete and update key-value pairs from etcd v3 asynchronous.
+ */
+class Etcd3Producer extends DefaultAsyncProducer {
+
+ /**
+ * The configuration of the producer.
+ */
+ private final Etcd3Configuration configuration;
+ /**
+ * The default path against which the action is performed.
+ */
+ private final String path;
+ /**
+ * The client to access to etcd.
+ */
+ private final Client client;
+ /**
+ * The client to access to the key-value pairs stored into etcd.
+ */
+ private final KV kvClient;
+ /**
+ * The default charset to use for the keys.
+ */
+ private final Charset defaultKeyCharset;
+ /**
+ * The default charset to use for the values.
+ */
+ private final Charset defaultValueCharset;
+
+ /**
+ * Construct a {@code Etcd3Producer} with the given parameters.
+ *
+ * @param endpoint the endpoint corresponding to the producer.
+ * @param configuration the configuration of the producer.
+ * @param path the default path against which the action is performed.
+ */
+ Etcd3Producer(Etcd3Endpoint endpoint, Etcd3Configuration configuration, String path) {
+ super(endpoint);
+ this.configuration = configuration;
+ this.path = path;
+ this.client = configuration.createClient();
+ this.kvClient = client.getKVClient();
+ this.defaultKeyCharset = Charset.forName(configuration.getKeyCharset());
+ this.defaultValueCharset = Charset.forName(configuration.getValueCharset());
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ try {
+ String action = exchange.getIn().getHeader(Etcd3Constants.ETCD_ACTION, String.class);
+ String targetPath = exchange.getIn().getHeader(Etcd3Constants.ETCD_PATH, String.class);
+ if (targetPath == null) {
+ targetPath = path;
+ }
+
+ StringHelper.notEmpty(targetPath, Etcd3Constants.ETCD_PATH);
+ StringHelper.notEmpty(action, Etcd3Constants.ETCD_ACTION);
+
+ switch (action) {
+ case Etcd3Constants.ETCD_KEYS_ACTION_SET:
+ processSetAsync(targetPath, exchange, callback);
+ break;
+ case Etcd3Constants.ETCD_KEYS_ACTION_GET:
+ processGetAsync(targetPath, exchange, callback);
+ break;
+ case Etcd3Constants.ETCD_KEYS_ACTION_DELETE:
+ processDelAsync(targetPath, exchange, callback);
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown action " + action);
+ }
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ try {
+ client.close();
+ } finally {
+ super.doStop();
+ }
+ }
+
+ /**
+ * Add actions to perform once the given future is complete.
+ *
+ * @param future the future to complete with specific actions.
+ * @param exchange the exchange into which the result of the future (response or exception) is set.
+ * @param callback the callback to call once the future is done.
+ * @param <T> the result type returned by the future.
+ */
+ private <T> void onComplete(CompletableFuture<T> future, Exchange exchange, AsyncCallback callback) {
+ future.thenAccept(r -> exchange.getIn().setBody(r))
+ .whenComplete(
+ (r, e) -> {
+ try {
+ if (e != null) {
+ exchange.setException(new CamelExchangeException(
+ "An error occurred while executing the action", exchange, e));
+ }
+ } finally {
+ callback.done(false);
+ }
+ });
+ }
+
+ /**
+ * Deletes asynchronously the key-value pair corresponding to the given path.
+ *
+ * @param targetPath the key of the key-value pair to delete.
+ * @param exchange the exchange into which the result (delete response or exception) is set.
+ * @param callback the callback to call once the delete operation is done.
+ */
+ private void processDelAsync(String targetPath, Exchange exchange, AsyncCallback callback) {
+ onComplete(
+ kvClient.delete(
+ ByteSequence.from(targetPath, getKeyCharset(exchange)),
+ DeleteOption.newBuilder().isPrefix(isPrefix(exchange)).build()),
+ exchange, callback);
+ }
+
+ /**
+ * Gets asynchronously the key-value pair corresponding to the given path.
+ *
+ * @param targetPath the key of the key-value pair to get.
+ * @param exchange the exchange into which the result (get response or exception) is set.
+ * @param callback the callback to call once the get operation is done.
+ */
+ private void processGetAsync(String targetPath, Exchange exchange, AsyncCallback callback) {
+ onComplete(
+ kvClient.get(
+ ByteSequence.from(targetPath, getKeyCharset(exchange)),
+ GetOption.newBuilder().isPrefix(isPrefix(exchange)).build()),
+ exchange, callback);
+ }
+
+ /**
+ * Puts asynchronously a key-value pair with the given path as key and the message body as value.
+ *
+ * @param targetPath the key of the key-value pair to put.
+ * @param exchange the exchange from which the message body is extracted and into which the result (put response
+ * or exception) is set.
+ * @param callback the callback to call once the put operation is done.
+ */
+ private void processSetAsync(String targetPath, Exchange exchange, AsyncCallback callback) {
+ onComplete(
+ kvClient.put(
+ ByteSequence.from(targetPath, getKeyCharset(exchange)),
+ ByteSequence.from(exchange.getIn().getBody(String.class), getValueCharset(exchange))),
+ exchange, callback);
+ }
+
+ /**
+ * Indicates whether the path given for the operation is a prefix or not.
+ *
+ * @param exchange the exchange from which the value of the header {@link Etcd3Constants#ETCD_IS_PREFIX} is
+ * extracted.
+ * @return the value of the header {@link Etcd3Constants#ETCD_IS_PREFIX} if it has been set, otherwise the
+ * value extracted from the configuration.
+ */
+ private boolean isPrefix(Exchange exchange) {
+ final Boolean isPrefix = exchange.getIn().getHeader(Etcd3Constants.ETCD_IS_PREFIX, Boolean.class);
+ return isPrefix == null ? configuration.isPrefix() : isPrefix;
+ }
+
+ /**
+ * Indicates the charset to use for the keys.
+ *
+ * @param exchange the exchange from which the value of the header {@link Etcd3Constants#ETCD_KEY_CHARSET} is
+ * extracted.
+ * @return the value of the header {@link Etcd3Constants#ETCD_KEY_CHARSET} if it has been set, otherwise
+ * the value extracted from the configuration.
+ */
+ private Charset getKeyCharset(Exchange exchange) {
+ final String charset = exchange.getIn().getHeader(Etcd3Constants.ETCD_KEY_CHARSET, String.class);
+ return charset == null ? defaultKeyCharset : Charset.forName(charset);
+ }
+
+ /**
+ * Indicates the charset to use for the values.
+ *
+ * @param exchange the exchange from which the value of the header {@link Etcd3Constants#ETCD_VALUE_CHARSET} is
+ * extracted.
+ * @return the value of the header {@link Etcd3Constants#ETCD_VALUE_CHARSET} if it has been set, otherwise
+ * the value extracted from the configuration.
+ */
+ private Charset getValueCharset(Exchange exchange) {
+ final String charset = exchange.getIn().getHeader(Etcd3Constants.ETCD_VALUE_CHARSET, String.class);
+ return charset == null ? defaultValueCharset : Charset.forName(charset);
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3GetServicesResponse.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3GetServicesResponse.java
new file mode 100644
index 00000000000..8b511cc60bb
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3GetServicesResponse.java
@@ -0,0 +1,61 @@
+/*
+ * 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.etcd3.cloud;
+
+import java.util.List;
+
+import org.apache.camel.cloud.ServiceDefinition;
+
+/**
+ * A plain Java object representing the list of services that could be found at a specific revision.
+ */
+final class Etcd3GetServicesResponse {
+
+ /**
+ * The revision at which the list of services has been retrieved.
+ */
+ private final long revision;
+ /**
+ * The list of services found.
+ */
+ private final List<ServiceDefinition> services;
+
+ /**
+ * Construct a {@code Etcd3GetServicesResponse} with the given parameters.
+ *
+ * @param revision the revision at which the list of services has been retrieved.
+ * @param services the list of services found.
+ */
+ Etcd3GetServicesResponse(long revision, List<ServiceDefinition> services) {
+ this.revision = revision;
+ this.services = services;
+ }
+
+ /**
+ * @return the revision at which the list of services has been retrieved.
+ */
+ long getRevision() {
+ return revision;
+ }
+
+ /**
+ * @return the list of services found.
+ */
+ List<ServiceDefinition> getServices() {
+ return services;
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3OnDemandServiceDiscovery.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3OnDemandServiceDiscovery.java
new file mode 100644
index 00000000000..2e5353a4058
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3OnDemandServiceDiscovery.java
@@ -0,0 +1,36 @@
+/*
+ * 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.etcd3.cloud;
+
+import java.util.List;
+
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.component.etcd3.Etcd3Configuration;
+
+/**
+ * An implementation of a {@link Etcd3ServiceDiscovery} that retrieves the service definitions from etcd at each call.
+ */
+public class Etcd3OnDemandServiceDiscovery extends Etcd3ServiceDiscovery {
+ public Etcd3OnDemandServiceDiscovery(Etcd3Configuration configuration) {
+ super(configuration);
+ }
+
+ @Override
+ public List<ServiceDefinition> getServices(String name) {
+ return findServices(s -> name.equalsIgnoreCase(s.getName())).getServices();
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDefinition.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDefinition.java
new file mode 100644
index 00000000000..24cace9536e
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDefinition.java
@@ -0,0 +1,39 @@
+/*
+ * 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.etcd3.cloud;
+
+import java.util.Comparator;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.camel.impl.cloud.DefaultServiceDefinition;
+
+public class Etcd3ServiceDefinition extends DefaultServiceDefinition {
+ public static final Comparator<Etcd3ServiceDefinition> COMPARATOR = Comparator.comparing(Etcd3ServiceDefinition::getHost)
+ .thenComparingInt(Etcd3ServiceDefinition::getPort);
+
+ @JsonCreator
+ public Etcd3ServiceDefinition(
+ @JsonProperty("id") final String id,
+ @JsonProperty("name") final String name,
+ @JsonProperty("address") final String address,
+ @JsonProperty("port") final Integer port,
+ @JsonProperty("tags") final Map<String, String> tags) {
+ super(id, name, address, port, tags);
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscovery.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscovery.java
new file mode 100644
index 00000000000..020504dbc60
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscovery.java
@@ -0,0 +1,159 @@
+/*
+ * 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.etcd3.cloud;
+
+import java.nio.charset.Charset;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.options.GetOption;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.component.etcd3.Etcd3Configuration;
+import org.apache.camel.component.etcd3.Etcd3Helper;
+import org.apache.camel.impl.cloud.DefaultServiceDiscovery;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.etcd3.Etcd3Helper.toPathPrefix;
+
+/**
+ * The root implementation of {@code ServiceDiscovery} fetching the data from etcd.
+ */
+abstract class Etcd3ServiceDiscovery extends DefaultServiceDiscovery {
+ /**
+ * The logger.
+ */
+ private static final Logger LOGGER = LoggerFactory.getLogger(Etcd3ServiceDiscovery.class);
+ /**
+ * The object mapper used to unmarshall the service definitions.
+ */
+ private static final ObjectMapper MAPPER = Etcd3Helper.createObjectMapper();
+
+ /**
+ * The path prefix of the key-value pairs containing the service definitions.
+ */
+ private final String servicePath;
+ /**
+ * The client to access to etcd.
+ */
+ private final Client client;
+ /**
+ * The client to access to the key-value pairs stored into etcd.
+ */
+ private final KV kvClient;
+ /**
+ * The charset to use for the keys.
+ */
+ private final Charset keyCharset;
+ /**
+ * The charset to use for the keys.
+ */
+ private final Charset valueCharset;
+
+ /**
+ * Construct a {@code Etcd3ServiceDiscovery} with the given configuration.
+ *
+ * @param configuration the configuration used to set up the service discovery.
+ */
+ Etcd3ServiceDiscovery(Etcd3Configuration configuration) {
+ this.servicePath = ObjectHelper.notNull(configuration.getServicePath(), "servicePath");
+ this.client = configuration.createClient();
+ this.kvClient = client.getKVClient();
+ this.keyCharset = Charset.forName(configuration.getKeyCharset());
+ this.valueCharset = Charset.forName(configuration.getValueCharset());
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ try {
+ client.close();
+ } finally {
+ super.doStop();
+ }
+ }
+
+ /**
+ * @return all the service definitions that could be found.
+ */
+ protected Etcd3GetServicesResponse findServices() {
+ return findServices(s -> true);
+ }
+
+ /**
+ * @param filter the filter to apply.
+ * @return all the matching service definitions that could be found.
+ */
+ protected Etcd3GetServicesResponse findServices(Predicate<Etcd3ServiceDefinition> filter) {
+ List<ServiceDefinition> servers = Collections.emptyList();
+ long revision = 0;
+
+ if (isRunAllowed()) {
+ try {
+ final GetResponse response = kvClient.get(
+ ByteSequence.from(toPathPrefix(servicePath), keyCharset),
+ GetOption.newBuilder().isPrefix(true).build()).get();
+
+ revision = response.getHeader().getRevision();
+ if (response.getCount() > 0) {
+ servers = response.getKvs().stream()
+ .map(KeyValue::getValue)
+ .filter(ObjectHelper::isNotEmpty)
+ .map(kv -> kv.toString(valueCharset))
+ .map(this::nodeFromString)
+ .filter(Objects::nonNull)
+ .filter(filter)
+ .sorted(Etcd3ServiceDefinition.COMPARATOR)
+ .collect(Collectors.toList());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeCamelException(e);
+ } catch (Exception e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+
+ return new Etcd3GetServicesResponse(revision, servers);
+ }
+
+ /**
+ * Unmarshalls the given value into a service definition.
+ *
+ * @param value the json payload representing a service definition.
+ * @return the corresponding service definition if it could be unmarshalled, {@code null} otherwise.
+ */
+ private Etcd3ServiceDefinition nodeFromString(String value) {
+ Etcd3ServiceDefinition server = null;
+ try {
+ server = MAPPER.readValue(value, Etcd3ServiceDefinition.class);
+ } catch (Exception e) {
+ LOGGER.warn("Could not parse the json payload {}", value, e);
+ }
+ return server;
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscoveryFactory.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscoveryFactory.java
new file mode 100644
index 00000000000..352256e9430
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3ServiceDiscoveryFactory.java
@@ -0,0 +1,285 @@
+/*
+ * 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.etcd3.cloud;
+
+import java.time.Duration;
+import java.util.Map;
+
+import io.netty.handler.ssl.SslContext;
+import org.apache.camel.CamelContext;
+import org.apache.camel.cloud.ServiceDiscovery;
+import org.apache.camel.cloud.ServiceDiscoveryFactory;
+import org.apache.camel.component.etcd3.Etcd3Configuration;
+import org.apache.camel.spi.Configurer;
+import org.apache.camel.spi.annotations.CloudServiceFactory;
+import org.apache.camel.util.ObjectHelper;
+
+@CloudServiceFactory("etcd-service-discovery")
+@Configurer
+public class Etcd3ServiceDiscoveryFactory implements ServiceDiscoveryFactory {
+ /**
+ * The configuration of the factory.
+ */
+ private final Etcd3Configuration configuration;
+ /**
+ * The type of the expected service discovery.
+ */
+ private String type;
+
+ public Etcd3ServiceDiscoveryFactory() {
+ this(new Etcd3Configuration());
+ }
+
+ public Etcd3ServiceDiscoveryFactory(Etcd3Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ // *************************************************************************
+ // Properties
+ // *************************************************************************
+
+ public String[] getEndpoints() {
+ return configuration.getEndpoints();
+ }
+
+ /**
+ * Configure etcd server endpoints using the IPNameResolver.
+ */
+ public void setEndpoints(String... endpoints) {
+ configuration.setEndpoints(endpoints);
+ }
+
+ public String getUserName() {
+ return configuration.getUserName();
+ }
+
+ /**
+ * Configure etcd auth user.
+ */
+ public void setUserName(String userName) {
+ configuration.setUserName(userName);
+ }
+
+ public String getPassword() {
+ return configuration.getPassword();
+ }
+
+ /**
+ * Configure etcd auth password.
+ */
+ public void setPassword(String password) {
+ configuration.setPassword(password);
+ }
+
+ public String getNamespace() {
+ return configuration.getNamespace();
+ }
+
+ /**
+ * Configure the namespace of keys used. "/" will be treated as no namespace.
+ */
+ public void setNamespace(String namespace) {
+ configuration.setNamespace(namespace);
+ }
+
+ public SslContext getSslContext() {
+ return configuration.getSslContext();
+ }
+
+ /**
+ * Configure SSL/TLS context to use instead of the system default.
+ */
+ public void setSslContext(SslContext sslContext) {
+ configuration.setSslContext(sslContext);
+ }
+
+ public String getLoadBalancerPolicy() {
+ return configuration.getLoadBalancerPolicy();
+ }
+
+ /**
+ * Configure etcd load balancer policy.
+ */
+ public void setLoadBalancerPolicy(String loadBalancerPolicy) {
+ configuration.setLoadBalancerPolicy(loadBalancerPolicy);
+ }
+
+ public String getAuthority() {
+ return configuration.getAuthority();
+ }
+
+ /**
+ * Configure the authority used to authenticate connections to servers.
+ */
+ public void setAuthority(String authority) {
+ configuration.setAuthority(authority);
+ }
+
+ public Integer getMaxInboundMessageSize() {
+ return configuration.getMaxInboundMessageSize();
+ }
+
+ /**
+ * Configure the maximum message size allowed for a single gRPC frame.
+ */
+ public void setMaxInboundMessageSize(Integer maxInboundMessageSize) {
+ configuration.setMaxInboundMessageSize(maxInboundMessageSize);
+ }
+
+ public Map<String, String> getHeaders() {
+ return configuration.getHeaders();
+ }
+
+ /**
+ * Configure the headers to be added to http request headers.
+ */
+ public void setHeaders(Map<String, String> headers) {
+ configuration.setHeaders(headers);
+ }
+
+ public Map<String, String> getAuthHeaders() {
+ return configuration.getAuthHeaders();
+ }
+
+ /**
+ * Configure the headers to be added to auth request headers.
+ */
+ public void setAuthHeaders(Map<String, String> authHeaders) {
+ configuration.setAuthHeaders(authHeaders);
+ }
+
+ public long getRetryDelay() {
+ return configuration.getRetryDelay();
+ }
+
+ /**
+ * Configure the delay between retries in milliseconds.
+ */
+ public void setRetryDelay(long retryDelay) {
+ configuration.setRetryDelay(retryDelay);
+ }
+
+ public long getRetryMaxDelay() {
+ return configuration.getRetryMaxDelay();
+ }
+
+ /**
+ * Configure the max backing off delay between retries in milliseconds.
+ */
+ public void setRetryMaxDelay(long retryMaxDelay) {
+ configuration.setRetryMaxDelay(retryMaxDelay);
+ }
+
+ public Duration getKeepAliveTime() {
+ return configuration.getKeepAliveTime();
+ }
+
+ /**
+ * Configure the interval for gRPC keepalives. The current minimum allowed by gRPC is 10 seconds.
+ */
+ public void setKeepAliveTime(Duration keepAliveTime) {
+ configuration.setKeepAliveTime(keepAliveTime);
+ }
+
+ public Duration getKeepAliveTimeout() {
+ return configuration.getKeepAliveTimeout();
+ }
+
+ /**
+ * Configure the timeout for gRPC keepalives.
+ */
+ public void setKeepAliveTimeout(Duration keepAliveTimeout) {
+ configuration.setKeepAliveTimeout(keepAliveTimeout);
+ }
+
+ public Duration getRetryMaxDuration() {
+ return configuration.getRetryMaxDuration();
+ }
+
+ /**
+ * Configure the retries max duration.
+ */
+ public void setRetryMaxDuration(Duration retryMaxDuration) {
+ this.configuration.setRetryMaxDuration(retryMaxDuration);
+ }
+
+ public Duration getConnectionTimeout() {
+ return configuration.getConnectionTimeout();
+ }
+
+ /**
+ * Configure the connection timeout.
+ */
+ public void setConnectionTimeout(Duration connectionTimeout) {
+ this.configuration.setConnectionTimeout(connectionTimeout);
+ }
+
+ public String getServicePath() {
+ return configuration.getServicePath();
+ }
+
+ /**
+ * The path to look for service discovery.
+ */
+ public void setServicePath(String servicePath) {
+ configuration.setServicePath(servicePath);
+ }
+
+ public String getKeyCharset() {
+ return configuration.getKeyCharset();
+ }
+
+ /**
+ * Configure the charset to use for the keys.
+ */
+ public void setKeyCharset(String keyCharset) {
+ configuration.setKeyCharset(keyCharset);
+ }
+
+ public String getValueCharset() {
+ return configuration.getValueCharset();
+ }
+
+ /**
+ * Configure the charset to use for the values.
+ */
+ public void setValueCharset(String valueCharset) {
+ configuration.setValueCharset(valueCharset);
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ /**
+ * Configure the type of service discovery. Possible values "watch" or any other.
+ */
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ // *************************************************************************
+ // Factory
+ // *************************************************************************
+
+ @Override
+ public ServiceDiscovery newInstance(CamelContext camelContext) throws Exception {
+ return ObjectHelper.equal("watch", type, true)
+ ? new Etcd3WatchServiceDiscovery(configuration)
+ : new Etcd3OnDemandServiceDiscovery(configuration);
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3WatchServiceDiscovery.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3WatchServiceDiscovery.java
new file mode 100644
index 00000000000..2376065f218
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/cloud/Etcd3WatchServiceDiscovery.java
@@ -0,0 +1,169 @@
+/*
+ * 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.etcd3.cloud;
+
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.Watch;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchResponse;
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.component.etcd3.Etcd3Configuration;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.etcd3.Etcd3Helper.toPathPrefix;
+
+/**
+ * An implementation of a {@link Etcd3ServiceDiscovery} that retrieves all the service definitions from etcd at first
+ * call, then refresh the list when a change has been detected.
+ */
+public class Etcd3WatchServiceDiscovery extends Etcd3ServiceDiscovery
+ implements Watch.Listener {
+
+ /**
+ * The logger.
+ */
+ private static final Logger LOGGER = LoggerFactory.getLogger(Etcd3WatchServiceDiscovery.class);
+
+ /**
+ * The current list of service definitions found.
+ */
+ private volatile List<ServiceDefinition> allServices;
+ /**
+ * The revision of the last update of the list of service definitions.
+ */
+ private final AtomicLong revision;
+ /**
+ * The path prefix of the key-value pairs containing the service definitions.
+ */
+ private final String servicePath;
+ /**
+ * The client to access to etcd.
+ */
+ private final Client client;
+ /**
+ * The client to watch key-value pairs stored into etcd corresponding to the service definitions.
+ */
+ private final Watch watch;
+ /**
+ * The charset to use for the keys.
+ */
+ private final Charset keyCharset;
+ /**
+ * The current watcher used to watch the changes of the service definitions.
+ */
+ private final AtomicReference<Watch.Watcher> watcher = new AtomicReference<>();
+ /**
+ * The mutex used to prevent concurrent load of the list of service definitions.
+ */
+ private final Object mutex = new Object();
+
+ /**
+ * Construct a {@code Etcd3WatchServiceDiscovery} with the given configuration.
+ *
+ * @param configuration the configuration used to set up the service discovery.
+ */
+ public Etcd3WatchServiceDiscovery(Etcd3Configuration configuration) {
+ super(configuration);
+ this.revision = new AtomicLong();
+ this.servicePath = ObjectHelper.notNull(configuration.getServicePath(), "servicePath");
+ this.client = configuration.createClient();
+ this.watch = client.getWatchClient();
+ this.keyCharset = Charset.forName(configuration.getKeyCharset());
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ try {
+ client.close();
+ } finally {
+ super.doStop();
+ }
+ }
+
+ @Override
+ public List<ServiceDefinition> getServices(String name) {
+ List<ServiceDefinition> servers = allServices;
+ if (servers == null) {
+ synchronized (mutex) {
+ servers = allServices;
+ if (servers == null) {
+ servers = reloadServices();
+ doWatch();
+ }
+ }
+ }
+
+ return servers.stream()
+ .filter(s -> name.equalsIgnoreCase(s.getName()))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Gets the list of service definitions from the etcd then update the current list.
+ *
+ * @return list of all service definitions that could be found.
+ */
+ private List<ServiceDefinition> reloadServices() {
+ Etcd3GetServicesResponse response = findServices();
+ revision.getAndUpdate(r -> Math.max(r, response.getRevision() + 1));
+ this.allServices = response.getServices();
+ return response.getServices();
+ }
+
+ /**
+ * If allowed, starts to watch the changes on the service definitions.
+ */
+ private void doWatch() {
+ if (!isRunAllowed()) {
+ return;
+ }
+ watcher.getAndUpdate(w -> {
+ if (w != null) {
+ w.close();
+ }
+ return watch.watch(
+ ByteSequence.from(toPathPrefix(servicePath), keyCharset),
+ WatchOption.newBuilder().isPrefix(true).withRevision(revision.get()).build(),
+ this);
+ });
+ }
+
+ @Override
+ public void onNext(WatchResponse response) {
+ // A change has been received let's reload the list
+ reloadServices();
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ LOGGER.debug("Cloud not fetch the index, key={}, cause={}", servicePath, throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ doWatch();
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicy.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicy.java
new file mode 100644
index 00000000000..0371604fdc8
--- /dev/null
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicy.java
@@ -0,0 +1,461 @@
+/*
+ * 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.etcd3.policy;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Txn;
+import io.etcd.jetcd.common.exception.ErrorCode;
+import io.etcd.jetcd.common.exception.EtcdException;
+import io.etcd.jetcd.kv.TxnResponse;
+import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
+import io.etcd.jetcd.op.Cmp;
+import io.etcd.jetcd.op.CmpTarget;
+import io.etcd.jetcd.op.Op;
+import io.etcd.jetcd.options.PutOption;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Route;
+import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.component.etcd3.Etcd3Configuration;
+import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.etcd3.Etcd3Constants.ETCD_DEFAULT_ENDPOINTS;
+
+/**
+ * An implementation of a route policy based on etcd.
+ */
+@ManagedResource(description = "Route policy using Etcd as clustered lock")
+public class Etcd3RoutePolicy extends RoutePolicySupport implements CamelContextAware {
+
+ /**
+ * The logger
+ */
+ private static final Logger LOGGER = LoggerFactory.getLogger(Etcd3RoutePolicy.class);
+ /**
+ * The mutex used to prevent concurrent access to {@code suspendedRoutes}.
+ */
+ private final Object mutex = new Object();
+ /**
+ * The flag indicating whether the current node is a leader.
+ */
+ private final AtomicBoolean leader = new AtomicBoolean();
+ /**
+ * The routes that have been suspended.
+ */
+ private final Set<Route> suspendedRoutes = new HashSet<>();
+ /**
+ * The time to live in seconds of a key-value pair inserted into etcd. Default value is {@code 60}.
+ */
+ private int ttl = 60;
+ /**
+ * The timeout in seconds of all requests. Default value is {@code 10}.
+ */
+ private int timeout = 10;
+ /**
+ * The route to which the policy is applied.
+ */
+ private volatile Route route;
+ /**
+ * The etcd service name.
+ */
+ private String serviceName;
+ /**
+ * The etcd service path.
+ */
+ private String servicePath;
+ /**
+ * The camel context associated to the policy.
+ */
+ private CamelContext camelContext;
+ /**
+ * The etcd endpoints.
+ */
+ private String[] endpoints;
+ /**
+ * The scheduler used to evaluate regularly the leadership.
+ */
+ private volatile ScheduledExecutorService executorService;
+ /**
+ * The flag indicating whether the consumer should be stopped.
+ */
+ private final AtomicBoolean shouldStopConsumer = new AtomicBoolean(true);
+ /**
+ * The id of the current lease. Only set if the current node is the leader.
+ */
+ private final AtomicLong leaseId = new AtomicLong();
+ /**
+ * The client to access to etcd.
+ */
+ private final AtomicReference<Client> client = new AtomicReference<>();
+ /**
+ * The client to access to the key-value pairs stored into etcd.
+ */
+ private final AtomicReference<KV> kv = new AtomicReference<>();
+ /**
+ * The client to access to the leases stored into etcd.
+ */
+ private final AtomicReference<Lease> lease = new AtomicReference<>();
+ /**
+ * The flag indicating whether the client has been created by the policy or outside the policy.
+ */
+ private final boolean managedClient;
+
+ public Etcd3RoutePolicy() {
+ this(ETCD_DEFAULT_ENDPOINTS);
+ }
+
+ public Etcd3RoutePolicy(Etcd3Configuration configuration) {
+ this(configuration.createClient(), true);
+ }
+
+ public Etcd3RoutePolicy(Client client) {
+ this(client, false);
+ }
+
+ private Etcd3RoutePolicy(Client client, boolean managedClient) {
+ this.client.set(ObjectHelper.notNull(client, "client"));
+ this.managedClient = managedClient;
+ }
+
+ public Etcd3RoutePolicy(String... endpoints) {
+ this.endpoints = endpoints;
+ this.managedClient = true;
+ }
+
+ @Override
+ public void onInit(Route route) {
+ super.onInit(route);
+ this.route = route;
+ if (executorService == null) {
+ executorService = ObjectHelper.notNull(camelContext, "camelContext", this)
+ .getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "Etcd3RoutePolicy[" + route.getRouteId() + "]");
+ }
+ }
+
+ @Override
+ public void onStart(Route route) {
+ if (!leader.get() && shouldStopConsumer.get()) {
+ stopConsumer(route);
+ }
+ }
+
+ @Override
+ public void onStop(Route route) {
+ synchronized (mutex) {
+ suspendedRoutes.remove(route);
+ }
+ }
+
+ @Override
+ public synchronized void onSuspend(Route route) {
+ synchronized (mutex) {
+ suspendedRoutes.remove(route);
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ Client c = client.get();
+ if (c == null) {
+ c = Client.builder().endpoints(ObjectHelper.notNull(endpoints, "endpoints")).build();
+ this.client.set(c);
+ }
+ lease.set(c.getLeaseClient());
+ kv.set(c.getKVClient());
+ evaluateLeadershipAndSchedule();
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (executorService != null) {
+ camelContext.getExecutorServiceManager().shutdownNow(executorService);
+ executorService = null;
+ }
+ try {
+ Client c = client.get();
+ if (managedClient && c != null) {
+ c.close();
+ }
+ } finally {
+ super.doStop();
+ }
+ }
+
+ // *************************************************************************
+ //
+ // *************************************************************************
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ /**
+ * Sets the latest leadership state of the current node and potentially triggers actions if the state has changed.
+ *
+ * @param isLeader {@code true} if the current node is the leader, {@code false} otherwise.
+ */
+ protected void setLeader(boolean isLeader) {
+ if (isLeader) {
+ if (leader.compareAndSet(false, true)) {
+ LOGGER.info("Leadership taken (path={}, name={})", servicePath, serviceName);
+ startAllStoppedConsumers();
+ }
+ } else if (leader.compareAndSet(true, false)) {
+ LOGGER.info("Leadership lost (path={}, name={})", servicePath, serviceName);
+ }
+ }
+
+ /**
+ * Stops the consumer of the given route.
+ *
+ * @param route the route for which the consumer should be stopped.
+ */
+ private void stopConsumer(Route route) {
+ synchronized (mutex) {
+ try {
+ if (!suspendedRoutes.contains(route)) {
+ LOGGER.debug("Stopping consumer for {} ({})", route.getId(), route.getConsumer());
+ stopConsumer(route.getConsumer());
+ suspendedRoutes.add(route);
+ }
+ } catch (Exception e) {
+ handleException(e);
+ }
+ }
+ }
+
+ /**
+ * Start all the consumers that have been stopped.
+ */
+ private void startAllStoppedConsumers() {
+ synchronized (mutex) {
+ try {
+ for (Route suspendedRoute : suspendedRoutes) {
+ LOGGER.debug("Starting consumer for {} ({})", suspendedRoute.getId(), suspendedRoute.getConsumer());
+ startConsumer(suspendedRoute.getConsumer());
+ }
+
+ suspendedRoutes.clear();
+ } catch (Exception e) {
+ handleException(e);
+ }
+ }
+ }
+
+ // *************************************************************************
+ // Getter/Setters
+ // *************************************************************************
+ public Client getClient() {
+ return client.get();
+ }
+
+ @ManagedAttribute(description = "The route id")
+ public String getRouteId() {
+ if (route != null) {
+ return route.getId();
+ }
+ return null;
+ }
+
+ @ManagedAttribute(description = "The consumer endpoint", mask = true)
+ public String getEndpointUrl() {
+ if (route != null && route.getConsumer() != null && route.getConsumer().getEndpoint() != null) {
+ return route.getConsumer().getEndpoint().toString();
+ }
+ return null;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ @ManagedAttribute(description = "The etcd service name")
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+
+ @ManagedAttribute(description = "The etcd service path")
+ public String getServicePath() {
+ return servicePath;
+ }
+
+ public void setServicePath(String servicePath) {
+ this.servicePath = servicePath;
+ }
+
+ @ManagedAttribute(description = "The time to live (seconds)")
+ public int getTtl() {
+ return ttl;
+ }
+
+ public void setTtl(int ttl) {
+ this.ttl = ttl;
+ }
+
+ @ManagedAttribute(description = "The request timeout (seconds)")
+ public int getTimeout() {
+ return timeout;
+ }
+
+ public void setTimeout(int timeout) {
+ this.timeout = timeout;
+ }
+
+ @ManagedAttribute(description = "Whether to stop consumer when starting up and failed to become master")
+ public boolean isShouldStopConsumer() {
+ return shouldStopConsumer.get();
+ }
+
+ public void setShouldStopConsumer(boolean shouldStopConsumer) {
+ this.shouldStopConsumer.set(shouldStopConsumer);
+ }
+
+ @ManagedAttribute(description = "Is this route the master or a slave")
+ public boolean isLeader() {
+ return leader.get();
+ }
+
+ @ManagedAttribute(description = "Etcd endpoints")
+ public String getEndpoints() {
+ return endpoints == null ? "" : String.join(",", endpoints);
+ }
+
+ public void setEndpoints(String[] endpoints) {
+ this.endpoints = endpoints;
+ }
+
+ public void setEndpoints(String endpoints) {
+ this.endpoints = endpoints.split(",");
+ }
+
+ long getLeaseId() {
+ return leaseId.get();
+ }
+
+ /**
+ * Evaluates the leadership and schedule the next evaluation.
+ */
+ private void evaluateLeadershipAndSchedule() {
+ evaluateLeadership();
+ executorService.schedule(this::evaluateLeadershipAndSchedule, Math.max(2 * ttl / 3, 1), TimeUnit.SECONDS);
+ }
+
+ /**
+ * In case the current node is the leader, it tries to renew the lease if it failed or the current node is a not
+ * leader, it tries to take the leadership.
+ */
+ private void evaluateLeadership() {
+ if (isLeader() && renewLease()) {
+ // The lease could be renewed successfully, so the status of the node did not change
+ return;
+ }
+ setLeader(tryTakeLeadership());
+ }
+
+ /**
+ * Renew the lease using a KeepAlive request to avoid losing the leadership.
+ *
+ * @return {@code true} if the lease could be renewed within the timeout, {@code false} otherwise.
+ */
+ private boolean renewLease() {
+ long id = leaseId.get();
+ if (id == 0) {
+ return false;
+ }
+ try {
+ LeaseKeepAliveResponse keepAliveResponse = lease.get().keepAliveOnce(id).get(timeout, TimeUnit.SECONDS);
+ LOGGER.debug("New TTL of the lease {} is {} seconds", id, keepAliveResponse.getTTL());
+ return true;
+ } catch (ExecutionException e) {
+ boolean notFound = false;
+ if (e.getCause() instanceof EtcdException) {
+ EtcdException ex = (EtcdException) e.getCause();
+ notFound = ex.getErrorCode() == ErrorCode.NOT_FOUND;
+ }
+ if (notFound) {
+ LOGGER.debug("The lease {} doesn't exist anymore", id);
+ leaseId.set(0);
+ } else {
+ LOGGER.debug("Could not renew the lease {}", id, e);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (TimeoutException e) {
+ LOGGER.debug("Timeout while trying to renew the lease {}", id);
+ }
+ return false;
+ }
+
+ /**
+ * Grants a new lease and tries to put a key-value pair corresponding to the {@code servicePath} if it is absent.
+ *
+ * @return {@code true} if the put operation is successful indicating that the leadership has been taken,
+ * {@code false} otherwise.
+ */
+ private boolean tryTakeLeadership() {
+ try {
+ long id = lease.get().grant(ttl, timeout, TimeUnit.SECONDS).get(timeout, TimeUnit.SECONDS).getID();
+ Txn transaction = kv.get().txn();
+ ByteSequence key = ByteSequence.from(servicePath.getBytes());
+ // Put if absent
+ TxnResponse response = transaction.If(new Cmp(key, Cmp.Op.EQUAL, CmpTarget.version(0)))
+ .Then(Op.put(key, ByteSequence.from(serviceName.getBytes()),
+ PutOption.newBuilder().withLeaseId(id).build()))
+ .commit()
+ .get(timeout, TimeUnit.SECONDS);
+
+ boolean succeeded = response.isSucceeded();
+ if (succeeded) {
+ leaseId.set(id);
+ } else {
+ lease.get().revoke(id);
+ }
+ return succeeded;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ LOGGER.debug("Could not try to take the leadership", e);
+ } catch (TimeoutException e) {
+ LOGGER.debug("Timeout while trying to take the leadership");
+ }
+ return false;
+ }
+}
diff --git a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/processor/aggregate/Etcd3AggregationRepository.java b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/processor/aggregate/Etcd3AggregationRepository.java
index 5c290e6bf77..79e865cecc3 100644
--- a/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/processor/aggregate/Etcd3AggregationRepository.java
+++ b/components/camel-etcd3/src/main/java/org/apache/camel/component/etcd3/processor/aggregate/Etcd3AggregationRepository.java
@@ -176,7 +176,7 @@ public class Etcd3AggregationRepository extends ServiceSupport
.If(new Cmp(
ByteSequence.from(String.format("%s/%s", prefixName, key).getBytes()),
Cmp.Op.EQUAL,
- CmpTarget.ModRevisionCmpTarget.modRevision(modRevision)))
+ CmpTarget.modRevision(modRevision)))
.Then(Op.put(ByteSequence
.from(String.format("%s/%s", prefixName, key).getBytes()), convertToEtcd3Format(newHolder),
PutOption.DEFAULT))
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/AggregateEtcd3ManualTest.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/AggregateEtcd3ManualTest.java
index 1182ed0ecea..5a14c95bbb1 100644
--- a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/AggregateEtcd3ManualTest.java
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/AggregateEtcd3ManualTest.java
@@ -18,9 +18,8 @@ package org.apache.camel.component.etcd3;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.etcd3.processor.aggregate.Etcd3AggregationRepository;
+import org.apache.camel.component.etcd3.support.Etcd3TestSupport;
import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
/**
@@ -30,15 +29,10 @@ import org.junit.jupiter.api.Test;
* of message 1,2 and 4 as they use the same correlation key.
* <p/>
*/
-@Disabled("Requires manually testing")
-public class AggregateEtcd3ManualTest extends CamelTestSupport {
-
- // TODO: use docker test-containers for testing
-
- private String endpoint = System.getProperty("endpoint"); //http://ip:port
+class AggregateEtcd3ManualTest extends Etcd3TestSupport {
@Test
- public void testABC() throws Exception {
+ void testABC() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedBodiesReceived("ABC");
template.sendBodyAndHeader("direct:start", "A", "myId", 1);
@@ -56,7 +50,7 @@ public class AggregateEtcd3ManualTest extends CamelTestSupport {
from("direct:start")
.log("Sending ${body} with correlation key ${header.myId}")
.aggregate(header("myId"), new MyAggregationStrategy())
- .aggregationRepository(new Etcd3AggregationRepository("aggregation", endpoint))
+ .aggregationRepository(new Etcd3AggregationRepository("aggregation", service.getServiceAddress()))
.completionSize(3)
.log("Sending out ${body}")
.to("mock:result");
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/cloud/integration/Etcd3ServiceDiscoveryIT.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/cloud/integration/Etcd3ServiceDiscoveryIT.java
new file mode 100644
index 00000000000..c615a3602e3
--- /dev/null
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/cloud/integration/Etcd3ServiceDiscoveryIT.java
@@ -0,0 +1,142 @@
+/*
+ * 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.etcd3.cloud.integration;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.common.exception.EtcdException;
+import io.etcd.jetcd.options.DeleteOption;
+import org.apache.camel.cloud.ServiceDefinition;
+import org.apache.camel.component.etcd3.Etcd3Configuration;
+import org.apache.camel.component.etcd3.Etcd3Helper;
+import org.apache.camel.component.etcd3.cloud.Etcd3OnDemandServiceDiscovery;
+import org.apache.camel.component.etcd3.cloud.Etcd3WatchServiceDiscovery;
+import org.apache.camel.component.etcd3.support.Etcd3TestSupport;
+import org.junit.jupiter.api.Test;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class Etcd3ServiceDiscoveryIT extends Etcd3TestSupport {
+ private static final ObjectMapper MAPPER = Etcd3Helper.createObjectMapper();
+ private static final AtomicInteger PORT = new AtomicInteger();
+
+ private Client client;
+ private Etcd3Configuration configuration;
+
+ @Override
+ public void doPreSetup() throws Exception {
+ configuration = new Etcd3Configuration();
+ configuration.setEndpoints(service.getServiceAddress());
+
+ client = getClient();
+ try {
+ client.getKVClient().delete(ByteSequence.from(configuration.getServicePath().getBytes()),
+ DeleteOption.newBuilder().isPrefix(true).build()).get();
+ } catch (EtcdException e) {
+ // Ignore
+ }
+ }
+
+ @Override
+ protected void cleanupResources() throws Exception {
+ try {
+ client.getKVClient().delete(ByteSequence.from(configuration.getServicePath().getBytes()),
+ DeleteOption.newBuilder().isPrefix(true).build()).get();
+ client.close();
+ client = null;
+ } catch (EtcdException e) {
+ // Ignore
+ }
+
+ super.cleanupResources();
+ }
+
+ @Test
+ void testOnDemandDiscovery() throws Exception {
+ for (int i = 0; i < 3; i++) {
+ addServer("serviceType-1");
+ }
+ for (int i = 0; i < 2; i++) {
+ addServer("serviceType-2");
+ }
+
+ try (Etcd3OnDemandServiceDiscovery strategy = new Etcd3OnDemandServiceDiscovery(configuration)) {
+ strategy.start();
+
+ List<ServiceDefinition> type1 = strategy.getServices("serviceType-1");
+ assertEquals(3, type1.size());
+ for (ServiceDefinition service : type1) {
+ assertNotNull(service.getMetadata());
+ assertTrue(service.getMetadata().containsKey("service_name"));
+ assertTrue(service.getMetadata().containsKey("port_delta"));
+ }
+
+ List<ServiceDefinition> type2 = strategy.getServices("serviceType-2");
+ assertEquals(2, type2.size());
+ for (ServiceDefinition service : type2) {
+ assertNotNull(service.getMetadata());
+ assertTrue(service.getMetadata().containsKey("service_name"));
+ assertTrue(service.getMetadata().containsKey("port_delta"));
+ }
+ }
+ }
+
+ @Test
+ void testWatchDiscovery() throws Exception {
+ addServer("serviceType-3");
+
+ try (Etcd3WatchServiceDiscovery strategy = new Etcd3WatchServiceDiscovery(configuration)) {
+ strategy.start();
+
+ assertEquals(1, strategy.getServices("serviceType-3").size());
+
+ addServer("serviceType-3");
+ addServer("serviceType-3");
+ addServer("serviceType-4");
+
+ await().atMost(1, SECONDS).untilAsserted(
+ () -> assertEquals(3, strategy.getServices("serviceType-3").size()));
+ }
+ }
+
+ private void addServer(String name) throws Exception {
+ int port = PORT.incrementAndGet();
+
+ Map<String, String> tags = new HashMap<>();
+ tags.put("service_name", name);
+ tags.put("port_delta", Integer.toString(port));
+
+ Map<String, Object> server = new HashMap<>();
+ server.put("name", name);
+ server.put("address", "127.0.0.1");
+ server.put("port", 8000 + port);
+ server.put("tags", tags);
+
+ client.getKVClient().put(ByteSequence.from((configuration.getServicePath() + "service-" + port).getBytes()),
+ ByteSequence.from(MAPPER.writeValueAsString(server).getBytes())).get();
+ }
+}
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/integration/Etcd3ConsumerIT.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/integration/Etcd3ConsumerIT.java
new file mode 100644
index 00000000000..5ef1a0c6bac
--- /dev/null
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/integration/Etcd3ConsumerIT.java
@@ -0,0 +1,149 @@
+/*
+ * 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.etcd3.integration;
+
+import java.nio.charset.StandardCharsets;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.watch.WatchEvent;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.etcd3.Etcd3Constants;
+import org.apache.camel.component.etcd3.support.Etcd3TestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+class Etcd3ConsumerIT extends Etcd3TestSupport {
+
+ private static final int BATCH_SIZE_PART_1 = 2_345;
+ private static final String ENCODING_TEST_KEY = "/éç";
+ private static final Processor NODE_TO_VALUE_IN_ISO_8859_1 = exchange -> {
+ WatchEvent event = exchange.getMessage().getBody(WatchEvent.class);
+ if (event != null) {
+ KeyValue keyValue = event.getKeyValue();
+ exchange.getMessage().setBody(keyValue.getKey().toString(StandardCharsets.ISO_8859_1) + "="
+ + keyValue.getValue().toString(StandardCharsets.ISO_8859_1));
+ }
+ };
+
+ @Test
+ void testWatchWithPath() throws Exception {
+ testWatch("mock:watch-with-path", "/myKey1", 10);
+ }
+
+ @Test
+ void testWatchWithConfigPath() throws Exception {
+ testWatch("mock:watch-with-config-path", "/myKey2", 11);
+ }
+
+ @Test
+ void testWatchPrefix() throws Exception {
+ testWatch("mock:watch-prefix", "/prefix/myKey1", 12);
+ }
+
+ @Test
+ void testWatchEncoding() throws Exception {
+ String value = "àè";
+ MockEndpoint mock = getMockEndpoint("mock:watch-encoding");
+ mock.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, ENCODING_TEST_KEY);
+ mock.expectedBodiesReceived(String.format("%s=%s", ENCODING_TEST_KEY, value));
+
+ final Client client = getClient();
+ client.getKVClient().put(
+ ByteSequence.from(ENCODING_TEST_KEY.getBytes(StandardCharsets.ISO_8859_1)),
+ ByteSequence.from(value.getBytes(StandardCharsets.ISO_8859_1))).get();
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testWatchRecovery() throws Exception {
+ final String key = "/myKeyRecovery";
+ final Client client = getClient();
+
+ final int batchSizePart2 = 13;
+ // Fill the vent backlog ( > 1000)
+ for (int i = 0; i < BATCH_SIZE_PART_1 + batchSizePart2; i++) {
+ client.getKVClient().put(ByteSequence.from(key.getBytes()), ByteSequence.from(("v" + i).getBytes())).get();
+ client.getKVClient().put(ByteSequence.from((key + "/" + i).getBytes()), ByteSequence.from("v".getBytes())).get();
+ }
+
+ final Object[] values = new String[batchSizePart2];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = key + "=v" + (i + BATCH_SIZE_PART_1);
+ }
+ MockEndpoint mock = getMockEndpoint("mock:watch-recovery");
+ mock.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, key);
+ mock.expectedBodiesReceived(values);
+ context().getRouteController().startRoute("watchRecovery");
+
+ mock.assertIsSatisfied();
+ }
+
+ private void testWatch(String mockEndpoint, final String key, int updates) throws Exception {
+ final Object[] values = new String[updates];
+ for (int i = 0; i < updates; i++) {
+ values[i] = key + "=myValue-" + i;
+ }
+
+ MockEndpoint mock = getMockEndpoint(mockEndpoint);
+ mock.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, key);
+ mock.expectedBodiesReceivedInAnyOrder(values);
+
+ final Client client = getClient();
+ for (int i = 0; i < updates; i++) {
+ client.getKVClient().put(ByteSequence.from(key.getBytes()), ByteSequence.from(("myValue-" + i).getBytes())).get();
+ }
+
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+ GetResponse response = getClient().getKVClient().get(
+ ByteSequence.from("/".getBytes()), GetOption.newBuilder().isPrefix(true).withCountOnly(true).build())
+ .get();
+ // To make sure that the recovery test won't be affected by other tests
+ long toSkip = response.getCount() == 0 ? 0L : response.getHeader().getRevision();
+ from("etcd3:myKey1")
+ .process(NODE_TO_VALUE_IN)
+ .to("mock:watch-with-path");
+ fromF("etcd3:myKeyRecovery?fromIndex=%s", toSkip + 2 * BATCH_SIZE_PART_1 + 1)
+ .id("watchRecovery")
+ .autoStartup(false)
+ .process(NODE_TO_VALUE_IN)
+ .to("mock:watch-recovery");
+ from("etcd3:prefix?prefix=true")
+ .process(NODE_TO_VALUE_IN)
+ .to("log:org.apache.camel.component.etcd?level=INFO")
+ .to("mock:watch-prefix");
+ from("etcd3:myKey2")
+ .process(NODE_TO_VALUE_IN)
+ .to("mock:watch-with-config-path");
+ fromF("etcd3:%s?keyCharset=ISO-8859-1", ENCODING_TEST_KEY)
+ .process(NODE_TO_VALUE_IN_ISO_8859_1)
+ .to("mock:watch-encoding");
+ }
+ };
+ }
+}
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/integration/Etcd3ProducerIT.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/integration/Etcd3ProducerIT.java
new file mode 100644
index 00000000000..d854ae088df
--- /dev/null
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/integration/Etcd3ProducerIT.java
@@ -0,0 +1,344 @@
+/*
+ * 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.etcd3.integration;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.kv.DeleteResponse;
+import io.etcd.jetcd.kv.GetResponse;
+import org.apache.camel.Exchange;
+import org.apache.camel.Predicate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.etcd3.Etcd3Constants;
+import org.apache.camel.component.etcd3.support.Etcd3TestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class Etcd3ProducerIT extends Etcd3TestSupport {
+
+ @Test
+ void testKeys() throws Exception {
+ final String path = "/camel/" + UUID.randomUUID();
+ final String value = UUID.randomUUID().toString();
+ final Client client = getClient();
+ final Map<String, Object> headers = new HashMap<>();
+
+ // *******************************************
+ // SET
+ // *******************************************
+
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_SET);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+
+ MockEndpoint mockSet = getMockEndpoint("mock:result-set");
+ mockSet.expectedMessageCount(1);
+ mockSet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ sendBody("direct:keys-set", value, headers);
+ mockSet.assertIsSatisfied();
+
+ // *******************************************
+ // GET
+ // *******************************************
+
+ headers.clear();
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_GET);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+
+ MockEndpoint mockGet = getMockEndpoint("mock:result-get");
+ mockGet.expectedMessageCount(1);
+ mockGet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ mockGet.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ GetResponse keysResponse = exchange.getIn().getBody(GetResponse.class);
+ assertNotNull(keysResponse);
+ assertEquals(1, keysResponse.getCount());
+ assertNotNull(keysResponse.getKvs());
+ assertFalse(keysResponse.getKvs().isEmpty());
+ ByteSequence actual = keysResponse.getKvs().get(0).getValue();
+ assertNotNull(actual);
+ return actual.toString().equals(value);
+ }
+ });
+
+ sendBody("direct:keys-get", value, headers);
+ mockGet.assertIsSatisfied();
+
+ // *******************************************
+ // DELETE
+ // *******************************************
+
+ headers.clear();
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_DELETE);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+
+ MockEndpoint mockDel = getMockEndpoint("mock:result-del");
+ mockDel.expectedMessageCount(1);
+ mockDel.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ sendBody("direct:keys-del", "value", headers);
+ mockDel.assertIsSatisfied();
+
+ // *******************************************
+ // VALIDATION
+ // *******************************************
+
+ GetResponse response = client.getKVClient().get(ByteSequence.from(path.getBytes())).get();
+ assertNotNull(response);
+ assertEquals(0, response.getCount());
+ }
+
+ @Test
+ void testEncoding() throws Exception {
+ final String path = "/âè/" + UUID.randomUUID();
+ final String value = "çà" + UUID.randomUUID();
+ final Map<String, Object> headers = new HashMap<>();
+
+ // *******************************************
+ // SET
+ // *******************************************
+
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_SET);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+ headers.put(Etcd3Constants.ETCD_KEY_CHARSET, "ISO-8859-1");
+ headers.put(Etcd3Constants.ETCD_VALUE_CHARSET, "ISO-8859-1");
+
+ MockEndpoint mockSet = getMockEndpoint("mock:result-set");
+ mockSet.expectedMessageCount(1);
+ mockSet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ sendBody("direct:keys-set", value, headers);
+ mockSet.assertIsSatisfied();
+
+ // *******************************************
+ // GET
+ // *******************************************
+
+ headers.clear();
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_GET);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+ headers.put(Etcd3Constants.ETCD_KEY_CHARSET, "ISO-8859-1");
+ headers.put(Etcd3Constants.ETCD_VALUE_CHARSET, "ISO-8859-1");
+
+ MockEndpoint mockGet = getMockEndpoint("mock:result-get");
+ mockGet.expectedMessageCount(1);
+ mockGet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ mockGet.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ GetResponse keysResponse = exchange.getIn().getBody(GetResponse.class);
+ assertNotNull(keysResponse);
+ assertEquals(1, keysResponse.getCount());
+ assertNotNull(keysResponse.getKvs());
+ assertFalse(keysResponse.getKvs().isEmpty());
+ ByteSequence actual = keysResponse.getKvs().get(0).getValue();
+ assertNotNull(actual);
+ return actual.toString(StandardCharsets.ISO_8859_1).equals(value);
+ }
+ });
+
+ sendBody("direct:keys-get", value, headers);
+ mockGet.assertIsSatisfied();
+
+ // *******************************************
+ // DELETE
+ // *******************************************
+
+ headers.clear();
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_DELETE);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+ headers.put(Etcd3Constants.ETCD_KEY_CHARSET, "ISO-8859-1");
+ headers.put(Etcd3Constants.ETCD_VALUE_CHARSET, "ISO-8859-1");
+
+ MockEndpoint mockDel = getMockEndpoint("mock:result-del");
+ mockDel.expectedMessageCount(1);
+ mockDel.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ sendBody("direct:keys-del", "value", headers);
+ mockDel.assertIsSatisfied();
+
+ // *******************************************
+ // VALIDATION
+ // *******************************************
+
+ GetResponse response = getClient().getKVClient().get(
+ ByteSequence.from(path, StandardCharsets.ISO_8859_1)).get();
+ assertNotNull(response);
+ assertEquals(0, response.getCount());
+ }
+
+ @Test
+ void testPrefix() throws Exception {
+ final String path = "/camel/" + UUID.randomUUID();
+ final String value = UUID.randomUUID().toString();
+ final Client client = getClient();
+ final Map<String, Object> headers = new HashMap<>();
+
+ // *******************************************
+ // SET
+ // *******************************************
+
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_SET);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+
+ MockEndpoint mockSet = getMockEndpoint("mock:result-set");
+ mockSet.expectedMessageCount(1);
+ mockSet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ sendBody("direct:keys-set", value, headers);
+ mockSet.assertIsSatisfied();
+
+ final String pathChild = String.format("%s/child", path);
+ headers.put(Etcd3Constants.ETCD_PATH, pathChild);
+
+ mockSet.reset();
+ mockSet.expectedMessageCount(1);
+ mockSet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, pathChild);
+ sendBody("direct:keys-set", value, headers);
+ mockSet.assertIsSatisfied();
+
+ final String pathSiblingWithSamePrefix = String.format("%s-sibling", path);
+ headers.put(Etcd3Constants.ETCD_PATH, pathSiblingWithSamePrefix);
+
+ mockSet.reset();
+ mockSet.expectedMessageCount(1);
+ mockSet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, pathSiblingWithSamePrefix);
+ sendBody("direct:keys-set", value, headers);
+ mockSet.assertIsSatisfied();
+
+ // *******************************************
+ // GET
+ // *******************************************
+
+ headers.clear();
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_GET);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+
+ MockEndpoint mockGet = getMockEndpoint("mock:result-get");
+ mockGet.expectedMessageCount(1);
+ mockGet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ mockGet.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ GetResponse keysResponse = exchange.getIn().getBody(GetResponse.class);
+ assertNotNull(keysResponse);
+ assertEquals(1, keysResponse.getCount());
+ assertNotNull(keysResponse.getKvs());
+ assertFalse(keysResponse.getKvs().isEmpty());
+ ByteSequence actual = keysResponse.getKvs().get(0).getValue();
+ assertNotNull(actual);
+ return actual.toString().equals(value);
+ }
+ });
+
+ sendBody("direct:keys-get", value, headers);
+ mockGet.assertIsSatisfied();
+
+ headers.put(Etcd3Constants.ETCD_IS_PREFIX, true);
+
+ mockGet.reset();
+ mockGet.expectedMessageCount(1);
+ mockGet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ mockGet.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ GetResponse response = exchange.getIn().getBody(GetResponse.class);
+ assertNotNull(response);
+ assertEquals(3, response.getCount());
+ assertNotNull(response.getKvs());
+ assertEquals(3, response.getKvs().size());
+ Set<String> keys = new HashSet<>(List.of(path, pathChild, pathSiblingWithSamePrefix));
+ for (KeyValue kv : response.getKvs()) {
+ ByteSequence actualValue = kv.getValue();
+ assertNotNull(actualValue);
+ assertEquals(value, actualValue.toString());
+ assertTrue(keys.remove(kv.getKey().toString()));
+ }
+ assertTrue(keys.isEmpty());
+ return true;
+ }
+ });
+
+ sendBody("direct:keys-get", value, headers);
+ mockGet.assertIsSatisfied();
+
+ // *******************************************
+ // DELETE
+ // *******************************************
+
+ headers.clear();
+ headers.put(Etcd3Constants.ETCD_ACTION, Etcd3Constants.ETCD_KEYS_ACTION_DELETE);
+ headers.put(Etcd3Constants.ETCD_PATH, path);
+ headers.put(Etcd3Constants.ETCD_IS_PREFIX, true);
+
+ MockEndpoint mockDel = getMockEndpoint("mock:result-del");
+ mockDel.expectedMessageCount(1);
+ mockGet.expectedHeaderReceived(Etcd3Constants.ETCD_PATH, path);
+ mockGet.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ DeleteResponse response = exchange.getIn().getBody(DeleteResponse.class);
+ assertNotNull(response);
+ assertEquals(3, response.getDeleted());
+ return true;
+ }
+ });
+ sendBody("direct:keys-del", "value", headers);
+ mockDel.assertIsSatisfied();
+
+ // *******************************************
+ // VALIDATION
+ // *******************************************
+
+ GetResponse response = client.getKVClient().get(ByteSequence.from(path.getBytes())).get();
+ assertNotNull(response);
+ assertEquals(0, response.getCount());
+ response = client.getKVClient().get(ByteSequence.from(pathChild.getBytes())).get();
+ assertNotNull(response);
+ assertEquals(0, response.getCount());
+ response = client.getKVClient().get(ByteSequence.from(pathSiblingWithSamePrefix.getBytes())).get();
+ assertNotNull(response);
+ assertEquals(0, response.getCount());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:keys-set")
+ .to("etcd3:dummy")
+ .to("mock:result-set");
+ from("direct:keys-get")
+ .to("etcd3:dummy")
+ .to("mock:result-get");
+ from("direct:keys-del")
+ .to("etcd3:dummy")
+ .to("mock:result-del");
+ }
+ };
+ }
+}
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicyIT.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicyIT.java
new file mode 100644
index 00000000000..9236887a98e
--- /dev/null
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicyIT.java
@@ -0,0 +1,101 @@
+/*
+ * 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.etcd3.policy;
+
+import io.etcd.jetcd.ByteSequence;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.etcd3.support.Etcd3TestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class Etcd3RoutePolicyIT extends Etcd3TestSupport {
+
+ private Etcd3RoutePolicy policyAsLeader;
+ private Etcd3RoutePolicy policyAsNotLeader;
+
+ @AfterEach
+ public void cleanUp() throws Exception {
+ if (policyAsLeader != null) {
+ policyAsLeader.doStop();
+ }
+ if (policyAsNotLeader != null) {
+ policyAsNotLeader.doStop();
+ }
+ }
+
+ @Test
+ void testAsLeader() throws Exception {
+ await().atMost(10, SECONDS).untilAsserted(() -> assertTrue(policyAsLeader.isLeader()));
+ MockEndpoint mock = getMockEndpoint("mock:leader");
+ mock.expectedBodiesReceived("ABC");
+ template.sendBody("direct:leader", "ABC");
+ assertMockEndpointsSatisfied();
+ getClient().getLeaseClient().revoke(policyAsLeader.getLeaseId());
+ getClient().getKVClient().put(
+ ByteSequence.from(policyAsLeader.getServicePath().getBytes()),
+ ByteSequence.from("not-leader".getBytes()))
+ .get();
+ await().atMost(10, SECONDS).untilAsserted(() -> assertFalse(policyAsLeader.isLeader()));
+ }
+
+ @Test
+ void testAsNotLeader() throws Exception {
+ await().atMost(10, SECONDS).untilAsserted(() -> assertFalse(policyAsNotLeader.isLeader()));
+ getClient().getKVClient().delete(
+ ByteSequence.from(policyAsNotLeader.getServicePath().getBytes())).get();
+ await().atMost(10, SECONDS).untilAsserted(() -> assertTrue(policyAsNotLeader.isLeader()));
+ MockEndpoint mock = getMockEndpoint("mock:not-leader");
+ mock.expectedBodiesReceived("DEF");
+ template.sendBody("direct:not-leader", "DEF");
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+ policyAsLeader = new Etcd3RoutePolicy(getClient());
+ policyAsLeader.setServicePath("/Etcd3RoutePolicyIT/leader");
+ policyAsLeader.setServiceName("leader");
+ policyAsLeader.setTtl(5);
+ from("direct:leader")
+ .routePolicy(policyAsLeader)
+ .id("as-leader")
+ .to("mock:leader");
+ policyAsNotLeader = new Etcd3RoutePolicy(getClient());
+ policyAsNotLeader.setServicePath("/Etcd3RoutePolicyIT/not-leader");
+ policyAsNotLeader.setServiceName("not-leader");
+ policyAsNotLeader.setTtl(5);
+ getClient().getKVClient().put(
+ ByteSequence.from(policyAsNotLeader.getServicePath().getBytes()),
+ ByteSequence.from("leader".getBytes()))
+ .get();
+ from("direct:not-leader")
+ .routePolicy(policyAsNotLeader)
+ .id("as-not-leader")
+ .to("mock:not-leader");
+
+ }
+ };
+ }
+}
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicyMain.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicyMain.java
new file mode 100644
index 00000000000..02a15bcfbf8
--- /dev/null
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/policy/Etcd3RoutePolicyMain.java
@@ -0,0 +1,49 @@
+/*
+ * 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.etcd3.policy;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.etcd3.Etcd3Constants;
+import org.apache.camel.main.Main;
+
+public final class Etcd3RoutePolicyMain {
+
+ private Etcd3RoutePolicyMain() {
+ }
+
+ public static void main(final String[] args) throws Exception {
+ Main main = new Main();
+ main.configure().addRoutesBuilder(new RouteBuilder() {
+ public void configure() {
+ Etcd3RoutePolicy policy = new Etcd3RoutePolicy();
+ policy.setEndpoints(Etcd3Constants.ETCD_DEFAULT_ENDPOINTS);
+ policy.setServicePath("/camel/services/leader");
+ policy.setServiceName(args[0]);
+ policy.setTtl(15);
+
+ from("file:///tmp/camel?delete=true")
+ .routeId(args[1])
+ .routePolicy(policy)
+ .setHeader("EtcdRouteID", constant(args[1]))
+ .setHeader("EtcdServiceName", constant(args[0]))
+ .to("log:org.apache.camel.component.etcd3?level=INFO&showAll=true");
+ }
+ });
+
+ main.run();
+ }
+}
diff --git a/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/support/Etcd3TestSupport.java b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/support/Etcd3TestSupport.java
new file mode 100644
index 00000000000..f2f0e647606
--- /dev/null
+++ b/components/camel-etcd3/src/test/java/org/apache/camel/component/etcd3/support/Etcd3TestSupport.java
@@ -0,0 +1,56 @@
+/*
+ * 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.etcd3.support;
+
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.watch.WatchEvent;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Processor;
+import org.apache.camel.component.etcd3.Etcd3Component;
+import org.apache.camel.test.infra.etcd3.services.Etcd3Service;
+import org.apache.camel.test.infra.etcd3.services.Etcd3ServiceFactory;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public abstract class Etcd3TestSupport extends CamelTestSupport {
+ @RegisterExtension
+ public static Etcd3Service service = Etcd3ServiceFactory.createService();
+
+ protected static final Processor NODE_TO_VALUE_IN = exchange -> {
+ WatchEvent event = exchange.getMessage().getBody(WatchEvent.class);
+ if (event != null) {
+ KeyValue keyValue = event.getKeyValue();
+ exchange.getMessage().setBody(keyValue.getKey() + "=" + keyValue.getValue());
+ }
+ };
+
+ protected Client getClient() {
+ return Client.builder().endpoints(service.getServiceAddress()).build();
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ Etcd3Component component = new Etcd3Component();
+ component.getConfiguration().setEndpoints(service.getServiceAddress());
+
+ CamelContext context = super.createCamelContext();
+ context.addComponent("etcd3", component);
+
+ return context;
+ }
+}
diff --git a/parent/pom.xml b/parent/pom.xml
index 3486913c3f1..7092768f4f5 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -298,8 +298,9 @@
<jcommander-version>1.72</jcommander-version>
<jcr-version>2.0</jcr-version>
<jedis-client-version>3.7.1</jedis-client-version>
- <jetcd-version>0.7.3</jetcd-version>
<jetcd-grpc-version>1.47.0</jetcd-grpc-version>
+ <jetcd-guava-version>31.1-jre</jetcd-guava-version>
+ <jetcd-version>0.7.3</jetcd-version>
<jetty9-version>9.4.48.v20220622</jetty9-version>
<jetty-version>${jetty9-version}</jetty-version>
<jetty-plugin-version>${jetty-version}</jetty-plugin-version>
diff --git a/test-infra/camel-test-infra-etcd3/pom.xml b/test-infra/camel-test-infra-etcd3/pom.xml
new file mode 100644
index 00000000000..9af33c2765c
--- /dev/null
+++ b/test-infra/camel-test-infra-etcd3/pom.xml
@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>camel-test-infra-parent</artifactId>
+ <groupId>org.apache.camel</groupId>
+ <relativePath>../camel-test-infra-parent/pom.xml</relativePath>
+ <version>3.20.0-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>camel-test-infra-etcd3</artifactId>
+ <name>Camel :: Test Infra :: EtcD v3</name>
+ <description>EtcD v3 test infrastructure for Camel</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.etcd</groupId>
+ <artifactId>jetcd-launcher</artifactId>
+ <version>${jetcd-version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+</project>
+
diff --git a/test-infra/camel-test-infra-etcd3/src/main/resources/META-INF/MANIFEST.MF b/test-infra/camel-test-infra-etcd3/src/main/resources/META-INF/MANIFEST.MF
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/common/Etcd3Properties.java b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/common/Etcd3Properties.java
new file mode 100644
index 00000000000..e221ab757af
--- /dev/null
+++ b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/common/Etcd3Properties.java
@@ -0,0 +1,26 @@
+/*
+ * 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.test.infra.etcd3.common;
+
+public final class Etcd3Properties {
+ public static final String SERVICE_ADDRESS = "etcd.service.address";
+ public static final String ETCD_CONTAINER = "etcd.container";
+
+ private Etcd3Properties() {
+
+ }
+}
diff --git a/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3LocalContainerService.java b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3LocalContainerService.java
new file mode 100644
index 00000000000..98d63fcb61f
--- /dev/null
+++ b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3LocalContainerService.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.etcd3.services;
+
+import java.util.List;
+import java.util.UUID;
+
+import io.etcd.jetcd.launcher.EtcdContainer;
+import org.apache.camel.test.infra.common.services.ContainerService;
+import org.apache.camel.test.infra.etcd3.common.Etcd3Properties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Etcd3LocalContainerService implements Etcd3Service, ContainerService<EtcdContainer> {
+ public static final String CONTAINER_IMAGE = "gcr.io/etcd-development/etcd:v3.5.5";
+ public static final String CONTAINER_NAME = "etcd";
+ public static final int ETCD_CLIENT_PORT = 2379;
+ public static final int ETCD_PEER_PORT = 2380;
+
+ private static final Logger LOG = LoggerFactory.getLogger(Etcd3LocalContainerService.class);
+
+ private final EtcdContainer container;
+
+ public Etcd3LocalContainerService() {
+ this(System.getProperty("etcd.container", CONTAINER_IMAGE));
+ }
+
+ public Etcd3LocalContainerService(String imageName) {
+ container = initContainer(imageName, CONTAINER_NAME);
+ }
+
+ public Etcd3LocalContainerService(EtcdContainer container) {
+ this.container = container;
+ }
+
+ public EtcdContainer initContainer(String imageName, String containerName) {
+ return new EtcdContainer(imageName, CONTAINER_NAME, List.of(CONTAINER_NAME))
+ .withNetworkAliases(containerName)
+ .withClusterToken(UUID.randomUUID().toString())
+ .withExposedPorts(ETCD_CLIENT_PORT, ETCD_PEER_PORT);
+ }
+
+ @Override
+ public void registerProperties() {
+ System.setProperty(Etcd3Properties.SERVICE_ADDRESS, getServiceAddress());
+ }
+
+ @Override
+ public void initialize() {
+ LOG.info("Trying to start the Etcd container");
+ container.start();
+
+ registerProperties();
+ LOG.info("Etcd instance running at {}", getServiceAddress());
+ }
+
+ @Override
+ public void shutdown() {
+ LOG.info("Stopping the Etcd container");
+ container.stop();
+ }
+
+ @Override
+ public EtcdContainer getContainer() {
+ return container;
+ }
+
+ @Override
+ public String getServiceAddress() {
+ return String.format("http://%s:%d", container.getHost(), container.getMappedPort(ETCD_CLIENT_PORT));
+ }
+}
diff --git a/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3RemoteService.java b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3RemoteService.java
new file mode 100644
index 00000000000..7680f91f11c
--- /dev/null
+++ b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3RemoteService.java
@@ -0,0 +1,42 @@
+/*
+ * 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.test.infra.etcd3.services;
+
+import org.apache.camel.test.infra.etcd3.common.Etcd3Properties;
+
+public class Etcd3RemoteService implements Etcd3Service {
+
+ @Override
+ public void registerProperties() {
+ // NO-OP
+ }
+
+ @Override
+ public void initialize() {
+ registerProperties();
+ }
+
+ @Override
+ public void shutdown() {
+ // NO-OP
+ }
+
+ @Override
+ public String getServiceAddress() {
+ return System.getProperty(Etcd3Properties.SERVICE_ADDRESS);
+ }
+}
diff --git a/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3Service.java b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3Service.java
new file mode 100644
index 00000000000..6db74e51ca4
--- /dev/null
+++ b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3Service.java
@@ -0,0 +1,40 @@
+/*
+ * 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.test.infra.etcd3.services;
+
+import org.apache.camel.test.infra.common.services.TestService;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+/**
+ * Test infra service for EtcD
+ */
+public interface Etcd3Service extends BeforeAllCallback, AfterAllCallback, TestService {
+
+ String getServiceAddress();
+
+ @Override
+ default void beforeAll(ExtensionContext extensionContext) throws Exception {
+ initialize();
+ }
+
+ @Override
+ default void afterAll(ExtensionContext extensionContext) throws Exception {
+ shutdown();
+ }
+}
diff --git a/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3ServiceFactory.java b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3ServiceFactory.java
new file mode 100644
index 00000000000..73e34b04085
--- /dev/null
+++ b/test-infra/camel-test-infra-etcd3/src/test/java/org/apache/camel/test/infra/etcd3/services/Etcd3ServiceFactory.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.etcd3.services;
+
+import org.apache.camel.test.infra.common.services.SimpleTestServiceBuilder;
+
+public final class Etcd3ServiceFactory {
+
+ private Etcd3ServiceFactory() {
+
+ }
+
+ public static SimpleTestServiceBuilder<Etcd3Service> builder() {
+ return new SimpleTestServiceBuilder<>("etcd");
+ }
+
+ public static Etcd3Service createService() {
+ return builder()
+ .addLocalMapping(Etcd3LocalContainerService::new)
+ .addRemoteMapping(Etcd3RemoteService::new)
+ .build();
+ }
+}
diff --git a/test-infra/pom.xml b/test-infra/pom.xml
index 42d645aad52..1b90299b5ee 100644
--- a/test-infra/pom.xml
+++ b/test-infra/pom.xml
@@ -75,5 +75,6 @@
<module>camel-test-infra-ignite</module>
<module>camel-test-infra-hashicorp-vault</module>
<module>camel-test-infra-jetty</module>
+ <module>camel-test-infra-etcd3</module>
</modules>
</project>