You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by lb...@apache.org on 2017/09/25 12:39:06 UTC
[8/9] camel git commit: Refine Cluster Service API and implementations
Refine Cluster Service API and implementations
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/8ef99f5c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/8ef99f5c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/8ef99f5c
Branch: refs/heads/master
Commit: 8ef99f5cffb268e4fd838b66f5dbcd4cc841c776
Parents: 24cb543
Author: lburgazzoli <lb...@gmail.com>
Authored: Sat Sep 23 17:40:57 2017 +0200
Committer: lburgazzoli <lb...@gmail.com>
Committed: Mon Sep 25 14:21:01 2017 +0200
----------------------------------------------------------------------
.../camel/ha/CamelClusterEventListener.java | 6 +-
.../org/apache/camel/ha/CamelClusterMember.java | 5 ++
.../camel/impl/ha/AbstractCamelClusterView.java | 3 +-
.../camel/impl/ha/ClusteredRoutePolicy.java | 3 +-
.../atomix/ha/AtomixClusterService.java | 4 +-
.../component/atomix/ha/AtomixClusterView.java | 35 ++++++++---
.../ha/AtomixClientRoutePolicyTestSupport.java | 18 +++---
.../atomix/ha/AtomixRoutePolicyTest.java | 20 +++----
.../src/test/resources/log4j2.properties | 4 ++
components/camel-consul/pom.xml | 2 +-
.../component/consul/ha/ConsulClusterView.java | 37 ++++++++----
.../consul/ha/ConsulClusteredRoutePolicyIT.java | 18 +++---
.../src/test/resources/log4j2.properties | 16 ++++-
.../kubernetes/ha/KubernetesClusterView.java | 8 ++-
.../ha/lock/KubernetesLockConfiguration.java | 2 +-
.../kubernetes/ha/utils/LeaderRecorder.java | 5 +-
.../zookeeper/ha/ZooKeeperClusterView.java | 61 ++++++++++++--------
.../ha/ZooKeeperClusteredRoutePolicyTest.java | 18 +++---
.../src/test/resources/log4j2.properties | 12 ++--
.../AtomixClusterServiceAutoConfiguration.java | 8 +--
.../AtomixClusterServiceConfiguration.java | 2 +-
.../ConsulClusterServiceAutoConfiguration.java | 2 +-
.../ha/ConsulClusterServiceConfiguration.java | 2 +-
...ooKeeperClusterServiceAutoConfiguration.java | 2 +-
.../ZooKeeperClusterServiceConfiguration.java | 2 +-
25 files changed, 188 insertions(+), 107 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/camel-core/src/main/java/org/apache/camel/ha/CamelClusterEventListener.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/ha/CamelClusterEventListener.java b/camel-core/src/main/java/org/apache/camel/ha/CamelClusterEventListener.java
index 1a972ca..8fcd4d6 100644
--- a/camel-core/src/main/java/org/apache/camel/ha/CamelClusterEventListener.java
+++ b/camel-core/src/main/java/org/apache/camel/ha/CamelClusterEventListener.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.ha;
+import java.util.Optional;
+
/**
* Marker interface for cluster events
*/
@@ -27,9 +29,9 @@ public interface CamelClusterEventListener {
* Notify a change in the leadership for a particular cluster.
*
* @param view the cluster view
- * @param leader the new leader or null (when there are no active leaders)
+ * @param leader the optional new leader
*/
- void leadershipChanged(CamelClusterView view, CamelClusterMember leader);
+ void leadershipChanged(CamelClusterView view, Optional<CamelClusterMember> leader);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/camel-core/src/main/java/org/apache/camel/ha/CamelClusterMember.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/ha/CamelClusterMember.java b/camel-core/src/main/java/org/apache/camel/ha/CamelClusterMember.java
index 285c7aa..951ea9e 100644
--- a/camel-core/src/main/java/org/apache/camel/ha/CamelClusterMember.java
+++ b/camel-core/src/main/java/org/apache/camel/ha/CamelClusterMember.java
@@ -23,4 +23,9 @@ public interface CamelClusterMember extends HasId {
* @return true if this member is the master.
*/
boolean isLeader();
+
+ /**
+ * @return true if this member is local.
+ */
+ boolean isLocal();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/camel-core/src/main/java/org/apache/camel/impl/ha/AbstractCamelClusterView.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ha/AbstractCamelClusterView.java b/camel-core/src/main/java/org/apache/camel/impl/ha/AbstractCamelClusterView.java
index 1605c6d..8a060c6 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ha/AbstractCamelClusterView.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ha/AbstractCamelClusterView.java
@@ -18,6 +18,7 @@ package org.apache.camel.impl.ha;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.locks.StampedLock;
import java.util.function.Consumer;
@@ -92,7 +93,7 @@ public abstract class AbstractCamelClusterView extends ServiceSupport implements
);
}
- protected void fireLeadershipChangedEvent(CamelClusterMember leader) {
+ protected void fireLeadershipChangedEvent(Optional<CamelClusterMember> leader) {
doWithListener(
CamelClusterEventListener.Leadership.class,
listener -> listener.leadershipChanged(this, leader)
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/camel-core/src/main/java/org/apache/camel/impl/ha/ClusteredRoutePolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ha/ClusteredRoutePolicy.java b/camel-core/src/main/java/org/apache/camel/impl/ha/ClusteredRoutePolicy.java
index 26cee62..8e923d1 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ha/ClusteredRoutePolicy.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ha/ClusteredRoutePolicy.java
@@ -19,6 +19,7 @@ package org.apache.camel.impl.ha;
import java.time.Duration;
import java.util.EventObject;
import java.util.HashSet;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -262,7 +263,7 @@ public class ClusteredRoutePolicy extends RoutePolicySupport implements CamelCon
private class CamelClusterLeadershipListener implements CamelClusterEventListener.Leadership {
@Override
- public void leadershipChanged(CamelClusterView view, CamelClusterMember leader) {
+ public void leadershipChanged(CamelClusterView view, Optional<CamelClusterMember> leader) {
setLeader(clusterView.getLocalMember().isLeader());
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java
index b2315ae..1937a82 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java
@@ -147,8 +147,8 @@ public final class AtomixClusterService extends AbstractCamelClusterService<Atom
super.doStop();
if (atomix != null) {
- LOGGER.debug("Shutdown atomix replica {}", atomix);
- atomix.shutdown().join();
+ LOGGER.debug("Leaving atomix cluster replica {}", atomix);
+ atomix.leave().join();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java
index be24d64..91367e9 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java
@@ -95,21 +95,21 @@ final class AtomixClusterView extends AbstractCamelClusterView {
LOGGER.debug("Listen election events");
group.election().onElection(term -> {
if (isRunAllowed()) {
- fireLeadershipChangedEvent(new AtomixClusterMember(term.leader()));
+ fireLeadershipChangedEvent(Optional.of(toClusterMember(term.leader())));
}
});
LOGGER.debug("Listen join events");
group.onJoin(member -> {
if (isRunAllowed()) {
- fireMemberAddedEvent(new AtomixClusterMember(member));
+ fireMemberAddedEvent(toClusterMember(member));
}
});
LOGGER.debug("Listen leave events");
group.onLeave(member -> {
if (isRunAllowed()) {
- fireMemberRemovedEvent(new AtomixClusterMember(member));
+ fireMemberRemovedEvent(toClusterMember(member));
}
});
@@ -123,11 +123,17 @@ final class AtomixClusterView extends AbstractCamelClusterView {
localMember.leave();
}
+ protected CamelClusterMember toClusterMember(GroupMember member) {
+ return localMember != null && localMember.is(member)
+ ? localMember
+ : new AtomixClusterMember(member);
+ }
+
// ***********************************************
//
// ***********************************************
- class AtomixLocalMember implements CamelClusterMember {
+ final class AtomixLocalMember implements CamelClusterMember {
private LocalMember member;
@Override
@@ -153,6 +159,17 @@ final class AtomixClusterView extends AbstractCamelClusterView {
return member.equals(group.election().term().leader());
}
+ @Override
+ public boolean isLocal() {
+ return true;
+ }
+
+ boolean is(GroupMember member) {
+ return this.member != null
+ ? this.member.equals(member)
+ : false;
+ }
+
boolean hasJoined() {
return member != null;
}
@@ -183,8 +200,7 @@ final class AtomixClusterView extends AbstractCamelClusterView {
group.remove(id).join();
member = null;
-
- fireLeadershipChangedEvent(null);
+ fireLeadershipChangedEvent(Optional.empty());
}
return this;
@@ -199,7 +215,7 @@ final class AtomixClusterView extends AbstractCamelClusterView {
}
}
- class AtomixClusterMember implements CamelClusterMember {
+ final class AtomixClusterMember implements CamelClusterMember {
private final GroupMember member;
AtomixClusterMember(GroupMember member) {
@@ -224,6 +240,11 @@ final class AtomixClusterView extends AbstractCamelClusterView {
}
@Override
+ public boolean isLocal() {
+ return localMember != null ? localMember.is(member) : false;
+ }
+
+ @Override
public String toString() {
final StringBuilder sb = new StringBuilder("AtomixClusterMember{");
sb.append("group=").append(group);
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixClientRoutePolicyTestSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixClientRoutePolicyTestSupport.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixClientRoutePolicyTestSupport.java
index bf425df..d5b0f8b 100644
--- a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixClientRoutePolicyTestSupport.java
+++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixClientRoutePolicyTestSupport.java
@@ -81,7 +81,8 @@ public abstract class AtomixClientRoutePolicyTestSupport {
private void run(String id) {
try {
- CountDownLatch contextLatch = new CountDownLatch(1);
+ int events = ThreadLocalRandom.current().nextInt(2, 6);
+ CountDownLatch contextLatch = new CountDownLatch(events);
DefaultCamelContext context = new DefaultCamelContext();
context.disableJMX();
@@ -91,15 +92,10 @@ public abstract class AtomixClientRoutePolicyTestSupport {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("timer:atomix?delay=1s&period=1s&repeatCount=1")
+ from("timer:atomix?delay=1s&period=1s")
.routeId("route-" + id)
- .process(e -> {
- LOGGER.debug("Node {} done", id);
- results.add(id);
- // Shutdown the context later on to give a chance to
- // other members to catch-up
- scheduler.schedule(contextLatch::countDown, 2 + ThreadLocalRandom.current().nextInt(3), TimeUnit.SECONDS);
- });
+ .log("From ${routeId}")
+ .process(e -> contextLatch.countDown());
}
});
@@ -109,6 +105,10 @@ public abstract class AtomixClientRoutePolicyTestSupport {
context.start();
contextLatch.await();
+
+ LOGGER.debug("Shutting down client node {}", id);
+ results.add(id);
+
context.stop();
latch.countDown();
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixRoutePolicyTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixRoutePolicyTest.java
index 8e23d5d..e730291 100644
--- a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixRoutePolicyTest.java
+++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixRoutePolicyTest.java
@@ -47,7 +47,7 @@ public final class AtomixRoutePolicyTest {
);
private final Set<Address> results = new HashSet<>();
- private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(addresses.size() * 2);
+ private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(addresses.size());
private final CountDownLatch latch = new CountDownLatch(addresses.size());
// ************************************
@@ -73,7 +73,8 @@ public final class AtomixRoutePolicyTest {
private void run(Address address) {
try {
- CountDownLatch contextLatch = new CountDownLatch(1);
+ int events = ThreadLocalRandom.current().nextInt(2, 6);
+ CountDownLatch contextLatch = new CountDownLatch(events);
AtomixClusterService service = new AtomixClusterService();
service.setId("node-" + address.port());
@@ -89,15 +90,10 @@ public final class AtomixRoutePolicyTest {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("timer:atomix?delay=1s&period=1s&repeatCount=1")
+ from("timer:atomix?delay=1s&period=1s")
.routeId("route-" + address.port())
- .process(e -> {
- LOGGER.debug("Node {} done", address);
- results.add(address);
- // Shutdown the context later on to give a chance to
- // other members to catch-up
- scheduler.schedule(contextLatch::countDown, 2 + ThreadLocalRandom.current().nextInt(3), TimeUnit.SECONDS);
- });
+ .log("From ${routeId}")
+ .process(e -> contextLatch.countDown());
}
});
@@ -107,6 +103,10 @@ public final class AtomixRoutePolicyTest {
context.start();
contextLatch.await();
+
+ LOGGER.debug("Shutting down node {}", address);
+ results.add(address);
+
context.stop();
latch.countDown();
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-atomix/src/test/resources/log4j2.properties
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/test/resources/log4j2.properties b/components/camel-atomix/src/test/resources/log4j2.properties
index d424d7e..e11e374 100644
--- a/components/camel-atomix/src/test/resources/log4j2.properties
+++ b/components/camel-atomix/src/test/resources/log4j2.properties
@@ -27,6 +27,10 @@ appender.out.layout.pattern = [%30.30t] %-30.30c{1} %-5p %m%n
logger.atomix.name = io.atomix
logger.atomix.level = INFO
+logger.atomix-copycat.name = io.atomix.copycat
+logger.atomix-copycat.level = WARN
+logger.atomix-catalyst.name = io.atomix.catalyst
+logger.atomix-catalyst.level = WARN
logger.camel.name = org.apache.camel
logger.camel.level = INFO
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-consul/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-consul/pom.xml b/components/camel-consul/pom.xml
index 914e751..8133b47 100644
--- a/components/camel-consul/pom.xml
+++ b/components/camel-consul/pom.xml
@@ -163,7 +163,7 @@
<images>
<image>
<name>consul:latest</name>
- <alias>consul</alias>
+ <alias>0.9.3</alias>
<run>
<ports>
<port>consul.port:8500</port>
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-consul/src/main/java/org/apache/camel/component/consul/ha/ConsulClusterView.java
----------------------------------------------------------------------
diff --git a/components/camel-consul/src/main/java/org/apache/camel/component/consul/ha/ConsulClusterView.java b/components/camel-consul/src/main/java/org/apache/camel/component/consul/ha/ConsulClusterView.java
index ae84168..c5971461 100644
--- a/components/camel-consul/src/main/java/org/apache/camel/component/consul/ha/ConsulClusterView.java
+++ b/components/camel-consul/src/main/java/org/apache/camel/component/consul/ha/ConsulClusterView.java
@@ -35,6 +35,7 @@ import com.orbitz.consul.model.session.SessionInfo;
import com.orbitz.consul.option.QueryOptions;
import org.apache.camel.ha.CamelClusterMember;
import org.apache.camel.impl.ha.AbstractCamelClusterView;
+import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,7 +44,6 @@ final class ConsulClusterView extends AbstractCamelClusterView {
private final ConsulClusterConfiguration configuration;
private final ConsulLocalMember localMember;
- private final ConsulClusterMember nullMember;
private final AtomicReference<String> sessionId;
private final Watcher watcher;
@@ -57,7 +57,6 @@ final class ConsulClusterView extends AbstractCamelClusterView {
this.configuration = configuration;
this.localMember = new ConsulLocalMember();
- this.nullMember = new ConsulClusterMember();
this.sessionId = new AtomicReference<>();
this.watcher = new Watcher();
this.path = configuration.getRootPath() + "/" + namespace;
@@ -126,17 +125,21 @@ final class ConsulClusterView extends AbstractCamelClusterView {
LOGGER.debug("Successfully released lock on path '{}' with id '{}'", path, sessionId.get());
}
- sessionClient.destroySession(sessionId.getAndSet(null));
- localMember.setMaster(false);
+ synchronized (sessionId) {
+ sessionClient.destroySession(sessionId.getAndSet(null));
+ localMember.setMaster(false);
+ }
}
}
private boolean acquireLock() {
- String sid = sessionId.get();
+ synchronized (sessionId) {
+ String sid = sessionId.get();
- return (sid != null)
- ? keyValueClient.acquireLock(this.path, sid)
- : false;
+ return (sid != null)
+ ? sessionClient.getSessionInfo(sid).transform(si -> keyValueClient.acquireLock(path, sid)).or(Boolean.FALSE)
+ : false;
+ }
}
// ***********************************************
@@ -149,12 +152,12 @@ final class ConsulClusterView extends AbstractCamelClusterView {
void setMaster(boolean master) {
if (master && this.master.compareAndSet(false, true)) {
LOGGER.debug("Leadership taken for session id {}", sessionId.get());
- fireLeadershipChangedEvent(this);
+ fireLeadershipChangedEvent(Optional.of(this));
return;
}
if (!master && this.master.compareAndSet(true, false)) {
LOGGER.debug("Leadership lost for session id {}", sessionId.get());
- fireLeadershipChangedEvent(getMaster().orElse(nullMember));
+ fireLeadershipChangedEvent(getMaster());
return;
}
}
@@ -165,6 +168,11 @@ final class ConsulClusterView extends AbstractCamelClusterView {
}
@Override
+ public boolean isLocal() {
+ return true;
+ }
+
+ @Override
public String getId() {
return sessionId.get();
}
@@ -210,6 +218,15 @@ final class ConsulClusterView extends AbstractCamelClusterView {
}
@Override
+ public boolean isLocal() {
+ if (id == null) {
+ return false;
+ }
+
+ return ObjectHelper.equal(id, localMember.getId());
+ }
+
+ @Override
public String toString() {
return "ConsulClusterMember{"
+ "id='" + id + '\''
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-consul/src/test/java/org/apache/camel/component/consul/ha/ConsulClusteredRoutePolicyIT.java
----------------------------------------------------------------------
diff --git a/components/camel-consul/src/test/java/org/apache/camel/component/consul/ha/ConsulClusteredRoutePolicyIT.java b/components/camel-consul/src/test/java/org/apache/camel/component/consul/ha/ConsulClusteredRoutePolicyIT.java
index effa2f3..2a79682 100644
--- a/components/camel-consul/src/test/java/org/apache/camel/component/consul/ha/ConsulClusteredRoutePolicyIT.java
+++ b/components/camel-consul/src/test/java/org/apache/camel/component/consul/ha/ConsulClusteredRoutePolicyIT.java
@@ -67,7 +67,8 @@ public class ConsulClusteredRoutePolicyIT {
private static void run(String id) {
try {
- CountDownLatch contextLatch = new CountDownLatch(1);
+ int events = ThreadLocalRandom.current().nextInt(2, 6);
+ CountDownLatch contextLatch = new CountDownLatch(events);
ConsulClusterService service = new ConsulClusterService();
service.setId("node-" + id);
@@ -83,15 +84,10 @@ public class ConsulClusteredRoutePolicyIT {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("timer:consul?delay=1s&period=1s&repeatCount=1")
+ from("timer:consul?delay=1s&period=1s")
.routeId("route-" + id)
- .process(e -> {
- LOGGER.debug("Node {} done", id);
- RESULTS.add(id);
- // Shutdown the context later on to give a chance to
- // other members to catch-up
- SCHEDULER.schedule(contextLatch::countDown, 2 + ThreadLocalRandom.current().nextInt(3), TimeUnit.SECONDS);
- });
+ .log("From ${routeId}")
+ .process(e -> contextLatch.countDown());
}
});
@@ -101,6 +97,10 @@ public class ConsulClusteredRoutePolicyIT {
context.start();
contextLatch.await();
+
+ LOGGER.debug("Shutting down node {}", id);
+ RESULTS.add(id);
+
context.stop();
LATCH.countDown();
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-consul/src/test/resources/log4j2.properties
----------------------------------------------------------------------
diff --git a/components/camel-consul/src/test/resources/log4j2.properties b/components/camel-consul/src/test/resources/log4j2.properties
index c2510bc..a4da2bf 100644
--- a/components/camel-consul/src/test/resources/log4j2.properties
+++ b/components/camel-consul/src/test/resources/log4j2.properties
@@ -24,9 +24,18 @@ appender.out.type = Console
appender.out.name = out
appender.out.layout.type = PatternLayout
appender.out.layout.pattern = [%30.30t] %-30.30c{1} %-5p %m%n
-logger.consul.name = org.apache.camel.component.consul
-logger.consul.level = DEBUG
+logger.atomix.name = io.atomix
+logger.atomix.level = INFO
+logger.atomix-copycat.name = io.atomix.copycat
+logger.atomix-copycat.level = WARN
+logger.atomix-catalyst.name = io.atomix.catalyst
+logger.atomix-catalyst.level = WARN
+
+logger.camel.name = org.apache.camel
+logger.camel.level = INFO
+logger.camel-consul.name = org.apache.camel.component.consul
+logger.camel-consul.level = DEBUG
logger.camel-ha.name = org.apache.camel.ha
logger.camel-ha.level = DEBUG
logger.camel-impl-ha.name = org.apache.camel.impl.ha
@@ -34,4 +43,5 @@ logger.camel-impl-ha.level = DEBUG
rootLogger.level = INFO
#rootLogger.appenderRef.stdout.ref = out
-rootLogger.appenderRef.out.ref = file
\ No newline at end of file
+rootLogger.appenderRef.out.ref = file
+
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterView.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterView.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterView.java
index 148f47d..23b4305 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterView.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterView.java
@@ -95,7 +95,7 @@ public class KubernetesClusterView extends AbstractCamelClusterView {
// New leader
Optional<String> leader = KubernetesClusterEvent.KubernetesClusterLeaderChangedEvent.class.cast(event).getData();
currentLeader = leader.map(this::toMember);
- fireLeadershipChangedEvent(currentLeader.orElse(null));
+ fireLeadershipChangedEvent(currentLeader);
} else if (event instanceof KubernetesClusterEvent.KubernetesClusterMemberListChangedEvent) {
Set<String> members = KubernetesClusterEvent.KubernetesClusterMemberListChangedEvent.class.cast(event).getData();
Set<String> oldMembers = currentMembers.stream().map(CamelClusterMember::getId).collect(Collectors.toSet());
@@ -153,11 +153,15 @@ public class KubernetesClusterView extends AbstractCamelClusterView {
}
@Override
+ public boolean isLocal() {
+ return ObjectHelper.equal(lockConfiguration.getPodName(), podName);
+ }
+
+ @Override
public String getId() {
return podName;
}
-
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("KubernetesClusterMember{");
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLockConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLockConfiguration.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLockConfiguration.java
index 69c54d9..94dcca8 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLockConfiguration.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLockConfiguration.java
@@ -44,7 +44,7 @@ public class KubernetesLockConfiguration implements Cloneable {
private String configMapName = DEFAULT_CONFIGMAP_NAME;
/**
- * Name of the lock group (or namespace according to the Camel cluster convention) within the chosen ConfgMap.
+ * Name of the lock group (or namespace according to the Camel cluster convention) within the chosen ConfigMap.
*/
private String groupName;
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/utils/LeaderRecorder.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/utils/LeaderRecorder.java b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/utils/LeaderRecorder.java
index 7d7147b..0f8199d 100644
--- a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/utils/LeaderRecorder.java
+++ b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/utils/LeaderRecorder.java
@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
@@ -41,9 +42,9 @@ public class LeaderRecorder implements CamelClusterEventListener.Leadership {
private List<LeadershipInfo> leaderships = new CopyOnWriteArrayList<>();
@Override
- public void leadershipChanged(CamelClusterView view, CamelClusterMember leader) {
+ public void leadershipChanged(CamelClusterView view, Optional<CamelClusterMember> leader) {
LOG.info("Cluster view {} - leader changed to: {}", view.getLocalMember(), leader);
- this.leaderships.add(new LeadershipInfo(leader != null ? leader.getId() : null, System.currentTimeMillis()));
+ this.leaderships.add(new LeadershipInfo(leader.map(CamelClusterMember::getId).orElse(null), System.currentTimeMillis()));
}
public List<LeadershipInfo> getLeadershipInfo() {
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusterView.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusterView.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusterView.java
index 6310ded..78ceed5 100644
--- a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusterView.java
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusterView.java
@@ -19,7 +19,6 @@ package org.apache.camel.component.zookeeper.ha;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.apache.camel.RuntimeCamelException;
@@ -27,6 +26,7 @@ import org.apache.camel.component.zookeeper.ZooKeeperCuratorConfiguration;
import org.apache.camel.ha.CamelClusterMember;
import org.apache.camel.ha.CamelClusterService;
import org.apache.camel.impl.ha.AbstractCamelClusterView;
+import org.apache.camel.util.ObjectHelper;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
@@ -40,7 +40,7 @@ final class ZooKeeperClusterView extends AbstractCamelClusterView {
private final ZooKeeperCuratorConfiguration configuration;
private final CuratorFramework client;
private final CuratorLocalMember localMember;
- private LeaderSelector leaderSelector;
+ private volatile LeaderSelector leaderSelector;
public ZooKeeperClusterView(CamelClusterService cluster, ZooKeeperCuratorConfiguration configuration, CuratorFramework client, String namespace) {
super(cluster, namespace);
@@ -57,12 +57,16 @@ final class ZooKeeperClusterView extends AbstractCamelClusterView {
@Override
public Optional<CamelClusterMember> getMaster() {
- if (leaderSelector == null) {
+ if (leaderSelector == null || isStoppingOrStopped()) {
return Optional.empty();
}
try {
- return Optional.of(new CuratorClusterMember(leaderSelector.getLeader()));
+ Participant participant = leaderSelector.getLeader();
+
+ return ObjectHelper.equal(participant.getId(), localMember.getId())
+ ? Optional.of(localMember)
+ : Optional.of(new CuratorClusterMember(participant));
} catch (Exception e) {
throw new RuntimeCamelException(e);
}
@@ -97,19 +101,27 @@ final class ZooKeeperClusterView extends AbstractCamelClusterView {
protected void doStop() throws Exception {
if (leaderSelector != null) {
leaderSelector.interruptLeadership();
- leaderSelector.close();
- leaderSelector = null;
+ fireLeadershipChangedEvent(getMaster());
+ } else {
+ leaderSelector.requeue();
+ }
+ }
- localMember.setMaster(false);
- fireLeadershipChangedEvent(getMaster().orElse(null));
+ @Override
+ protected void doShutdown() throws Exception {
+ if (leaderSelector != null) {
+ leaderSelector.close();
}
}
- class CamelLeaderElectionListener extends LeaderSelectorListenerAdapter {
+ // ***********************************************
+ //
+ // ***********************************************
+
+ private final class CamelLeaderElectionListener extends LeaderSelectorListenerAdapter {
@Override
public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
- localMember.setMaster(true);
- fireLeadershipChangedEvent(localMember);
+ fireLeadershipChangedEvent(Optional.of(localMember));
while (isRunAllowed()) {
try {
@@ -119,26 +131,20 @@ final class ZooKeeperClusterView extends AbstractCamelClusterView {
break;
}
}
-
- localMember.setMaster(false);
- fireLeadershipChangedEvent(getMaster().orElse(null));
+
+ fireLeadershipChangedEvent(getMaster());
}
}
- // ***********************************************
- //
- // ***********************************************
-
private final class CuratorLocalMember implements CamelClusterMember {
- private AtomicBoolean master = new AtomicBoolean(false);
-
- void setMaster(boolean master) {
- this.master.set(master);
+ @Override
+ public boolean isLeader() {
+ return leaderSelector != null ? leaderSelector.hasLeadership() : false;
}
@Override
- public boolean isLeader() {
- return master.get();
+ public boolean isLocal() {
+ return true;
}
@Override
@@ -160,6 +166,13 @@ final class ZooKeeperClusterView extends AbstractCamelClusterView {
}
@Override
+ public boolean isLocal() {
+ return (participant.getId() != null)
+ ? ObjectHelper.equal(participant.getId(), localMember.getId())
+ : false;
+ }
+
+ @Override
public boolean isLeader() {
try {
return leaderSelector.getLeader().equals(this.participant);
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusteredRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusteredRoutePolicyTest.java b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusteredRoutePolicyTest.java
index 1c32787..6aa7469 100644
--- a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusteredRoutePolicyTest.java
+++ b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ha/ZooKeeperClusteredRoutePolicyTest.java
@@ -79,7 +79,8 @@ public final class ZooKeeperClusteredRoutePolicyTest {
private static void run(String id) {
try {
- CountDownLatch contextLatch = new CountDownLatch(1);
+ int events = ThreadLocalRandom.current().nextInt(2, 6);
+ CountDownLatch contextLatch = new CountDownLatch(events);
ZooKeeperClusterService service = new ZooKeeperClusterService();
service.setId("node-" + id);
@@ -94,15 +95,10 @@ public final class ZooKeeperClusteredRoutePolicyTest {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("timer:zookeeper?delay=1s&period=1s&repeatCount=1")
+ from("timer:zookeeper?delay=1s&period=1s")
.routeId("route-" + id)
- .process(e -> {
- LOGGER.debug("Node {} done", id);
- RESULTS.add(id);
- // Shutdown the context later on to give a chance to
- // other members to catch-up
- SCHEDULER.schedule(contextLatch::countDown, 2 + ThreadLocalRandom.current().nextInt(3), TimeUnit.SECONDS);
- });
+ .log("From ${routeId}")
+ .process(e -> contextLatch.countDown());
}
});
@@ -112,6 +108,10 @@ public final class ZooKeeperClusteredRoutePolicyTest {
context.start();
contextLatch.await();
+
+ LOGGER.debug("Shutting down node {}", id);
+ RESULTS.add(id);
+
context.stop();
LATCH.countDown();
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/components/camel-zookeeper/src/test/resources/log4j2.properties
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/test/resources/log4j2.properties b/components/camel-zookeeper/src/test/resources/log4j2.properties
index bab98a6..52dfed8 100644
--- a/components/camel-zookeeper/src/test/resources/log4j2.properties
+++ b/components/camel-zookeeper/src/test/resources/log4j2.properties
@@ -28,16 +28,19 @@ appender.out.layout.pattern = [%t] %c{1} %-5p %m%n
logger.zookeeper.name = org.apache.zookeeper
logger.zookeeper.level = INFO
+
+logger.camel.name = org.apache.camel
+logger.camel.level = INFO
logger.camel-zookeeper.name = org.apache.camel.component.zookeeper
logger.camel-zookeeper.level = INFO
logger.camel-zookeeper-policy.name = org.apache.camel.component.zookeeper.policy
logger.camel-zookeeper-policy.level = INFO
+logger.camel-ha.name = org.apache.camel.ha
+logger.camel-ha.level = DEBUG
+logger.camel-impl-ha.name = org.apache.camel.impl.ha
+logger.camel-impl-ha.level = DEBUG
logger.camel-support.name = org.apache.camel.support
logger.camel-support.level = INFO
-logger.camel-ha-impl.name = org.apache.camel.impl.ha
-logger.camel-ha-impl.level = DEBUG
-logger.camel.name = org.apache.camel
-logger.camel.level = INFO
logger.springframework.name = org.springframework
logger.springframework.level = WARN
@@ -46,4 +49,3 @@ rootLogger.level = INFO
#rootLogger.appenderRef.stdout.ref = out
rootLogger.appenderRef.file.ref = file
-
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceAutoConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceAutoConfiguration.java
index a6e1e6b..7d7c6cd 100644
--- a/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceAutoConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceAutoConfiguration.java
@@ -47,7 +47,7 @@ public class AtomixClusterServiceAutoConfiguration {
@Bean(initMethod = "start", destroyMethod = "stop")
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
- @ConditionalOnProperty(prefix = "camel.clustered.service.atomix", name = "mode", havingValue = "node")
+ @ConditionalOnProperty(prefix = "camel.component.atomix.cluster.service", name = "mode", havingValue = "node")
public CamelClusterService atomixClusterService() {
AtomixClusterService service = new AtomixClusterService();
service.setNodes(configuration.getNodes().stream().map(Address::new).collect(Collectors.toList()));
@@ -64,7 +64,7 @@ public class AtomixClusterServiceAutoConfiguration {
@Bean(initMethod = "start", destroyMethod = "stop")
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
- @ConditionalOnProperty(prefix = "camel.clustered.service.atomix", name = "mode", havingValue = "client")
+ @ConditionalOnProperty(prefix = "camel.component.atomix.cluster.service", name = "mode", havingValue = "client")
public CamelClusterService atomixClusterClientService() {
AtomixClusterClientService service = new AtomixClusterClientService();
service.setNodes(configuration.getNodes().stream().map(Address::new).collect(Collectors.toList()));
@@ -84,11 +84,11 @@ public class AtomixClusterServiceAutoConfiguration {
super(ConfigurationPhase.REGISTER_BEAN);
}
- @ConditionalOnProperty(prefix = "camel.clustered.service.atomix", name = "enabled")
+ @ConditionalOnProperty(prefix = "camel.component.atomix.cluster.service", name = "enabled")
static class IfEnabled {
}
- @ConditionalOnProperty(prefix = "camel.clustered.service.atomix", name = "mode")
+ @ConditionalOnProperty(prefix = "camel.component.atomix.cluster.service", name = "mode")
static class WithMode {
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceConfiguration.java b/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceConfiguration.java
index 27b281a..8317d56 100644
--- a/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-atomix-starter/src/main/java/org/apache/camel/component/atomix/ha/springboot/AtomixClusterServiceConfiguration.java
@@ -22,7 +22,7 @@ import java.util.Set;
import io.atomix.copycat.server.storage.StorageLevel;
import org.springframework.boot.context.properties.ConfigurationProperties;
-@ConfigurationProperties(prefix = "camel.clustered.service.atomix")
+@ConfigurationProperties(prefix = "camel.component.atomix.cluster.service")
public class AtomixClusterServiceConfiguration {
enum Mode {
node,
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceAutoConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceAutoConfiguration.java
index bd0dd8c..082bb7b 100644
--- a/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceAutoConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceAutoConfiguration.java
@@ -33,7 +33,7 @@ import org.springframework.context.annotation.Scope;
@Configuration
@AutoConfigureBefore({ ClusteredRouteControllerAutoConfiguration.class, CamelAutoConfiguration.class })
-@ConditionalOnProperty(prefix = "camel.clustered.service.consul", name = "enabled")
+@ConditionalOnProperty(prefix = "camel.component.consul.cluster.service", name = "enabled")
@EnableConfigurationProperties(ConsulClusterServiceConfiguration.class)
public class ConsulClusterServiceAutoConfiguration {
@Autowired
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceConfiguration.java b/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceConfiguration.java
index d47aa71..bc619c2 100644
--- a/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-consul-starter/src/main/java/org/apache/camel/component/consul/springboot/ha/ConsulClusterServiceConfiguration.java
@@ -19,7 +19,7 @@ package org.apache.camel.component.consul.springboot.ha;
import org.apache.camel.component.consul.ha.ConsulClusterConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
-@ConfigurationProperties(prefix = "camel.clustered.service.consul")
+@ConfigurationProperties(prefix = "camel.component.consul.cluster.service")
public class ConsulClusterServiceConfiguration extends ConsulClusterConfiguration {
/**
* Sets if the zookeeper cluster service should be enabled or not, default is false.
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceAutoConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceAutoConfiguration.java
index 9cb27a0..de80367 100644
--- a/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceAutoConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceAutoConfiguration.java
@@ -33,7 +33,7 @@ import org.springframework.context.annotation.Scope;
@Configuration
@AutoConfigureBefore({ ClusteredRouteControllerAutoConfiguration.class, CamelAutoConfiguration.class })
-@ConditionalOnProperty(prefix = "camel.clustered.service.zookeeper", name = "enabled")
+@ConditionalOnProperty(prefix = "camel.component.zookeeper.cluster.service", name = "enabled")
@EnableConfigurationProperties(ZooKeeperClusterServiceConfiguration.class)
public class ZooKeeperClusterServiceAutoConfiguration {
@Autowired
http://git-wip-us.apache.org/repos/asf/camel/blob/8ef99f5c/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceConfiguration.java b/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceConfiguration.java
index 620ec0e..cb66058 100644
--- a/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-starter/src/main/java/org/apache/camel/component/zookeeper/ha/springboot/ZooKeeperClusterServiceConfiguration.java
@@ -19,7 +19,7 @@ package org.apache.camel.component.zookeeper.ha.springboot;
import org.apache.camel.component.zookeeper.ZooKeeperCuratorConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
-@ConfigurationProperties(prefix = "camel.clustered.service.zookeeper")
+@ConfigurationProperties(prefix = "camel.component.zookeeper.cluster.service")
public class ZooKeeperClusterServiceConfiguration extends ZooKeeperCuratorConfiguration {
/**
* Sets if the zookeeper cluster service should be enabled or not, default is false.