Refine Cluster Service API and implementations

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 @@ interface Leadership extends 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);
 
     }
 

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();
 }

camel-core/src/main/java/org/apache/camel/impl/ha/AbstractCamelClusterView.java

@@ -18,6 +18,7 @@
 
 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 @@ private <T extends CamelClusterEventListener> void doWithListener(Class<T> type,
         );
     }
 
-    protected void fireLeadershipChangedEvent(CamelClusterMember leader) {
+    protected void fireLeadershipChangedEvent(Optional<CamelClusterMember> leader) {
         doWithListener(
             CamelClusterEventListener.Leadership.class,
             listener -> listener.leadershipChanged(this, leader)

camel-core/src/main/java/org/apache/camel/impl/ha/ClusteredRoutePolicy.java

@@ -19,6 +19,7 @@
 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 @@ private void onCamelContextStarted() {
 
     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());
         }
     }

components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java

@@ -147,8 +147,8 @@ protected void doStop() throws Exception {
         super.doStop();
 
         if (atomix != null) {
-            LOGGER.debug("Shutdown atomix replica {}", atomix);
-            atomix.shutdown().join();
+            LOGGER.debug("Leaving atomix cluster replica {}", atomix);
+            atomix.leave().join();
         }
     }
 

components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java

@@ -95,21 +95,21 @@ protected void doStart() throws Exception {
             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 @@ protected void doStop() throws Exception {
         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 @@ public boolean isLeader() {
             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 @@ AtomixLocalMember leave() {
                 group.remove(id).join();
 
                 member = null;
-
-                fireLeadershipChangedEvent(null);
+                fireLeadershipChangedEvent(Optional.empty());
             }
 
             return this;
@@ -199,7 +215,7 @@ public String toString() {
         }
     }
 
-    class AtomixClusterMember implements CamelClusterMember {
+    final class AtomixClusterMember implements CamelClusterMember {
         private final GroupMember member;
 
         AtomixClusterMember(GroupMember member) {
@@ -223,6 +239,11 @@ public boolean isLeader() {
             return member.equals(group.election().term().leader());
         }
 
+        @Override
+        public boolean isLocal() {
+            return localMember != null ? localMember.is(member) : false;
+        }
+
         @Override
         public String toString() {
             final StringBuilder sb = new StringBuilder("AtomixClusterMember{");

components/camel-atomix/src/test/java/org/apache/camel/component/atomix/ha/AtomixClientRoutePolicyTestSupport.java

@@ -81,7 +81,8 @@ public void test() throws Exception {
 
     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 @@ private void run(String id) {
             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 void configure() throws Exception {
             context.start();
 
             contextLatch.await();
+
+            LOGGER.debug("Shutting down client node {}", id);
+            results.add(id);
+
             context.stop();
 
             latch.countDown();

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 void test() throws Exception {
 
     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 @@ private void run(Address address) {
             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 void configure() throws Exception {
             context.start();
 
             contextLatch.await();
+
+            LOGGER.debug("Shutting down node {}", address);
+            results.add(address);
+
             context.stop();
 
             latch.countDown();

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

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>

components/camel-consul/src/main/java/org/apache/camel/component/consul/ha/ConsulClusterView.java

@@ -35,6 +35,7 @@
 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 @@ protected void doStop() throws Exception {
                 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 @@ private final class ConsulLocalMember implements CamelClusterMember {
         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;
             }
         }
@@ -164,6 +167,11 @@ public boolean isLeader() {
             return master.get();
         }
 
+        @Override
+        public boolean isLocal() {
+            return true;
+        }
+
         @Override
         public String getId() {
             return sessionId.get();
@@ -209,6 +217,15 @@ public boolean isLeader() {
             return id.equals(keyValueClient.getSession(path));
         }
 
+        @Override
+        public boolean isLocal() {
+            if (id == null) {
+                return false;
+            }
+
+            return ObjectHelper.equal(id, localMember.getId());
+        }
+
         @Override
         public String toString() {
             return "ConsulClusterMember{"