camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nferr...@apache.org
Subject [04/14] camel git commit: CAMEL-11331: Clock-drift-free version of the protocol
Date Tue, 08 Aug 2017 14:43:22 GMT
CAMEL-11331: Clock-drift-free version of the protocol


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f0b00ab9
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f0b00ab9
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f0b00ab9

Branch: refs/heads/master
Commit: f0b00ab927e9e55c238c8ea77e1b08c65335ca01
Parents: a0b2320
Author: Nicola Ferraro <ni.ferraro@gmail.com>
Authored: Mon Jul 31 17:17:34 2017 +0200
Committer: Nicola Ferraro <ni.ferraro@gmail.com>
Committed: Tue Aug 8 16:39:43 2017 +0200

----------------------------------------------------------------------
 .../kubernetes/ha/KubernetesClusterService.java |  84 ++--
 .../kubernetes/ha/KubernetesClusterView.java    |   6 +-
 .../kubernetes/ha/lock/ConfigMapLockUtils.java  |  15 +-
 .../ha/lock/KubernetesLeadershipController.java | 352 +++++++++++++++++
 ...ubernetesLeaseBasedLeadershipController.java | 384 -------------------
 .../ha/lock/KubernetesLockConfiguration.java    |  84 ++--
 .../ha/lock/KubernetesMembersMonitor.java       | 237 ------------
 .../kubernetes/ha/lock/LeaderInfo.java          |  46 ++-
 .../kubernetes/ha/lock/TimedLeaderNotifier.java | 179 +++++++++
 .../ha/KubernetesClusterServiceTest.java        |  59 +--
 .../kubernetes/ha/TimedLeaderNotifierTest.java  | 117 ++++++
 .../kubernetes/ha/utils/LeaderRecorder.java     |   5 +
 .../kubernetes/ha/utils/LockTestServer.java     |  31 +-
 13 files changed, 813 insertions(+), 786 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterService.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterService.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterService.java
index a868d16..08ebb70 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterService.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterService.java
@@ -82,25 +82,22 @@ public class KubernetesClusterService extends AbstractCamelClusterService<Kubern
         if (config.getJitterFactor() < 1) {
             throw new IllegalStateException("jitterFactor must be >= 1 (found: " + config.getJitterFactor() + ")");
         }
-        if (config.getRetryOnErrorIntervalSeconds() <= 0) {
-            throw new IllegalStateException("retryOnErrorIntervalSeconds must be > 0 (found: " + config.getRetryOnErrorIntervalSeconds() + ")");
+        if (config.getRetryPeriodMillis() <= 0) {
+            throw new IllegalStateException("retryPeriodMillis must be > 0 (found: " + config.getRetryPeriodMillis() + ")");
         }
-        if (config.getRetryPeriodSeconds() <= 0) {
-            throw new IllegalStateException("retryPeriodSeconds must be > 0 (found: " + config.getRetryPeriodSeconds() + ")");
+        if (config.getRenewDeadlineMillis() <= 0) {
+            throw new IllegalStateException("renewDeadlineMillis must be > 0 (found: " + config.getRenewDeadlineMillis() + ")");
         }
-        if (config.getRenewDeadlineSeconds() <= 0) {
-            throw new IllegalStateException("renewDeadlineSeconds must be > 0 (found: " + config.getRenewDeadlineSeconds() + ")");
+        if (config.getLeaseDurationMillis() <= 0) {
+            throw new IllegalStateException("leaseDurationMillis must be > 0 (found: " + config.getLeaseDurationMillis() + ")");
         }
-        if (config.getLeaseDurationSeconds() <= 0) {
-            throw new IllegalStateException("leaseDurationSeconds must be > 0 (found: " + config.getLeaseDurationSeconds() + ")");
+        if (config.getLeaseDurationMillis() <= config.getRenewDeadlineMillis()) {
+            throw new IllegalStateException("leaseDurationMillis must be greater than renewDeadlineMillis "
+                    + "(" + config.getLeaseDurationMillis() + " is not greater than " + config.getRenewDeadlineMillis() + ")");
         }
-        if (config.getLeaseDurationSeconds() <= config.getRenewDeadlineSeconds()) {
-            throw new IllegalStateException("leaseDurationSeconds must be greater than renewDeadlineSeconds "
-                    + "(" + config.getLeaseDurationSeconds() + " is not greater than " + config.getRenewDeadlineSeconds() + ")");
-        }
-        if (config.getRenewDeadlineSeconds() <= config.getJitterFactor() * config.getRetryPeriodSeconds()) {
-            throw new IllegalStateException("renewDeadlineSeconds must be greater than jitterFactor*retryPeriodSeconds "
-                    + "(" + config.getRenewDeadlineSeconds() + " is not greater than " + config.getJitterFactor() + "*" + config.getRetryPeriodSeconds() + ")");
+        if (config.getRenewDeadlineMillis() <= config.getJitterFactor() * config.getRetryPeriodMillis()) {
+            throw new IllegalStateException("renewDeadlineMillis must be greater than jitterFactor*retryPeriodMillis "
+                    + "(" + config.getRenewDeadlineMillis() + " is not greater than " + config.getJitterFactor() + "*" + config.getRetryPeriodMillis() + ")");
         }
 
         return config;
@@ -176,16 +173,8 @@ public class KubernetesClusterService extends AbstractCamelClusterService<Kubern
         lockConfiguration.setKubernetesResourcesNamespace(kubernetesResourcesNamespace);
     }
 
-    public long getRetryOnErrorIntervalSeconds() {
-        return lockConfiguration.getRetryOnErrorIntervalSeconds();
-    }
-
-    /**
-     * Indicates the maximum amount of time a Kubernetes watch should be kept active, before being recreated.
-     * Watch recreation can be disabled by putting value <= 0.
-     */
-    public void setRetryOnErrorIntervalSeconds(long retryOnErrorIntervalSeconds) {
-        lockConfiguration.setRetryOnErrorIntervalSeconds(retryOnErrorIntervalSeconds);
+    public String getKubernetesResourcesNamespaceOrDefault(KubernetesClient kubernetesClient) {
+        return lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient);
     }
 
     public double getJitterFactor() {
@@ -193,56 +182,43 @@ public class KubernetesClusterService extends AbstractCamelClusterService<Kubern
     }
 
     /**
-     * A jitter factor to apply in order to prevent all pods to try to become leaders in the same instant.
+     * A jitter factor to apply in order to prevent all pods to call Kubernetes APIs in the same instant.
      */
     public void setJitterFactor(double jitterFactor) {
         lockConfiguration.setJitterFactor(jitterFactor);
     }
 
-    public long getLeaseDurationSeconds() {
-        return lockConfiguration.getLeaseDurationSeconds();
+    public long getLeaseDurationMillis() {
+        return lockConfiguration.getLeaseDurationMillis();
     }
 
     /**
      * The default duration of the lease for the current leader.
      */
-    public void setLeaseDurationSeconds(long leaseDurationSeconds) {
-        lockConfiguration.setLeaseDurationSeconds(leaseDurationSeconds);
-    }
-
-    public long getRenewDeadlineSeconds() {
-        return lockConfiguration.getRenewDeadlineSeconds();
-    }
-
-    /**
-     * The deadline after which the leader must stop trying to renew its leadership (and yield it).
-     */
-    public void setRenewDeadlineSeconds(long renewDeadlineSeconds) {
-        lockConfiguration.setRenewDeadlineSeconds(renewDeadlineSeconds);
+    public void setLeaseDurationMillis(long leaseDurationMillis) {
+        lockConfiguration.setLeaseDurationMillis(leaseDurationMillis);
     }
 
-    public long getRetryPeriodSeconds() {
-        return lockConfiguration.getRetryPeriodSeconds();
+    public long getRenewDeadlineMillis() {
+        return lockConfiguration.getRenewDeadlineMillis();
     }
 
     /**
-     * The time between two subsequent attempts to acquire/renew the leadership (or after the lease expiration).
-     * It is randomized using the jitter factor in case of new leader election (not renewal).
+     * The deadline after which the leader must stop its services because it may have lost the leadership.
      */
-    public void setRetryPeriodSeconds(long retryPeriodSeconds) {
-        lockConfiguration.setRetryPeriodSeconds(retryPeriodSeconds);
+    public void setRenewDeadlineMillis(long renewDeadlineMillis) {
+        lockConfiguration.setRenewDeadlineMillis(renewDeadlineMillis);
     }
 
-    public long getWatchRefreshIntervalSeconds() {
-        return lockConfiguration.getWatchRefreshIntervalSeconds();
+    public long getRetryPeriodMillis() {
+        return lockConfiguration.getRetryPeriodMillis();
     }
 
     /**
-     * Set this to a positive value in order to recreate watchers after a certain amount of time,
-     * to avoid having stale watchers.
+     * The time between two subsequent attempts to check and acquire the leadership.
+     * It is randomized using the jitter factor.
      */
-    public void setWatchRefreshIntervalSeconds(long watchRefreshIntervalSeconds) {
-        lockConfiguration.setWatchRefreshIntervalSeconds(watchRefreshIntervalSeconds);
+    public void setRetryPeriodMillis(long retryPeriodMillis) {
+        lockConfiguration.setRetryPeriodMillis(retryPeriodMillis);
     }
-
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/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 28f38a5..ddda675 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
@@ -30,7 +30,7 @@ import io.fabric8.kubernetes.client.KubernetesClient;
 import org.apache.camel.component.kubernetes.KubernetesConfiguration;
 import org.apache.camel.component.kubernetes.KubernetesHelper;
 import org.apache.camel.component.kubernetes.ha.lock.KubernetesClusterEvent;
-import org.apache.camel.component.kubernetes.ha.lock.KubernetesLeaseBasedLeadershipController;
+import org.apache.camel.component.kubernetes.ha.lock.KubernetesLeadershipController;
 import org.apache.camel.component.kubernetes.ha.lock.KubernetesLockConfiguration;
 import org.apache.camel.ha.CamelClusterMember;
 import org.apache.camel.impl.ha.AbstractCamelClusterView;
@@ -56,7 +56,7 @@ public class KubernetesClusterView extends AbstractCamelClusterView {
 
     private volatile List<CamelClusterMember> currentMembers = Collections.emptyList();
 
-    private KubernetesLeaseBasedLeadershipController controller;
+    private KubernetesLeadershipController controller;
 
     public KubernetesClusterView(KubernetesClusterService cluster, KubernetesConfiguration configuration, KubernetesLockConfiguration lockConfiguration) {
         super(cluster, lockConfiguration.getGroupName());
@@ -86,7 +86,7 @@ public class KubernetesClusterView extends AbstractCamelClusterView {
         if (controller == null) {
             this.kubernetesClient = KubernetesHelper.getKubernetesClient(configuration);
 
-            controller = new KubernetesLeaseBasedLeadershipController(kubernetesClient, this.lockConfiguration, event -> {
+            controller = new KubernetesLeadershipController(kubernetesClient, this.lockConfiguration, event -> {
                 if (event instanceof KubernetesClusterEvent.KubernetesClusterLeaderChangedEvent) {
                     // New leader
                     Optional<String> leader = KubernetesClusterEvent.KubernetesClusterLeaderChangedEvent.class.cast(event).getData();

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/ConfigMapLockUtils.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/ConfigMapLockUtils.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/ConfigMapLockUtils.java
index 70fa860..feea1c6 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/ConfigMapLockUtils.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/ConfigMapLockUtils.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.kubernetes.ha.lock;
 
 import java.text.SimpleDateFormat;
 import java.util.Date;
+import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
@@ -36,7 +37,7 @@ public final class ConfigMapLockUtils {
 
     private static final String LEADER_PREFIX = "leader.pod.";
 
-    private static final String TIMESTAMP_PREFIX = "leader.timestamp.";
+    private static final String LOCAL_TIMESTAMP_PREFIX = "leader.local.timestamp.";
 
     private ConfigMapLockUtils() {
     }
@@ -49,19 +50,19 @@ public final class ConfigMapLockUtils {
                     .addToLabels("kind", "locks").
                 endMetadata()
                 .addToData(LEADER_PREFIX + leaderInfo.getGroupName(), leaderInfo.getLeader())
-                .addToData(TIMESTAMP_PREFIX + leaderInfo.getGroupName(), formatDate(leaderInfo.getTimestamp()))
+                .addToData(LOCAL_TIMESTAMP_PREFIX + leaderInfo.getGroupName(), formatDate(leaderInfo.getLocalTimestamp()))
                 .build();
     }
 
     public static ConfigMap getConfigMapWithNewLeader(ConfigMap configMap, LeaderInfo leaderInfo) {
         return new ConfigMapBuilder(configMap)
                 .addToData(LEADER_PREFIX + leaderInfo.getGroupName(), leaderInfo.getLeader())
-                .addToData(TIMESTAMP_PREFIX + leaderInfo.getGroupName(), formatDate(leaderInfo.getTimestamp()))
+                .addToData(LOCAL_TIMESTAMP_PREFIX + leaderInfo.getGroupName(), formatDate(leaderInfo.getLocalTimestamp()))
                 .build();
     }
 
-    public static LeaderInfo getLeaderInfo(ConfigMap configMap, String group) {
-        return new LeaderInfo(group, getLeader(configMap, group), getTimestamp(configMap, group));
+    public static LeaderInfo getLeaderInfo(ConfigMap configMap, Set<String> members, String group) {
+        return new LeaderInfo(group, getLeader(configMap, group), getLocalTimestamp(configMap, group), members);
     }
 
     private static String getLeader(ConfigMap configMap, String group) {
@@ -81,8 +82,8 @@ public final class ConfigMapLockUtils {
         return null;
     }
 
-    private static Date getTimestamp(ConfigMap configMap, String group) {
-        String timestamp = getConfigMapValue(configMap, TIMESTAMP_PREFIX + group);
+    private static Date getLocalTimestamp(ConfigMap configMap, String group) {
+        String timestamp = getConfigMapValue(configMap, LOCAL_TIMESTAMP_PREFIX + group);
         if (timestamp == null) {
             return null;
         }

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeadershipController.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeadershipController.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeadershipController.java
new file mode 100644
index 0000000..f527779
--- /dev/null
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeadershipController.java
@@ -0,0 +1,352 @@
+/**
+ * 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.kubernetes.ha.lock;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClient;
+
+import org.apache.camel.Service;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Monitors current status and participate to leader election when no active leaders are present.
+ * It communicates changes in leadership and cluster members to the given event handler.
+ */
+public class KubernetesLeadershipController implements Service {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KubernetesLeadershipController.class);
+
+    private enum State {
+        NOT_LEADER,
+        BECOMING_LEADER,
+        LEADER
+    }
+
+    private KubernetesClient kubernetesClient;
+
+    private KubernetesLockConfiguration lockConfiguration;
+
+    private KubernetesClusterEventHandler eventHandler;
+
+    private State currentState = State.NOT_LEADER;
+
+    private ScheduledExecutorService serializedExecutor;
+
+    private TimedLeaderNotifier leaderNotifier;
+
+    private volatile LeaderInfo latestLeaderInfo;
+    private volatile ConfigMap latestConfigMap;
+    private volatile Set<String> latestMembers;
+
+    public KubernetesLeadershipController(KubernetesClient kubernetesClient, KubernetesLockConfiguration lockConfiguration, KubernetesClusterEventHandler eventHandler) {
+        this.kubernetesClient = kubernetesClient;
+        this.lockConfiguration = lockConfiguration;
+        this.eventHandler = eventHandler;
+    }
+
+    @Override
+    public void start() throws Exception {
+        if (serializedExecutor == null) {
+            LOG.debug("{} Starting leadership controller...", logPrefix());
+            serializedExecutor = Executors.newSingleThreadScheduledExecutor();
+            leaderNotifier = new TimedLeaderNotifier(this.eventHandler);
+
+            leaderNotifier.start();
+            serializedExecutor.execute(this::refreshStatus);
+        }
+    }
+
+    @Override
+    public void stop() throws Exception {
+        LOG.debug("{} Stopping leadership controller...", logPrefix());
+        if (serializedExecutor != null) {
+            serializedExecutor.shutdownNow();
+        }
+        serializedExecutor = null;
+
+        if (leaderNotifier != null) {
+            leaderNotifier.stop();
+        }
+        leaderNotifier = null;
+    }
+
+    private void refreshStatus() {
+        switch (currentState) {
+        case NOT_LEADER:
+            refreshStatusNotLeader();
+            break;
+        case BECOMING_LEADER:
+            refreshStatusBecomingLeader();
+            break;
+        case LEADER:
+            refreshStatusLeader();
+            break;
+        default:
+            throw new RuntimeException("Unsupported state " + currentState);
+        }
+    }
+
+    /**
+     * This pod is currently not leader. It should monitor the leader configuration and try
+     * to acquire the leadership if possible.
+     */
+    private void refreshStatusNotLeader() {
+        LOG.debug("{} Pod is not leader, pulling new data from the cluster", logPrefix());
+        boolean pulled = lookupNewLeaderInfo();
+        if (!pulled) {
+            rescheduleAfterDelay();
+            return;
+        }
+
+        if (this.latestLeaderInfo.hasEmptyLeader()) {
+            // There is no previous leader
+            LOG.info("{} The cluster has no leaders. Trying to acquire the leadership...", logPrefix());
+            boolean acquired = tryAcquireLeadership();
+            if (acquired) {
+                LOG.info("{} Leadership acquired by current pod ({}) with immediate effect", logPrefix(), this.lockConfiguration.getPodName());
+                this.currentState = State.LEADER;
+                this.serializedExecutor.execute(this::refreshStatus);
+                return;
+            } else {
+                LOG.info("{} Unable to acquire the leadership, it may have been acquired by another pod", logPrefix());
+            }
+        } else if (!this.latestLeaderInfo.hasValidLeader()) {
+            // There's a previous leader and it's invalid
+            LOG.info("{} Leadership has been lost by old owner. Trying to acquire the leadership...", logPrefix());
+            boolean acquired = tryAcquireLeadership();
+            if (acquired) {
+                LOG.info("{} Leadership acquired by current pod ({})", logPrefix(), this.lockConfiguration.getPodName());
+                this.currentState = State.BECOMING_LEADER;
+                this.serializedExecutor.execute(this::refreshStatus);
+                return;
+            } else {
+                LOG.info("{} Unable to acquire the leadership, it may have been acquired by another pod", logPrefix());
+            }
+        } else if (this.latestLeaderInfo.isValidLeader(this.lockConfiguration.getPodName())) {
+            // We are leaders for some reason (e.g. pod restart on failure)
+            LOG.info("{} Leadership is already owned by current pod ({})", logPrefix(), this.lockConfiguration.getPodName());
+            this.currentState = State.BECOMING_LEADER;
+            this.serializedExecutor.execute(this::refreshStatus);
+            return;
+        }
+
+        this.leaderNotifier.refreshLeadership(Optional.ofNullable(this.latestLeaderInfo.getLeader()),
+                System.currentTimeMillis(),
+                this.lockConfiguration.getLeaseDurationMillis(),
+                this.latestLeaderInfo.getMembers());
+        rescheduleAfterDelay();
+    }
+
+    /**
+     * This pod has acquired the leadership but it should wait for the old leader
+     * to tear down resources before starting the local services.
+     */
+    private void refreshStatusBecomingLeader() {
+        // Wait always the same amount of time before becoming the leader
+        // Even if the current pod is already leader, we should let a possible old version of the pod to shut down
+        long delay = this.lockConfiguration.getLeaseDurationMillis();
+        LOG.info("{} Current pod ({}) owns the leadership, but it will be effective in {} seconds...", logPrefix(), this.lockConfiguration.getPodName(), new BigDecimal(delay).divide(BigDecimal
+                .valueOf(1000), 2, BigDecimal.ROUND_HALF_UP));
+
+        try {
+            Thread.sleep(delay);
+        } catch (InterruptedException e) {
+            LOG.warn("Thread interrupted", e);
+        }
+
+        LOG.info("{} Current pod ({}) is becoming the new leader now...", logPrefix(), this.lockConfiguration.getPodName());
+        this.currentState = State.LEADER;
+        this.serializedExecutor.execute(this::refreshStatus);
+    }
+
+    private void refreshStatusLeader() {
+        LOG.debug("{} Pod should be the leader, pulling new data from the cluster", logPrefix());
+        long timeBeforePulling = System.currentTimeMillis();
+        boolean pulled = lookupNewLeaderInfo();
+        if (!pulled) {
+            rescheduleAfterDelay();
+            return;
+        }
+
+        if (this.latestLeaderInfo.isValidLeader(this.lockConfiguration.getPodName())) {
+            LOG.debug("{} Current Pod ({}) is still the leader", logPrefix(), this.lockConfiguration.getPodName());
+            this.leaderNotifier.refreshLeadership(Optional.of(this.lockConfiguration.getPodName()),
+                    timeBeforePulling,
+                    this.lockConfiguration.getRenewDeadlineMillis(),
+                    this.latestLeaderInfo.getMembers());
+            rescheduleAfterDelay();
+            return;
+        } else {
+            LOG.debug("{} Current Pod ({}) has lost the leadership", logPrefix(), this.lockConfiguration.getPodName());
+            this.currentState = State.NOT_LEADER;
+            // set a empty leader to signal leadership loss
+            this.leaderNotifier.refreshLeadership(Optional.empty(),
+                    System.currentTimeMillis(),
+                    lockConfiguration.getLeaseDurationMillis(),
+                    this.latestLeaderInfo.getMembers());
+
+            // wait a lease time and restart
+            this.serializedExecutor.schedule(this::refreshStatus, this.lockConfiguration.getLeaseDurationMillis(), TimeUnit.MILLISECONDS);
+        }
+    }
+
+    private void rescheduleAfterDelay() {
+        this.serializedExecutor.schedule(this::refreshStatus, jitter(this.lockConfiguration.getRetryPeriodMillis(), this.lockConfiguration.getJitterFactor()), TimeUnit.MILLISECONDS);
+    }
+
+    private boolean lookupNewLeaderInfo() {
+        LOG.debug("{} Looking up leadership information...", logPrefix());
+
+        ConfigMap configMap;
+        try {
+            configMap = pullConfigMap();
+        } catch (Throwable e) {
+            LOG.warn(logPrefix() + " Unable to retrieve the current ConfigMap " + this.lockConfiguration.getConfigMapName() + " from Kubernetes");
+            LOG.debug(logPrefix() + " Exception thrown during ConfigMap lookup", e);
+            return false;
+        }
+
+        Set<String> members;
+        try {
+            members = Objects.requireNonNull(pullClusterMembers(), "Retrieved a null set of members");
+        } catch (Throwable e) {
+            LOG.warn(logPrefix() + " Unable to retrieve the list of cluster members from Kubernetes");
+            LOG.debug(logPrefix() + " Exception thrown during Pod list lookup", e);
+            return false;
+        }
+
+        updateLatestLeaderInfo(configMap, members);
+        return true;
+    }
+
+    private boolean tryAcquireLeadership() {
+        LOG.debug("{} Trying to acquire the leadership...", logPrefix());
+
+        ConfigMap configMap = this.latestConfigMap;
+        Set<String> members = this.latestMembers;
+        LeaderInfo latestLeaderInfo = this.latestLeaderInfo;
+
+        if (latestLeaderInfo == null || members == null) {
+            LOG.warn(logPrefix() + " Unexpected condition. Latest leader info or list of members is empty.");
+            return false;
+        } else if (!members.contains(this.lockConfiguration.getPodName())) {
+            LOG.warn(logPrefix() + " The list of cluster members " + latestLeaderInfo.getMembers() + " does not contain the current pod (" + this.lockConfiguration.getPodName() + "). Cannot acquire"
+                    + " leadership.");
+            return false;
+        }
+
+        // Info we would set set in the configmap to become leaders
+        LeaderInfo newLeaderInfo = new LeaderInfo(this.lockConfiguration.getGroupName(), this.lockConfiguration.getPodName(), new Date(), members);
+
+        if (configMap == null) {
+            // No ConfigMap created so far
+            LOG.debug("{} Lock configmap is not present in the Kubernetes namespace. A new ConfigMap will be created", logPrefix());
+            ConfigMap newConfigMap = ConfigMapLockUtils.createNewConfigMap(this.lockConfiguration.getConfigMapName(), newLeaderInfo);
+
+            try {
+                kubernetesClient.configMaps()
+                        .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
+                        .create(newConfigMap);
+
+                LOG.debug("{} ConfigMap {} successfully created", logPrefix(), this.lockConfiguration.getConfigMapName());
+                updateLatestLeaderInfo(newConfigMap, members);
+                return true;
+            } catch (Exception ex) {
+                // Suppress exception
+                LOG.warn(logPrefix() + " Unable to create the ConfigMap, it may have been created by other cluster members concurrently. If the problem persists, check if the service account has "
+                        + "the right "
+                        + "permissions to create it");
+                LOG.debug(logPrefix() + " Exception while trying to create the ConfigMap", ex);
+                return false;
+            }
+        } else {
+            LOG.debug("{} Lock configmap already present in the Kubernetes namespace. Checking...", logPrefix());
+            LeaderInfo leaderInfo = ConfigMapLockUtils.getLeaderInfo(configMap, members, this.lockConfiguration.getGroupName());
+
+            boolean canAcquire = !leaderInfo.hasValidLeader();
+            if (canAcquire) {
+                // Try to be the new leader
+                try {
+                    ConfigMap updatedConfigMap = ConfigMapLockUtils.getConfigMapWithNewLeader(configMap, newLeaderInfo);
+                    kubernetesClient.configMaps()
+                            .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
+                            .withName(this.lockConfiguration.getConfigMapName())
+                            .lockResourceVersion(configMap.getMetadata().getResourceVersion())
+                            .replace(updatedConfigMap);
+
+                    LOG.debug("{} ConfigMap {} successfully updated", logPrefix(), this.lockConfiguration.getConfigMapName());
+                    updateLatestLeaderInfo(updatedConfigMap, members);
+                    return true;
+                } catch (Exception ex) {
+                    LOG.warn(logPrefix() + " Unable to update the lock ConfigMap to set leadership information");
+                    LOG.debug(logPrefix() + " Error received during configmap lock replace", ex);
+                    return false;
+                }
+            } else {
+                // Another pod is the leader and it's still active
+                LOG.debug("{} Another pod ({}) is the current leader and it is still active", logPrefix(), this.latestLeaderInfo.getLeader());
+                return false;
+            }
+        }
+    }
+
+    private void updateLatestLeaderInfo(ConfigMap configMap, Set<String> members) {
+        LOG.debug("{} Updating internal status about the current leader", logPrefix());
+        this.latestConfigMap = configMap;
+        this.latestMembers = members;
+        this.latestLeaderInfo = ConfigMapLockUtils.getLeaderInfo(configMap, members, this.lockConfiguration.getGroupName());
+        LOG.debug("{} Current leader info: {}", logPrefix(), this.latestLeaderInfo);
+    }
+
+    private ConfigMap pullConfigMap() {
+        return kubernetesClient.configMaps()
+                .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
+                .withName(this.lockConfiguration.getConfigMapName())
+                .get();
+    }
+
+    private Set<String> pullClusterMembers() {
+        List<Pod> pods = kubernetesClient.pods()
+                .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
+                .withLabels(this.lockConfiguration.getClusterLabels())
+                .list().getItems();
+
+        return pods.stream().map(pod -> pod.getMetadata().getName()).collect(Collectors.toSet());
+    }
+
+    private long jitter(long num, double factor) {
+        return (long) (num * (1 + Math.random() * (factor - 1)));
+    }
+
+    private String logPrefix() {
+        return "Leadership Controller [" + this.lockConfiguration.getPodName() + "]";
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeaseBasedLeadershipController.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeaseBasedLeadershipController.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeaseBasedLeadershipController.java
deleted file mode 100644
index 76e91bf..0000000
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesLeaseBasedLeadershipController.java
+++ /dev/null
@@ -1,384 +0,0 @@
-/**
- * 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.kubernetes.ha.lock;
-
-import java.util.Date;
-import java.util.Optional;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import io.fabric8.kubernetes.api.model.ConfigMap;
-import io.fabric8.kubernetes.client.KubernetesClient;
-
-import org.apache.camel.Service;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Monitors current status and participate to leader election when no active leaders are present.
- * It communicates changes in leadership and cluster members to the given event handler.
- */
-public class KubernetesLeaseBasedLeadershipController implements Service {
-
-    private static final Logger LOG = LoggerFactory.getLogger(KubernetesLeaseBasedLeadershipController.class);
-
-    private static final long FIXED_ADDITIONAL_DELAY = 100;
-
-    private KubernetesClient kubernetesClient;
-
-    private KubernetesLockConfiguration lockConfiguration;
-
-    private KubernetesClusterEventHandler eventHandler;
-
-    private ScheduledExecutorService serializedExecutor;
-    private ScheduledExecutorService eventDispatcherExecutor;
-
-    private KubernetesMembersMonitor membersMonitor;
-
-    private Optional<String> currentLeader = Optional.empty();
-
-    private volatile LeaderInfo latestLeaderInfo;
-
-    public KubernetesLeaseBasedLeadershipController(KubernetesClient kubernetesClient, KubernetesLockConfiguration lockConfiguration, KubernetesClusterEventHandler eventHandler) {
-        this.kubernetesClient = kubernetesClient;
-        this.lockConfiguration = lockConfiguration;
-        this.eventHandler = eventHandler;
-    }
-
-    @Override
-    public void start() throws Exception {
-        if (serializedExecutor == null) {
-            LOG.debug("Starting leadership controller...");
-            serializedExecutor = Executors.newSingleThreadScheduledExecutor();
-
-            eventDispatcherExecutor = Executors.newSingleThreadScheduledExecutor();
-
-            membersMonitor = new KubernetesMembersMonitor(this.serializedExecutor, this.kubernetesClient, this.lockConfiguration);
-            if (eventHandler != null) {
-                membersMonitor.addClusterEventHandler(eventHandler);
-            }
-
-            membersMonitor.start();
-            serializedExecutor.execute(this::initialization);
-        }
-    }
-
-    @Override
-    public void stop() throws Exception {
-        LOG.debug("Stopping leadership controller...");
-        if (serializedExecutor != null) {
-            serializedExecutor.shutdownNow();
-        }
-        if (eventDispatcherExecutor != null) {
-            eventDispatcherExecutor.shutdown();
-            eventDispatcherExecutor.awaitTermination(2, TimeUnit.SECONDS);
-            eventDispatcherExecutor.shutdownNow();
-        }
-        if (membersMonitor != null) {
-            membersMonitor.stop();
-        }
-
-        membersMonitor = null;
-        eventDispatcherExecutor = null;
-        serializedExecutor = null;
-    }
-
-    /**
-     * Get the first ConfigMap and setup the initial state.
-     */
-    private void initialization() {
-        LOG.debug("Reading (with retry) the configmap {} to detect the current leader", this.lockConfiguration.getConfigMapName());
-        refreshConfigMapFromCluster(true);
-
-        if (isCurrentPodTheActiveLeader()) {
-            serializedExecutor.execute(this::onLeadershipAcquired);
-        } else {
-            LOG.info("The current pod ({}) is not the leader of the group '{}' in ConfigMap '{}' at this time", this.lockConfiguration.getPodName(), this.lockConfiguration
-                    .getGroupName(), this.lockConfiguration.getConfigMapName());
-            serializedExecutor.execute(this::acquireLeadershipCycle);
-        }
-    }
-
-    /**
-     * Signals the acquisition of the leadership and move to the keep-leadership state.
-     */
-    private void onLeadershipAcquired() {
-        LOG.info("The current pod ({}) is now the leader of the group '{}' in ConfigMap '{}'", this.lockConfiguration.getPodName(), this.lockConfiguration
-                .getGroupName(), this.lockConfiguration.getConfigMapName());
-
-        this.eventDispatcherExecutor.execute(this::checkAndNotifyNewLeader);
-
-        long nextDelay = computeNextRenewWaitTime(this.latestLeaderInfo.getTimestamp(), this.latestLeaderInfo.getTimestamp());
-        serializedExecutor.schedule(this::keepLeadershipCycle, nextDelay + FIXED_ADDITIONAL_DELAY, TimeUnit.MILLISECONDS);
-    }
-
-    /**
-     * While in the keep-leadership state, the controller periodically renews the lease.
-     * If a renewal deadline is met and it was not possible to renew the lease, the leadership is lost.
-     */
-    private void keepLeadershipCycle() {
-        // renew lease periodically
-        refreshConfigMapFromCluster(false); // if possible, update
-
-        if (this.latestLeaderInfo.isTimeElapsedSeconds(lockConfiguration.getRenewDeadlineSeconds()) || !this.latestLeaderInfo.isLeader(this.lockConfiguration.getPodName())) {
-            // Time over for renewal or leadership lost
-            LOG.debug("The current pod ({}) has lost the leadership", this.lockConfiguration.getPodName());
-            serializedExecutor.execute(this::onLeadershipLost);
-            return;
-        }
-
-        boolean success = tryAcquireOrRenewLeadership();
-        LOG.debug("Attempted to renew the lease. Success={}", success);
-
-        long nextDelay = computeNextRenewWaitTime(this.latestLeaderInfo.getTimestamp(), new Date());
-        serializedExecutor.schedule(this::keepLeadershipCycle, nextDelay + FIXED_ADDITIONAL_DELAY, TimeUnit.MILLISECONDS);
-    }
-
-    /**
-     * Compute the timestamp of next event while in keep-leadership state.
-     */
-    private long computeNextRenewWaitTime(Date lastRenewal, Date lastRenewalAttempt) {
-        long timeDeadline = lastRenewal.getTime() + this.lockConfiguration.getRenewDeadlineSeconds() * 1000;
-        long timeRetry;
-        long counter = 0;
-        do {
-            counter++;
-            timeRetry = lastRenewal.getTime() + counter * this.lockConfiguration.getRetryPeriodSeconds() * 1000;
-        } while (timeRetry < lastRenewalAttempt.getTime() && timeRetry < timeDeadline);
-
-        long time = Math.min(timeRetry, timeDeadline);
-        long delay = Math.max(0, time - System.currentTimeMillis());
-        long delayJittered = jitter(delay, lockConfiguration.getJitterFactor());
-        LOG.debug("Next renewal timeout event will be fired in {} seconds", delayJittered / 1000);
-        return delayJittered;
-    }
-
-
-    /**
-     * Signals the loss of leadership and move to the acquire-leadership state.
-     */
-    private void onLeadershipLost() {
-        LOG.info("The local pod ({}) is no longer the leader of the group '{}' in ConfigMap '{}'", this.lockConfiguration.getPodName(), this.lockConfiguration.getGroupName(),
-                this.lockConfiguration.getConfigMapName());
-
-        this.eventDispatcherExecutor.execute(this::checkAndNotifyNewLeader);
-        serializedExecutor.execute(this::acquireLeadershipCycle);
-    }
-
-    /**
-     * While in the acquire-leadership state, the controller waits for the current lease to expire before trying to acquire the leadership.
-     */
-    private void acquireLeadershipCycle() {
-        // wait for the current lease to finish then fire an election
-        refreshConfigMapFromCluster(false); // if possible, update
-
-        // Notify about changes in current leader if any
-        this.eventDispatcherExecutor.execute(this::checkAndNotifyNewLeader);
-
-        if (!this.latestLeaderInfo.isTimeElapsedSeconds(lockConfiguration.getLeaseDurationSeconds())) {
-            // Wait for the lease to finish before trying leader election
-            long nextDelay = computeNextElectionWaitTime(this.latestLeaderInfo.getTimestamp());
-            serializedExecutor.schedule(this::acquireLeadershipCycle, nextDelay + FIXED_ADDITIONAL_DELAY, TimeUnit.MILLISECONDS);
-            return;
-        }
-
-        boolean acquired = tryAcquireOrRenewLeadership();
-        if (acquired) {
-            LOG.debug("Leadership acquired for ConfigMap {}. Notification in progress...", this.lockConfiguration.getConfigMapName());
-            serializedExecutor.execute(this::onLeadershipAcquired);
-            return;
-        }
-
-        // Notify about changes in current leader if any
-        this.eventDispatcherExecutor.execute(this::checkAndNotifyNewLeader);
-
-        LOG.debug("Cannot acquire the leadership for ConfigMap {}", this.lockConfiguration.getConfigMapName());
-        long nextDelay = computeNextElectionWaitTime(this.latestLeaderInfo.getTimestamp());
-        serializedExecutor.schedule(this::acquireLeadershipCycle, nextDelay + FIXED_ADDITIONAL_DELAY, TimeUnit.MILLISECONDS);
-    }
-
-    private long computeNextElectionWaitTime(Date lastRenewal) {
-        if (lastRenewal == null) {
-            LOG.debug("Error detected while getting leadership info, next election timeout event will be fired in {} seconds", this.lockConfiguration.getRetryOnErrorIntervalSeconds());
-            return this.lockConfiguration.getRetryOnErrorIntervalSeconds();
-        }
-        long time = lastRenewal.getTime() + this.lockConfiguration.getLeaseDurationSeconds() * 1000
-                + jitter(this.lockConfiguration.getRetryPeriodSeconds() * 1000, this.lockConfiguration.getJitterFactor());
-
-        long delay = Math.max(0, time - System.currentTimeMillis());
-        LOG.debug("Next election timeout event will be fired in {} seconds", delay / 1000);
-        return delay;
-    }
-
-    private long jitter(long num, double factor) {
-        return (long) (num * (1 + Math.random() * (factor - 1)));
-    }
-
-    private boolean tryAcquireOrRenewLeadership() {
-        LOG.debug("Trying to acquire or renew the leadership...");
-
-        ConfigMap configMap;
-        try {
-            configMap = pullConfigMap();
-        } catch (Exception e) {
-            LOG.warn("Unable to retrieve the current ConfigMap " + this.lockConfiguration.getConfigMapName() + " from Kubernetes", e);
-            return false;
-        }
-
-        // Info to set in the configmap to become leaders
-        LeaderInfo newLeaderInfo = new LeaderInfo(this.lockConfiguration.getGroupName(), this.lockConfiguration.getPodName(), new Date());
-
-        if (configMap == null) {
-            // No configmap created so far
-            LOG.debug("Lock configmap is not present in the Kubernetes namespace. A new ConfigMap will be created");
-            ConfigMap newConfigMap = ConfigMapLockUtils.createNewConfigMap(this.lockConfiguration.getConfigMapName(), newLeaderInfo);
-
-            try {
-                kubernetesClient.configMaps()
-                        .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
-                        .create(newConfigMap);
-            } catch (Exception ex) {
-                // Suppress exception
-                LOG.warn("Unable to create the ConfigMap, it may have been created by other cluster members concurrently. If the problem persists, check if the service account has the right "
-                        + "permissions to create it");
-                LOG.debug("Exception while trying to create the ConfigMap", ex);
-
-                // Try to get the configMap and return the current leader
-                refreshConfigMapFromCluster(false);
-                return isCurrentPodTheActiveLeader();
-            }
-
-            LOG.debug("ConfigMap {} successfully created and local pod is leader", this.lockConfiguration.getConfigMapName());
-            updateLatestLeaderInfo(newConfigMap);
-            scheduleCheckForPossibleLeadershipLoss();
-            return true;
-        } else {
-            LOG.debug("Lock configmap already present in the Kubernetes namespace. Checking...");
-            LeaderInfo leaderInfo = ConfigMapLockUtils.getLeaderInfo(configMap, this.lockConfiguration.getGroupName());
-
-            boolean weWereLeader = leaderInfo.isLeader(this.lockConfiguration.getPodName());
-            boolean leaseExpired = leaderInfo.isTimeElapsedSeconds(this.lockConfiguration.getLeaseDurationSeconds());
-
-            if (weWereLeader || leaseExpired) {
-                // Renew the lease or set the new leader
-                try {
-                    ConfigMap updatedConfigMap = ConfigMapLockUtils.getConfigMapWithNewLeader(configMap, newLeaderInfo);
-                    kubernetesClient.configMaps()
-                            .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
-                            .withName(this.lockConfiguration.getConfigMapName())
-                            .lockResourceVersion(configMap.getMetadata().getResourceVersion())
-                            .replace(updatedConfigMap);
-
-                    LOG.debug("ConfigMap {} successfully updated and local pod is leader", this.lockConfiguration.getConfigMapName());
-                    updateLatestLeaderInfo(updatedConfigMap);
-                    scheduleCheckForPossibleLeadershipLoss();
-                    return true;
-                } catch (Exception ex) {
-                    LOG.warn("An attempt to become leader has failed. It's possible that the leadership has been taken by another pod");
-                    LOG.debug("Error received during configmap lock replace", ex);
-
-                    // Try to get the configMap and return the current leader
-                    refreshConfigMapFromCluster(false);
-                    return isCurrentPodTheActiveLeader();
-                }
-            } else {
-                // Another pod is the leader and lease is not expired
-                LOG.debug("Another pod is the current leader and lease has not expired yet");
-                updateLatestLeaderInfo(configMap);
-                return false;
-            }
-        }
-    }
-
-
-    private void refreshConfigMapFromCluster(boolean retry) {
-        LOG.debug("Retrieving configmap {}", this.lockConfiguration.getConfigMapName());
-        try {
-            updateLatestLeaderInfo(pullConfigMap());
-        } catch (Exception ex) {
-            if (retry) {
-                LOG.warn("ConfigMap pull failed. Retrying in " + this.lockConfiguration.getRetryOnErrorIntervalSeconds() + " seconds...", ex);
-                try {
-                    Thread.sleep(this.lockConfiguration.getRetryOnErrorIntervalSeconds() * 1000);
-                    refreshConfigMapFromCluster(true);
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    throw new RuntimeException("Controller Thread interrupted, shutdown in progress", e);
-                }
-            } else {
-                LOG.warn("Cannot retrieve the ConfigMap: pull failed", ex);
-            }
-        }
-    }
-
-    private boolean isCurrentPodTheActiveLeader() {
-        return latestLeaderInfo != null
-                && latestLeaderInfo.isLeader(this.lockConfiguration.getPodName())
-                && !latestLeaderInfo.isTimeElapsedSeconds(this.lockConfiguration.getRenewDeadlineSeconds());
-    }
-
-    private ConfigMap pullConfigMap() {
-        return kubernetesClient.configMaps()
-                .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
-                .withName(this.lockConfiguration.getConfigMapName())
-                .get();
-    }
-
-
-    private void updateLatestLeaderInfo(ConfigMap configMap) {
-        LOG.debug("Updating internal status about the current leader");
-        this.latestLeaderInfo = ConfigMapLockUtils.getLeaderInfo(configMap, this.lockConfiguration.getGroupName());
-    }
-
-    private void scheduleCheckForPossibleLeadershipLoss() {
-        // Adding check for the case of main thread busy on http calls
-        if (this.latestLeaderInfo.isLeader(this.lockConfiguration.getPodName())) {
-            this.eventDispatcherExecutor.schedule(this::checkAndNotifyNewLeader, this.lockConfiguration.getRenewDeadlineSeconds() * 1000 + FIXED_ADDITIONAL_DELAY, TimeUnit.MILLISECONDS);
-        }
-    }
-
-    private void checkAndNotifyNewLeader() {
-        LOG.debug("Checking if the current leader has changed to notify the event handler...");
-        LeaderInfo newLeaderInfo = this.latestLeaderInfo;
-        if (newLeaderInfo == null) {
-            return;
-        }
-
-        long leaderInfoDurationSeconds = newLeaderInfo.isLeader(this.lockConfiguration.getPodName())
-                ? this.lockConfiguration.getRenewDeadlineSeconds()
-                : this.lockConfiguration.getLeaseDurationSeconds();
-
-        Optional<String> newLeader;
-        if (newLeaderInfo.getLeader() != null && !newLeaderInfo.isTimeElapsedSeconds(leaderInfoDurationSeconds)) {
-            newLeader = Optional.of(newLeaderInfo.getLeader());
-        } else {
-            newLeader = Optional.empty();
-        }
-
-        // Sending notifications in case of leader change
-        if (!newLeader.equals(this.currentLeader)) {
-            LOG.info("Current leader has changed from {} to {}. Sending notification...", this.currentLeader, newLeader);
-            this.currentLeader = newLeader;
-            eventHandler.onKubernetesClusterEvent((KubernetesClusterEvent.KubernetesClusterLeaderChangedEvent) () -> newLeader);
-        } else {
-            LOG.debug("Current leader unchanged: {}", this.currentLeader);
-        }
-    }
-
-
-}

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/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 6461708..be0b424 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
@@ -28,14 +28,10 @@ public class KubernetesLockConfiguration implements Cloneable {
 
     public static final String DEFAULT_CONFIGMAP_NAME = "leaders";
 
-
     public static final double DEFAULT_JITTER_FACTOR = 1.2;
-    public static final long DEFAULT_LEASE_DURATION_SECONDS = 60;
-    public static final long DEFAULT_RENEW_DEADLINE_SECONDS = 45;
-    public static final long DEFAULT_RETRY_PERIOD_SECONDS = 9;
-
-    public static final long DEFAULT_RETRY_ON_ERROR_INTERVAL_SECONDS = 5;
-    public static final long DEFAULT_WATCH_REFRESH_INTERVAL_SECONDS = 1800;
+    public static final long DEFAULT_LEASE_DURATION_MILLIS = 60000;
+    public static final long DEFAULT_RENEW_DEADLINE_MILLIS = 45000;
+    public static final long DEFAULT_RETRY_PERIOD_MILLIS = 9000;
 
     /**
      * Kubernetes namespace containing the pods and the ConfigMap used for locking.
@@ -63,37 +59,25 @@ public class KubernetesLockConfiguration implements Cloneable {
     private Map<String, String> clusterLabels = new HashMap<>();
 
     /**
-     * Indicates the maximum amount of time a Kubernetes watch should be kept active, before being recreated.
-     * Watch recreation can be disabled by putting value <= 0.
-     */
-    private long retryOnErrorIntervalSeconds = DEFAULT_RETRY_ON_ERROR_INTERVAL_SECONDS;
-
-    /**
-     * A jitter factor to apply in order to prevent all pods to try to become leaders in the same instant.
+     * A jitter factor to apply in order to prevent all pods to call Kubernetes APIs in the same instant.
      */
     private double jitterFactor = DEFAULT_JITTER_FACTOR;
 
     /**
      * The default duration of the lease for the current leader.
      */
-    private long leaseDurationSeconds = DEFAULT_LEASE_DURATION_SECONDS;
-
-    /**
-     * The deadline after which the leader must stop trying to renew its leadership (and yield it).
-     */
-    private long renewDeadlineSeconds = DEFAULT_RENEW_DEADLINE_SECONDS;
+    private long leaseDurationMillis = DEFAULT_LEASE_DURATION_MILLIS;
 
     /**
-     * The time between two subsequent attempts to acquire/renew the leadership (or after the lease expiration).
-     * It is randomized using the jitter factor in case of new leader election (not renewal).
+     * The deadline after which the leader must stop its services because it may have lost the leadership.
      */
-    private long retryPeriodSeconds = DEFAULT_RETRY_PERIOD_SECONDS;
+    private long renewDeadlineMillis = DEFAULT_RENEW_DEADLINE_MILLIS;
 
     /**
-     * Set this to a positive value in order to recreate watchers after a certain amount of time
-     * (to prevent them becoming stale).
+     * The time between two subsequent attempts to check and acquire the leadership.
+     * It is randomized using the jitter factor.
      */
-    private long watchRefreshIntervalSeconds = DEFAULT_WATCH_REFRESH_INTERVAL_SECONDS;
+    private long retryPeriodMillis = DEFAULT_RETRY_PERIOD_MILLIS;
 
     public KubernetesLockConfiguration() {
     }
@@ -149,14 +133,6 @@ public class KubernetesLockConfiguration implements Cloneable {
         this.clusterLabels = clusterLabels;
     }
 
-    public long getRetryOnErrorIntervalSeconds() {
-        return retryOnErrorIntervalSeconds;
-    }
-
-    public void setRetryOnErrorIntervalSeconds(long retryOnErrorIntervalSeconds) {
-        this.retryOnErrorIntervalSeconds = retryOnErrorIntervalSeconds;
-    }
-
     public double getJitterFactor() {
         return jitterFactor;
     }
@@ -165,36 +141,28 @@ public class KubernetesLockConfiguration implements Cloneable {
         this.jitterFactor = jitterFactor;
     }
 
-    public long getLeaseDurationSeconds() {
-        return leaseDurationSeconds;
-    }
-
-    public void setLeaseDurationSeconds(long leaseDurationSeconds) {
-        this.leaseDurationSeconds = leaseDurationSeconds;
-    }
-
-    public long getRenewDeadlineSeconds() {
-        return renewDeadlineSeconds;
+    public long getLeaseDurationMillis() {
+        return leaseDurationMillis;
     }
 
-    public void setRenewDeadlineSeconds(long renewDeadlineSeconds) {
-        this.renewDeadlineSeconds = renewDeadlineSeconds;
+    public void setLeaseDurationMillis(long leaseDurationMillis) {
+        this.leaseDurationMillis = leaseDurationMillis;
     }
 
-    public long getRetryPeriodSeconds() {
-        return retryPeriodSeconds;
+    public long getRenewDeadlineMillis() {
+        return renewDeadlineMillis;
     }
 
-    public void setRetryPeriodSeconds(long retryPeriodSeconds) {
-        this.retryPeriodSeconds = retryPeriodSeconds;
+    public void setRenewDeadlineMillis(long renewDeadlineMillis) {
+        this.renewDeadlineMillis = renewDeadlineMillis;
     }
 
-    public long getWatchRefreshIntervalSeconds() {
-        return watchRefreshIntervalSeconds;
+    public long getRetryPeriodMillis() {
+        return retryPeriodMillis;
     }
 
-    public void setWatchRefreshIntervalSeconds(long watchRefreshIntervalSeconds) {
-        this.watchRefreshIntervalSeconds = watchRefreshIntervalSeconds;
+    public void setRetryPeriodMillis(long retryPeriodMillis) {
+        this.retryPeriodMillis = retryPeriodMillis;
     }
 
     public KubernetesLockConfiguration copy() {
@@ -214,12 +182,10 @@ public class KubernetesLockConfiguration implements Cloneable {
         sb.append(", groupName='").append(groupName).append('\'');
         sb.append(", podName='").append(podName).append('\'');
         sb.append(", clusterLabels=").append(clusterLabels);
-        sb.append(", retryOnErrorIntervalSeconds=").append(retryOnErrorIntervalSeconds);
         sb.append(", jitterFactor=").append(jitterFactor);
-        sb.append(", leaseDurationSeconds=").append(leaseDurationSeconds);
-        sb.append(", renewDeadlineSeconds=").append(renewDeadlineSeconds);
-        sb.append(", retryPeriodSeconds=").append(retryPeriodSeconds);
-        sb.append(", watchRefreshIntervalSeconds=").append(watchRefreshIntervalSeconds);
+        sb.append(", leaseDurationMillis=").append(leaseDurationMillis);
+        sb.append(", renewDeadlineMillis=").append(renewDeadlineMillis);
+        sb.append(", retryPeriodMillis=").append(retryPeriodMillis);
         sb.append('}');
         return sb.toString();
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesMembersMonitor.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesMembersMonitor.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesMembersMonitor.java
deleted file mode 100644
index 586a11f..0000000
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/KubernetesMembersMonitor.java
+++ /dev/null
@@ -1,237 +0,0 @@
-/**
- * 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.kubernetes.ha.lock;
-
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import io.fabric8.kubernetes.api.model.Pod;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.KubernetesClientException;
-import io.fabric8.kubernetes.client.Watch;
-import io.fabric8.kubernetes.client.Watcher;
-
-import org.apache.camel.Service;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Monitors the list of participants in a leader election and provides the most recently updated list.
- * It calls the callback eventHandlers only when the member set changes w.r.t. the previous invocation.
- */
-class KubernetesMembersMonitor implements Service {
-
-    private static final Logger LOG = LoggerFactory.getLogger(KubernetesMembersMonitor.class);
-
-    private ScheduledExecutorService serializedExecutor;
-
-    private KubernetesClient kubernetesClient;
-
-    private KubernetesLockConfiguration lockConfiguration;
-
-    private List<KubernetesClusterEventHandler> eventHandlers;
-
-    private Watch watch;
-
-    private boolean terminated;
-
-    private boolean refreshing;
-
-    private Set<String> previousMembers = new HashSet<>();
-
-    private Set<String> basePoll = new HashSet<>();
-    private Set<String> deleted = new HashSet<>();
-    private Set<String> added = new HashSet<>();
-
-    public KubernetesMembersMonitor(ScheduledExecutorService serializedExecutor, KubernetesClient kubernetesClient, KubernetesLockConfiguration lockConfiguration) {
-        this.serializedExecutor = serializedExecutor;
-        this.kubernetesClient = kubernetesClient;
-        this.lockConfiguration = lockConfiguration;
-        this.eventHandlers = new LinkedList<>();
-    }
-
-    public void addClusterEventHandler(KubernetesClusterEventHandler eventHandler) {
-        this.eventHandlers.add(eventHandler);
-    }
-
-    @Override
-    public void start() throws Exception {
-        serializedExecutor.execute(() -> doPoll(true));
-        serializedExecutor.execute(this::createWatch);
-
-        long recreationDelay = lockConfiguration.getWatchRefreshIntervalSeconds();
-        if (recreationDelay > 0) {
-            serializedExecutor.scheduleWithFixedDelay(this::refresh, recreationDelay, recreationDelay, TimeUnit.SECONDS);
-        }
-    }
-
-    private void createWatch() {
-        try {
-            LOG.debug("Starting cluster members watcher");
-            this.watch = kubernetesClient.pods()
-                    .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
-                    .withLabels(this.lockConfiguration.getClusterLabels())
-                    .watch(new Watcher<Pod>() {
-
-                        @Override
-                        public void eventReceived(Action action, Pod pod) {
-                            switch (action) {
-                            case DELETED:
-                                serializedExecutor.execute(() -> deleteAndNotify(podName(pod)));
-                                break;
-                            case ADDED:
-                                serializedExecutor.execute(() -> addAndNotify(podName(pod)));
-                                break;
-                            default:
-                            }
-                        }
-
-                        @Override
-                        public void onClose(KubernetesClientException e) {
-                            if (!terminated) {
-                                KubernetesMembersMonitor.this.watch = null;
-                                if (refreshing) {
-                                    LOG.info("Refreshing pod list watcher...");
-                                    serializedExecutor.execute(KubernetesMembersMonitor.this::createWatch);
-                                } else {
-                                    LOG.warn("Pod list watcher has been closed unexpectedly. Recreating it in 1 second...", e);
-                                    serializedExecutor.schedule(KubernetesMembersMonitor.this::createWatch, 1, TimeUnit.SECONDS);
-                                }
-                            }
-                        }
-                    });
-        } catch (Exception ex) {
-            LOG.warn("Unable to watch for pod list changes. Retrying in 5 seconds...");
-            LOG.debug("Error while trying to watch the pod list", ex);
-
-            serializedExecutor.schedule(this::createWatch, 5, TimeUnit.SECONDS);
-        }
-    }
-
-    @Override
-    public void stop() throws Exception {
-        this.terminated = true;
-        Watch watch = this.watch;
-        if (watch != null) {
-            watch.close();
-        }
-    }
-
-    public void refresh() {
-        serializedExecutor.execute(() -> {
-            if (!terminated) {
-                refreshing = true;
-                try {
-                    doPoll(false);
-
-                    Watch w = this.watch;
-                    if (w != null) {
-                        // It will be recreated
-                        w.close();
-                    }
-                } finally {
-                    refreshing = false;
-                }
-            }
-        });
-    }
-
-    private void doPoll(boolean retry) {
-        LOG.debug("Starting poll to get all cluster members");
-        List<Pod> pods;
-        try {
-            pods = pollPods();
-        } catch (Exception ex) {
-            if (retry) {
-                LOG.warn("Pods poll failed. Retrying in 5 seconds...", ex);
-                this.serializedExecutor.schedule(() -> doPoll(true), 5, TimeUnit.SECONDS);
-            } else {
-                LOG.warn("Pods poll failed", ex);
-            }
-            return;
-        }
-
-        this.basePoll = pods.stream()
-                .map(p -> Optional.ofNullable(podName(p)))
-                .filter(Optional::isPresent)
-                .map(Optional::get)
-                .collect(Collectors.toSet());
-
-        this.added = new HashSet<>();
-        this.deleted = new HashSet<>();
-
-        LOG.debug("Base list of members is {}", this.basePoll);
-
-        checkAndNotify();
-    }
-
-    private List<Pod> pollPods() {
-        return kubernetesClient.pods()
-                .inNamespace(this.lockConfiguration.getKubernetesResourcesNamespaceOrDefault(kubernetesClient))
-                .withLabels(this.lockConfiguration.getClusterLabels())
-                .list().getItems();
-    }
-
-    private String podName(Pod pod) {
-        if (pod != null && pod.getMetadata() != null) {
-            return pod.getMetadata().getName();
-        }
-        return null;
-    }
-
-    private void checkAndNotify() {
-        Set<String> newMembers = new HashSet<>(basePoll);
-        newMembers.addAll(added);
-        newMembers.removeAll(deleted);
-
-        LOG.debug("Current list of members is: {}", newMembers);
-
-        if (!newMembers.equals(this.previousMembers)) {
-            LOG.debug("List of members changed: sending notifications");
-            this.previousMembers = newMembers;
-
-            for (KubernetesClusterEventHandler eventHandler : eventHandlers) {
-                eventHandler.onKubernetesClusterEvent((KubernetesClusterEvent.KubernetesClusterMemberListChangedEvent) () -> newMembers);
-            }
-        } else {
-            LOG.debug("List of members has not changed");
-        }
-    }
-
-    private void addAndNotify(String member) {
-        LOG.debug("Adding new member to the list: {}", member);
-        if (member != null) {
-            this.added.add(member);
-            checkAndNotify();
-        }
-    }
-
-    private void deleteAndNotify(String member) {
-        LOG.debug("Deleting member to the list: {}", member);
-        if (member != null) {
-            this.deleted.add(member);
-            checkAndNotify();
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/LeaderInfo.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/LeaderInfo.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/LeaderInfo.java
index 50d1603..d061945 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/LeaderInfo.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/LeaderInfo.java
@@ -17,6 +17,7 @@
 package org.apache.camel.component.kubernetes.ha.lock;
 
 import java.util.Date;
+import java.util.Set;
 
 import org.apache.camel.util.ObjectHelper;
 
@@ -29,28 +30,31 @@ public class LeaderInfo {
 
     private String leader;
 
-    private Date timestamp;
+    private Date localTimestamp;
+
+    private Set<String> members;
 
     public LeaderInfo() {
     }
 
-    public LeaderInfo(String groupName, String leader, Date timestamp) {
+    public LeaderInfo(String groupName, String leader, Date timestamp, Set<String> members) {
         this.groupName = groupName;
         this.leader = leader;
-        this.timestamp = timestamp;
+        this.localTimestamp = timestamp;
+        this.members = members;
+    }
+
+    public boolean hasEmptyLeader() {
+        return this.leader == null;
     }
 
-    public boolean isTimeElapsedSeconds(long timeSeconds) {
-        if (timestamp == null) {
-            return true;
-        }
-        long now = System.currentTimeMillis();
-        return timestamp.getTime() + timeSeconds * 1000 <= now;
+    public boolean hasValidLeader() {
+        return this.leader != null && this.members.contains(this.leader);
     }
 
-    public boolean isLeader(String pod) {
+    public boolean isValidLeader(String pod) {
         ObjectHelper.notNull(pod, "pod");
-        return pod.equals(leader);
+        return hasValidLeader() && pod.equals(leader);
     }
 
     public String getGroupName() {
@@ -69,12 +73,20 @@ public class LeaderInfo {
         this.leader = leader;
     }
 
-    public Date getTimestamp() {
-        return timestamp;
+    public Date getLocalTimestamp() {
+        return localTimestamp;
+    }
+
+    public void setLocalTimestamp(Date localTimestamp) {
+        this.localTimestamp = localTimestamp;
+    }
+
+    public Set<String> getMembers() {
+        return members;
     }
 
-    public void setTimestamp(Date timestamp) {
-        this.timestamp = timestamp;
+    public void setMembers(Set<String> members) {
+        this.members = members;
     }
 
     @Override
@@ -82,9 +94,9 @@ public class LeaderInfo {
         final StringBuilder sb = new StringBuilder("LeaderInfo{");
         sb.append("groupName='").append(groupName).append('\'');
         sb.append(", leader='").append(leader).append('\'');
-        sb.append(", timestamp=").append(timestamp);
+        sb.append(", localTimestamp=").append(localTimestamp);
+        sb.append(", members=").append(members);
         sb.append('}');
         return sb.toString();
     }
-
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/TimedLeaderNotifier.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/TimedLeaderNotifier.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/TimedLeaderNotifier.java
new file mode 100644
index 0000000..6ada830
--- /dev/null
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/ha/lock/TimedLeaderNotifier.java
@@ -0,0 +1,179 @@
+/**
+ * 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.kubernetes.ha.lock;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.camel.Service;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Receives information about the current leader and handles expiration automatically.
+ */
+public class TimedLeaderNotifier implements Service {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TimedLeaderNotifier.class);
+
+    private static final long FIXED_DELAY = 10;
+
+    private KubernetesClusterEventHandler handler;
+
+    private Lock lock = new ReentrantLock();
+
+    private ScheduledExecutorService executor;
+
+    private Optional<String> lastCommunicatedLeader = Optional.empty();
+    private Set<String> lastCommunicatedMembers = Collections.emptySet();
+
+    private Optional<String> currentLeader = Optional.empty();
+
+    private Set<String> currentMembers;
+
+    private Long timestamp;
+
+    private Long lease;
+
+    private long changeCounter;
+
+    public TimedLeaderNotifier(KubernetesClusterEventHandler handler) {
+        this.handler = Objects.requireNonNull(handler, "Handler must be present");
+    }
+
+    public void refreshLeadership(Optional<String> leader, Long timestamp, Long lease, Set<String> members) {
+        Objects.requireNonNull(leader, "leader must be non null (use Optional.empty)");
+        Objects.requireNonNull(members, "members must be non null (use empty set)");
+        long version;
+        try {
+            lock.lock();
+
+            this.currentLeader = leader;
+            this.currentMembers = members;
+            this.timestamp = timestamp;
+            this.lease = lease;
+            version = ++changeCounter;
+        } finally {
+            lock.unlock();
+        }
+
+        LOG.debug("Updated leader to {} at version version {}", leader, version);
+        this.executor.execute(() -> checkAndNotify(version));
+        if (leader.isPresent()) {
+            long time = System.currentTimeMillis();
+            long delay = Math.max(timestamp + lease + FIXED_DELAY - time, FIXED_DELAY);
+            LOG.debug("Setting expiration in {} millis for version {}", delay, version);
+            this.executor.schedule(() -> expiration(version), delay, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    @Override
+    public void start() throws Exception {
+        if (this.executor == null) {
+            this.executor = Executors.newSingleThreadScheduledExecutor();
+        }
+    }
+
+    @Override
+    public void stop() throws Exception {
+        if (this.executor != null) {
+            ScheduledExecutorService executor = this.executor;
+            this.executor = null;
+
+            executor.shutdownNow();
+            executor.awaitTermination(1, TimeUnit.SECONDS);
+        }
+    }
+
+    private void expiration(long version) {
+        try {
+            lock.lock();
+
+            if (version != this.changeCounter) {
+                return;
+            }
+
+            long time = System.currentTimeMillis();
+            if (time < this.timestamp + this.lease) {
+                long delay = this.timestamp + this.lease - time;
+                LOG.debug("Delaying expiration by {} millis at version version {}", delay + FIXED_DELAY, version);
+                this.executor.schedule(() -> expiration(version), delay + FIXED_DELAY, TimeUnit.MILLISECONDS);
+                return;
+            }
+        } finally {
+            lock.unlock();
+        }
+
+        checkAndNotify(version);
+    }
+
+    private void checkAndNotify(long version) {
+        Optional<String> leader;
+        Set<String> members;
+        try {
+            lock.lock();
+
+            if (version != this.changeCounter) {
+                return;
+            }
+
+            leader = this.currentLeader;
+            members = this.currentMembers;
+
+            if (leader.isPresent()) {
+                long time = System.currentTimeMillis();
+                if (time >= this.timestamp + this.lease) {
+                    leader = Optional.empty();
+                }
+            }
+
+        } finally {
+            lock.unlock();
+        }
+
+        final Optional<String> newLeader = leader;
+        if (!newLeader.equals(lastCommunicatedLeader)) {
+            lastCommunicatedLeader = newLeader;
+            handler.onKubernetesClusterEvent(new KubernetesClusterEvent.KubernetesClusterLeaderChangedEvent() {
+                @Override
+                public Optional<String> getData() {
+                    return newLeader;
+                }
+            });
+        }
+
+        final Set<String> newMembers = members;
+        if (!newMembers.equals(lastCommunicatedMembers)) {
+            lastCommunicatedMembers = newMembers;
+            handler.onKubernetesClusterEvent(new KubernetesClusterEvent.KubernetesClusterMemberListChangedEvent() {
+                @Override
+                public Set<String> getData() {
+                    return newMembers;
+                }
+            });
+        }
+
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/f0b00ab9/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterServiceTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterServiceTest.java b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterServiceTest.java
index 3bdffbd..4a2a11e 100644
--- a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterServiceTest.java
+++ b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/ha/KubernetesClusterServiceTest.java
@@ -42,7 +42,10 @@ import org.junit.Test;
  */
 public class KubernetesClusterServiceTest extends CamelTestSupport {
 
-    private static final int LEASE_TIME_SECONDS = 5;
+    private static final int LEASE_TIME_MILLIS = 2000;
+    private static final int RENEW_DEADLINE_MILLIS = 1000;
+    private static final int RETRY_PERIOD_MILLIS = 200;
+    private static final double JITTER_FACTOR = 1.1;
 
     private ConfigMapLockSimulator lockSimulator;
 
@@ -75,6 +78,7 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
         mypod2.waitForAnyLeader(2, TimeUnit.SECONDS);
 
         String leader = mypod1.getCurrentLeader();
+        assertNotNull(leader);
         assertTrue(leader.startsWith("mypod"));
         assertEquals("Leaders should be equals", mypod2.getCurrentLeader(), leader);
     }
@@ -129,6 +133,7 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
         LeaderRecorder formerLoserRecorder = firstLeader.equals("mypod1") ? mypod2 : mypod1;
 
         refuseRequestsFromPod(firstLeader);
+        disconnectPod(firstLeader);
 
         formerLeaderRecorder.waitForALeaderChange(7, TimeUnit.SECONDS);
         formerLoserRecorder.waitForANewLeader(firstLeader, 7, TimeUnit.SECONDS);
@@ -139,12 +144,12 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
         Long lossTimestamp = formerLeaderRecorder.getLastTimeOf(l -> l == null);
         Long gainTimestamp = formerLoserRecorder.getLastTimeOf(secondLeader::equals);
 
-        assertTrue("At least 2 seconds must elapse from leadership loss and regain (see renewDeadlineSeconds)", gainTimestamp >= lossTimestamp + 2000);
-        checkLeadershipChangeDistance(LEASE_TIME_SECONDS, TimeUnit.SECONDS, mypod1, mypod2);
+        assertTrue("At least half distance must elapse from leadership loss and regain (see renewDeadlineSeconds)", gainTimestamp >= lossTimestamp + (LEASE_TIME_MILLIS - RENEW_DEADLINE_MILLIS) / 2);
+        checkLeadershipChangeDistance((LEASE_TIME_MILLIS - RENEW_DEADLINE_MILLIS) / 2, TimeUnit.MILLISECONDS, mypod1, mypod2);
     }
 
     @Test
-    public void testSlowLeaderLosingLeadership() throws Exception {
+    public void testSlowLeaderLosingLeadershipOnlyInternally() throws Exception {
         LeaderRecorder mypod1 = addMember("mypod1");
         LeaderRecorder mypod2 = addMember("mypod2");
         context.start();
@@ -159,17 +164,9 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
 
         delayRequestsFromPod(firstLeader, 10, TimeUnit.SECONDS);
 
-        formerLeaderRecorder.waitForALeaderChange(7, TimeUnit.SECONDS);
-        formerLoserRecorder.waitForANewLeader(firstLeader, 7, TimeUnit.SECONDS);
-
-        String secondLeader = formerLoserRecorder.getCurrentLeader();
-        assertNotEquals("The firstLeader should be different from the new one", firstLeader, secondLeader);
-
-        Long lossTimestamp = formerLeaderRecorder.getLastTimeOf(l -> l == null);
-        Long gainTimestamp = formerLoserRecorder.getLastTimeOf(secondLeader::equals);
-
-        assertTrue("At least 2 seconds must elapse from leadership loss and regain (see renewDeadlineSeconds)", gainTimestamp >= lossTimestamp + 2000);
-        checkLeadershipChangeDistance(LEASE_TIME_SECONDS, TimeUnit.SECONDS, mypod1, mypod2);
+        Thread.sleep(LEASE_TIME_MILLIS);
+        assertNull(formerLeaderRecorder.getCurrentLeader());
+        assertEquals(firstLeader, formerLoserRecorder.getCurrentLeader());
     }
 
     @Test
@@ -185,9 +182,9 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
 
         for (int i = 0; i < 3; i++) {
             refuseRequestsFromPod(firstLeader);
-            Thread.sleep(1000);
+            Thread.sleep(RENEW_DEADLINE_MILLIS);
             allowRequestsFromPod(firstLeader);
-            Thread.sleep(2000);
+            Thread.sleep(LEASE_TIME_MILLIS);
         }
 
         assertEquals(firstLeader, mypod1.getCurrentLeader());
@@ -229,6 +226,18 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
         this.lockServers.get(pod).setRefuseRequests(false);
     }
 
+    private void disconnectPod(String pod) {
+        for (LockTestServer server : this.lockServers.values()) {
+            server.removePod(pod);
+        }
+    }
+
+    private void connectPod(String pod) {
+        for (LockTestServer server : this.lockServers.values()) {
+            server.addPod(pod);
+        }
+    }
+
     private void checkLeadershipChangeDistance(long minimum, TimeUnit unit, LeaderRecorder... recorders) {
         List<LeaderRecorder.LeadershipInfo> infos = Arrays.stream(recorders)
                 .flatMap(lr -> lr.getLeadershipInfo().stream())
@@ -245,7 +254,8 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
                 } else if (info.getLeader() != null && !info.getLeader().equals(currentLeaderLastSeen.getLeader())) {
                     // switch
                     long delay = info.getChangeTimestamp() - currentLeaderLastSeen.getChangeTimestamp();
-                    assertTrue("Lease time not elapsed between switch", delay >= TimeUnit.MILLISECONDS.convert(minimum, unit));
+                    assertTrue("Lease time not elapsed between switch, minimum=" + TimeUnit.MILLISECONDS.convert(minimum, unit) + ", found=" + delay, delay >= TimeUnit.MILLISECONDS.convert(minimum,
+                            unit));
                     currentLeaderLastSeen = info;
                 }
             }
@@ -268,11 +278,10 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
         KubernetesClusterService member = new KubernetesClusterService(configuration);
         member.setKubernetesNamespace("test");
         member.setPodName(name);
-        member.setLeaseDurationSeconds(LEASE_TIME_SECONDS);
-        member.setRenewDeadlineSeconds(3); // 5-3 = at least 2 seconds for switching on leadership loss
-        member.setRetryPeriodSeconds(1);
-        member.setRetryOnErrorIntervalSeconds(1);
-        member.setJitterFactor(1.2);
+        member.setLeaseDurationMillis(LEASE_TIME_MILLIS);
+        member.setRenewDeadlineMillis(RENEW_DEADLINE_MILLIS);
+        member.setRetryPeriodMillis(RETRY_PERIOD_MILLIS);
+        member.setJitterFactor(JITTER_FACTOR);
 
         LeaderRecorder recorder = new LeaderRecorder();
         try {
@@ -281,6 +290,10 @@ public class KubernetesClusterServiceTest extends CamelTestSupport {
         } catch (Exception ex) {
             throw new RuntimeException(ex);
         }
+
+        for (String pod : this.lockServers.keySet()) {
+            connectPod(pod);
+        }
         return recorder;
     }
 


Mime
View raw message