This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new b1fbaa979314 CAMEL-26798: camel-hazelcast - HazelcastRoutePolicy
retries a failed consumer start and does not start the consumer of a stopping
route (#29585)
b1fbaa979314 is described below
commit b1fbaa979314a750d8e6555bb01e544bf6c45a8e
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 11 12:27:06 2026 +0530
CAMEL-26798: camel-hazelcast - HazelcastRoutePolicy retries a failed
consumer start and does not start the consumer of a stopping route (#29585)
* CAMEL-26798: camel-hazelcast - HazelcastRoutePolicy retries a failed
consumer start and does not start the consumer of a stopping route
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../hazelcast/policy/HazelcastRoutePolicy.java | 287 +++++++++----
.../policy/HazelcastRoutePolicyLeadershipTest.java | 452 +++++++++++++++++++++
2 files changed, 666 insertions(+), 73 deletions(-)
diff --git
a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicy.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicy.java
index 9bf14d299004..4cc630527656 100644
---
a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicy.java
+++
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicy.java
@@ -17,14 +17,18 @@
package org.apache.camel.component.hazelcast.policy;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;
@@ -35,6 +39,7 @@ import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.component.hazelcast.HazelcastUtil;
import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.StringHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,14 +47,22 @@ import org.slf4j.LoggerFactory;
@ManagedResource(description = "Route policy using Hazelcast as clustered
lock")
public class HazelcastRoutePolicy extends RoutePolicySupport implements
CamelContextAware {
private static final Logger LOGGER =
LoggerFactory.getLogger(HazelcastRoutePolicy.class);
+ // delay before the leader tries again to start a consumer whose start
failed
+ private static final long RETRY_DELAY_MILLIS = 5000;
private final boolean managedInstance;
private final AtomicBoolean leader;
private final Set<Route> suspendedRoutes;
private final Set<Route> routes;
private final Set<Route> suspendedWhileStopped;
- // the leadership threads told to end that may still wait for or release
the lock (guarded by lock)
- private final List<Future<Void>> releasedFutures;
+ // failed consumer starts in a row per route, for the log
+ private final Map<Route, Integer> failedStarts = new HashMap<>();
+ // guards the leader flag, the route sets and the leadership thread, so a
leadership taken while a route starts,
+ // stops or resumes is not missed. Not the lifecycle lock of the service:
stop() holds that one while it waits for
+ // the leadership thread
+ private final Lock leadershipLock = new ReentrantLock();
+ // the leadership threads told to end that may still wait for or release
the lock (guarded by leadershipLock)
+ private final List<Future<Void>> releasedFutures = new ArrayList<>();
// a route of the policy, for JMX
private volatile Route route;
@@ -63,8 +76,12 @@ public class HazelcastRoutePolicy extends RoutePolicySupport
implements CamelCon
private TimeUnit tryLockTimeoutUnit;
private IMap<String, String> locks;
private volatile Future<Void> future;
- // counted down to make the leadership thread release the lock and end
(guarded by lock)
- private CountDownLatch leadershipRelease;
+ // counted down to make the leadership thread release the lock and end
(set under leadershipLock, also read
+ // without it)
+ private volatile CountDownLatch leadershipRelease;
+ // the latch of the leadership task that runs (and may hold the cluster
lock)
+ private volatile CountDownLatch runningRelease;
+ private volatile long retryDelay = RETRY_DELAY_MILLIS;
private boolean shouldStopConsumer;
public HazelcastRoutePolicy() {
@@ -81,7 +98,6 @@ public class HazelcastRoutePolicy extends RoutePolicySupport
implements CamelCon
this.suspendedRoutes = new HashSet<>();
this.routes = new HashSet<>();
this.suspendedWhileStopped = new HashSet<>();
- this.releasedFutures = new ArrayList<>();
this.leader = new AtomicBoolean();
this.lockMapName = null;
this.lockKey = null;
@@ -107,20 +123,36 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
public void onInit(Route route) {
super.onInit(route);
+ // the lifecycle lock, so the start of the policy is not in progress
lock.lock();
+ leadershipLock.lock();
try {
- addRoute(route);
+ if (routes.add(route)) {
+ if (this.route == null) {
+ this.route = route;
+ }
+ // a route added to a policy that gave up the leadership when
its routes were removed
+ if (leadershipRelease == null && isStarted()) {
+ startLeadership();
+ }
+ }
} finally {
+ leadershipLock.unlock();
lock.unlock();
}
}
@Override
public void onRemove(Route route) {
- lock.lock();
+ if (camelContext != null && camelContext.isStopping()) {
+ // all the routes go: end the leadership first, so a consumer
start in flight does not delay the stop
+ interruptLeadership();
+ }
+ leadershipLock.lock();
try {
suspendedRoutes.remove(route);
suspendedWhileStopped.remove(route);
+ failedStarts.remove(route);
if (routes.remove(route)) {
if (this.route == route) {
this.route = routes.isEmpty() ? null :
routes.iterator().next();
@@ -132,43 +164,54 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
}
}
} finally {
- lock.unlock();
+ leadershipLock.unlock();
}
}
@Override
public void onStart(Route route) {
- if (!leader.get() && shouldStopConsumer) {
- stopConsumer(route);
+ // the flag is read under the lock that setLeader holds while it sets
it and starts the stopped consumers
+ leadershipLock.lock();
+ try {
+ if (!leader.get() && shouldStopConsumer) {
+ stopConsumer(route);
+ } else if (leader.get() &&
ServiceHelper.isStopped(route.getConsumer())) {
+ // a start by the leadership thread that overlapped the start
of the route stopped this consumer again
+ // (the route status was not started yet): the leadership
thread starts it
+ suspendedRoutes.add(route);
+ }
+ } finally {
+ leadershipLock.unlock();
}
}
@Override
public void onStop(Route route) {
- lock.lock();
+ leadershipLock.lock();
try {
suspendedRoutes.remove(route);
suspendedWhileStopped.remove(route);
+ failedStarts.remove(route);
} finally {
- lock.unlock();
+ leadershipLock.unlock();
}
}
@Override
public void onSuspend(Route route) {
- lock.lock();
+ leadershipLock.lock();
try {
if (suspendedRoutes.remove(route)) {
suspendedWhileStopped.add(route);
}
} finally {
- lock.unlock();
+ leadershipLock.unlock();
}
}
@Override
public void onResume(Route route) {
- lock.lock();
+ leadershipLock.lock();
try {
boolean stoppedByPolicy = suspendedWhileStopped.remove(route);
if (!leader.get() && shouldStopConsumer) {
@@ -178,13 +221,16 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
// the leadership was taken while the route was suspended:
resuming a consumer that this policy had
// stopped does not start it again (a SEDA consumer is marked
started without consuming), so restart it
LOGGER.debug("Restarting consumer for {} ({})", route.getId(),
route.getConsumer());
+ // the leadership thread tries again if the start fails
+ suspendedRoutes.add(route);
stopConsumer(route.getConsumer());
startConsumer(route.getConsumer());
+ suspendedRoutes.remove(route);
}
} catch (Exception e) {
handleException(e);
} finally {
- lock.unlock();
+ leadershipLock.unlock();
}
}
@@ -198,9 +244,13 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
executorService =
getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
"HazelcastRoutePolicy");
locks = instance.getMap(lockMapName);
- // under the lock: BaseService.start() holds it (the same lock as the
route callbacks of the policy)
- if (!routes.isEmpty()) {
- startLeadership();
+ leadershipLock.lock();
+ try {
+ if (!routes.isEmpty()) {
+ startLeadership();
+ }
+ } finally {
+ leadershipLock.unlock();
}
super.doStart();
@@ -208,9 +258,15 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
@Override
protected void doStop() throws Exception {
- // also interrupt the threads, so a wait for the lock does not delay
the stop on a Hazelcast client (on a member
- // tryLock ends after its timeout). Under the lock, which
BaseService.stop() holds
- stopLeadership(true);
+ // also interrupt the threads first, so a consumer start in flight, or
a wait for the lock on a Hazelcast client
+ // (on a member tryLock ends after its timeout), does not delay the
stop
+ interruptLeadership();
+ leadershipLock.lock();
+ try {
+ stopLeadership(true);
+ } finally {
+ leadershipLock.unlock();
+ }
if (managedInstance) {
instance.shutdown();
@@ -224,30 +280,29 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
//
//
*************************************************************************
- // under the lock
- private void addRoute(Route route) {
- if (routes.add(route)) {
- if (this.route == null) {
- this.route = route;
- }
- // a route added to a policy that gave up the leadership when its
routes were removed
- if (leadershipRelease == null && isStarted()) {
- startLeadership();
- }
- }
- }
-
- // under the lock
+ // under the leadership lock
private void startLeadership() {
CountDownLatch release = new CountDownLatch(1);
leadershipRelease = release;
future = executorService.submit(() -> acquireLeadership(release));
}
- // under the lock: the leadership thread releases the lock and ends, and
this node is no longer the leader. The
- // thread is told to end with its own latch, as isRunAllowed() stays true
while the policy runs, and is not
- // interrupted unless the policy stops: an interrupted thread cannot
release the lock on a client instance, and a
- // wait for the lock ends by itself after the tryLock timeout
+ // without the leadership lock: tells the leadership thread to end and
interrupts it (when stopping)
+ private void interruptLeadership() {
+ CountDownLatch release = leadershipRelease;
+ if (release != null) {
+ release.countDown();
+ }
+ Future<Void> task = future;
+ if (task != null) {
+ task.cancel(true);
+ }
+ }
+
+ // under the leadership lock: the leadership thread releases the lock and
ends, and this node is no longer the
+ // leader. The thread is told to end with its own latch, as isRunAllowed()
stays true while the policy runs, and is
+ // not interrupted unless the policy stops: an interrupted thread cannot
release the lock on a client instance, and
+ // a wait for the lock ends by itself after the tryLock timeout
private void stopLeadership(boolean interrupt) {
if (leadershipRelease != null) {
leadershipRelease.countDown();
@@ -268,23 +323,49 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
}
protected void setLeader(boolean isLeader) {
- if (isLeader && leader.compareAndSet(false, isLeader)) {
- LOGGER.info("Leadership taken (map={}, key={}, val={})",
- lockMapName,
- lockKey,
- lockValue);
-
- startAllStoppedConsumers();
- } else if (!isLeader && leader.getAndSet(false)) {
- LOGGER.info("Leadership lost (map={}, key={}, val={})",
- lockMapName,
- lockKey,
- lockValue);
+ if (isLeader) {
+ List<Route> toStart;
+ leadershipLock.lock();
+ try {
+ toStart = markLeader();
+ } finally {
+ leadershipLock.unlock();
+ }
+ startStoppedConsumers(toStart);
+ } else {
+ leadershipLock.lock();
+ try {
+ if (leader.compareAndSet(true, false)) {
+ LOGGER.info("Leadership lost (map={}, key={}, val={})",
+ lockMapName,
+ lockKey,
+ lockValue);
+ }
+ } finally {
+ leadershipLock.unlock();
+ }
+ }
+ }
+
+ // under the leadership lock: sets the leader flag, and returns the routes
whose consumer this policy stopped. Not
+ // when the routes were removed meanwhile (stopLeadership), so the
leadership is not taken after that
+ private List<Route> markLeader() {
+ // the latch of the running task, not only the current one: a task
that ended its leadership (its routes were
+ // removed) must not take it for a route added meanwhile, as it
releases the cluster lock
+ CountDownLatch release = leadershipRelease;
+ if (release == null || release.getCount() == 0 || release !=
runningRelease
+ || !leader.compareAndSet(false, true)) {
+ return List.of();
}
+ LOGGER.info("Leadership taken (map={}, key={}, val={})",
+ lockMapName,
+ lockKey,
+ lockValue);
+ return new ArrayList<>(suspendedRoutes);
}
+ // under the leadership lock
private void stopConsumer(Route route) {
- lock.lock();
try {
if (!suspendedRoutes.contains(route)) {
LOGGER.debug("Stopping consumer for {} ({})", route.getId(),
route.getConsumer());
@@ -293,25 +374,87 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
}
} catch (Exception e) {
handleException(e);
- } finally {
- lock.unlock();
}
}
- private void startAllStoppedConsumers() {
- lock.lock();
- try {
- for (Route route : suspendedRoutes) {
+ // Starts the consumers of the routes this policy stopped, without the
leadership lock: a consumer start can take
+ // long (a broker connect), and the route callbacks and the stop must not
wait for it. Each start is checked again
+ // under the lock: a route is forgotten only once its consumer started (so
the leader tries again a start that
+ // failed), and a consumer started for a route that is stopping or
suspending meanwhile is stopped again.
+ private void startStoppedConsumers(List<Route> toStart) {
+ for (Route route : toStart) {
+ if (!isStartWanted(route)) {
+ continue;
+ }
+ Exception cause = null;
+ try {
LOGGER.debug("Starting consumer for {} ({})", route.getId(),
route.getConsumer());
startConsumer(route.getConsumer());
+ } catch (Exception e) {
+ cause = e;
}
+ afterStart(route, cause);
+ }
+ }
+
+ private boolean isStartWanted(Route route) {
+ leadershipLock.lock();
+ try {
+ return isRunAllowed() && leader.get() &&
suspendedRoutes.contains(route) && isResumeOrStartConsumerAllowed(route);
+ } finally {
+ leadershipLock.unlock();
+ }
+ }
- suspendedRoutes.clear();
+ private void afterStart(Route route, Exception cause) {
+ leadershipLock.lock();
+ try {
+ boolean wanted = isRunAllowed() && leader.get() &&
isResumeOrStartConsumerAllowed(route);
+ if (cause != null) {
+ if (!wanted) {
+ LOGGER.debug("Failed to start consumer for {} while it is
no longer wanted", route.getId(), cause);
+ return;
+ }
+ int attempt = failedStarts.merge(route, 1, Integer::sum);
+ getExceptionHandler().handleException("Failed to start
consumer for route " + route.getId()
+ + " (attempt #" +
attempt + "). Trying again in " + retryDelay
+ + " millis.",
+ cause);
+ } else if (!wanted) {
+ // the route is stopping or suspending (or this node is
stopping): its consumer must not run
+ LOGGER.debug("Stopping consumer for {} again, as the route is
{}", route.getId(),
+
route.getCamelContext().getRouteController().getRouteStatus(route.getId()));
+ stopConsumer(route.getConsumer());
+ } else if (suspendedRoutes.remove(route)) {
+ Integer failed = failedStarts.remove(route);
+ if (failed != null) {
+ LOGGER.info("Consumer for route {} started after {} failed
attempts", route.getId(), failed);
+ }
+ }
} catch (Exception e) {
handleException(e);
} finally {
- lock.unlock();
+ leadershipLock.unlock();
+ }
+ }
+
+ // the leader starts the consumers whose start failed
+ private void retryStoppedConsumers() {
+ List<Route> toStart;
+ leadershipLock.lock();
+ try {
+ toStart = leader.get() ? new ArrayList<>(suspendedRoutes) :
List.of();
+ } finally {
+ leadershipLock.unlock();
}
+ startStoppedConsumers(toStart);
+ }
+
+ /**
+ * The delay before the leader tries again to start a consumer whose start
failed (5 seconds). For tests.
+ */
+ void setRetryDelay(long retryDelay) {
+ this.retryDelay = retryDelay;
}
//
*************************************************************************
@@ -405,6 +548,7 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
//
*************************************************************************
private Void acquireLeadership(CountDownLatch release) {
+ runningRelease = release;
boolean locked = false;
while (isRunAllowed() && release.getCount() > 0) {
try {
@@ -412,8 +556,11 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
if (locked && release.getCount() > 0) {
locks.put(lockKey, lockValue);
if (takeLeadership(release)) {
- // hold the lock until the policy stops or its routes
are removed
- release.await();
+ // hold the lock until the policy stops or its routes
are removed, and meanwhile try again to
+ // start the consumers whose start failed
+ while (!release.await(retryDelay,
TimeUnit.MILLISECONDS)) {
+ retryStoppedConsumers();
+ }
}
} else if (!locked) {
LOGGER.debug("Failed to acquire lock (map={}, key={},
val={}) after {} {}",
@@ -441,17 +588,11 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
}
private boolean takeLeadership(CountDownLatch release) {
- // under the lock, so the leadership is not taken after the routes
were removed (stopLeadership)
- lock.lock();
- try {
- if (release.getCount() == 0) {
- return false;
- }
- setLeader(true);
- return true;
- } finally {
- lock.unlock();
+ if (release.getCount() == 0) {
+ return false;
}
+ setLeader(true);
+ return true;
}
private void releaseLock() {
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicyLeadershipTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicyLeadershipTest.java
new file mode 100644
index 000000000000..bb5307475adc
--- /dev/null
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicyLeadershipTest.java
@@ -0,0 +1,452 @@
+/*
+ * 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.hazelcast.policy;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.IMap;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.support.service.ServiceHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The leadership taken by the policy's thread against the route start, the
route stop, a consumer start that fails and
+ * the policy stop, with an embedded member. The other node is the test
thread, which holds the cluster lock.
+ */
+class HazelcastRoutePolicyLeadershipTest {
+
+ private static final String LOCK_MAP = "route-policy-locks";
+ private static final String LOCK_KEY = "route-lock";
+
+ private HazelcastInstance member;
+ private CamelContext context;
+
+ @BeforeEach
+ void startMember() {
+ Config config = new Config();
+ config.setClusterName("route-policy-" + UUID.randomUUID());
+ config.getMetricsConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getAutoDetectionConfig().setEnabled(false);
+ config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(false);
+ member = Hazelcast.newHazelcastInstance(config);
+ }
+
+ @AfterEach
+ void stop() {
+ if (context != null) {
+ // the other node goes away first, so the stop does not wait for
the tryLock of the policy's thread
+ if (member != null) {
+ member.getMap(LOCK_MAP).forceUnlock(LOCK_KEY);
+ }
+ context.stop();
+ }
+ if (member != null) {
+ member.shutdown();
+ }
+ }
+
+ @Test
+ void leadershipTakenWhileTheRouteStartsIsNotLost() throws Exception {
+ // another node holds the cluster lock, so the policy's own thread
waits in tryLock during the test
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ TestPolicy policy = newPolicy(TimeUnit.MINUTES.toMillis(1));
+ startContext(policy);
+
+ // holds the lock that onStart on main waited for after it read the
flag, and takes the leadership meanwhile.
+ // The fixed onStart does not take this lock, so this checks the
ordered case; the race test covers the
+ // atomicity
+ ReentrantLock held = (ReentrantLock) policy.lifecycleLock();
+ held.lock();
+ CompletableFuture<Void> routeStart;
+ try {
+ routeStart = CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().startRoute("route");
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ });
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
held.hasQueuedThreads() || routeStart.isDone());
+
+ // meanwhile this node takes the leadership: acquireLeadership
calls setLeader(true) once it holds the
+ // cluster lock, which starts the consumers the policy has stopped
(none yet)
+ policy.setLeader(true);
+ } finally {
+ held.unlock();
+ }
+ routeStart.get(20, TimeUnit.SECONDS);
+
+ assertTrue(policy.isLeader());
+
assertTrue(ServiceHelper.isStarted(context.getRoute("route").getConsumer()),
+ "The node that holds the leadership must consume");
+ }
+
+ @Test
+ void leadershipTakenWhileRoutesStartNeverLeavesTheLeaderIdle() throws
Exception {
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ TestPolicy policy = newPolicy(TimeUnit.MINUTES.toMillis(1));
+ startContext(policy);
+ // a started route: the policy does not start the consumer of a route
that is stopped
+ context.getRouteController().startRoute("route");
+ Route route = context.getRoute("route");
+
+ // the route start (onStart) and the leadership thread
(setLeader(true)) race, no hooks
+ for (int i = 0; i < 1000; i++) {
+ ServiceHelper.startService(route.getConsumer());
+ policy.onStop(route);
+ policy.setLeader(false);
+
+ CyclicBarrier barrier = new CyclicBarrier(2);
+ long spinNanos = ThreadLocalRandom.current().nextLong(2000);
+ CompletableFuture<Void> routeStart = CompletableFuture.runAsync(()
-> {
+ awaitBarrier(barrier);
+ policy.onStart(route);
+ });
+ CompletableFuture<Void> leadership = CompletableFuture.runAsync(()
-> {
+ awaitBarrier(barrier);
+ long end = System.nanoTime() + spinNanos;
+ while (System.nanoTime() < end) {
+ Thread.onSpinWait();
+ }
+ policy.setLeader(true);
+ });
+ CompletableFuture.allOf(routeStart, leadership).get(20,
TimeUnit.SECONDS);
+
+ assertTrue(ServiceHelper.isStarted(route.getConsumer()),
+ "The node that holds the leadership must consume (round "
+ i + ")");
+ }
+ }
+
+ @Test
+ void stopWhileAConsumerStartsAtTakeoverDoesNotWait() throws Exception {
+ // another node holds the cluster lock, so the route starts on a
follower and the policy stops its consumer
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ // the consumer start at takeover blocks (a broker connect) until it
is interrupted
+ CountDownLatch starting = new CountDownLatch(1);
+ CountDownLatch neverReleased = new CountDownLatch(1);
+ TestPolicy policy = newPolicy(TimeUnit.SECONDS.toMillis(1));
+ startContext(policy);
+ context.getRouteController().startRoute("route");
+
assertFalse(ServiceHelper.isStarted(context.getRoute("route").getConsumer()));
+ policy.beforeStart = () -> {
+ starting.countDown();
+ try {
+ neverReleased.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while starting",
e);
+ }
+ };
+
+ // the other node goes away: this node takes the leadership and starts
the consumer
+ locks.unlock(LOCK_KEY);
+ assertTrue(starting.await(20, TimeUnit.SECONDS), "The consumer start
did not begin");
+
+ // the policy stops (the CamelContext shuts down) meanwhile: it must
not wait for the consumer start
+ CompletableFuture<Void> stop =
CompletableFuture.runAsync(policy::stop);
+ try {
+ await().atMost(5, TimeUnit.SECONDS).until(stop::isDone);
+ } finally {
+ neverReleased.countDown();
+ }
+ stop.get(20, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void consumerWhoseStartFailsAtTakeoverIsStartedAgain() throws Exception {
+ // another node holds the cluster lock, so the routes start on a
follower and the policy stops their consumers
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ AtomicBoolean firstFails = new AtomicBoolean();
+ AtomicInteger firstAttempts = new AtomicInteger();
+ TestPolicy policy = newPolicy(TimeUnit.SECONDS.toMillis(1));
+ policy.setRetryDelay(100);
+ context = new DefaultCamelContext();
+ context.addComponent("failing", new FailingStartComponent(firstFails,
firstAttempts));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("failing:first").routeId("first").routePolicy(policy).to("log:first");
+
from("failing:second").routeId("second").routePolicy(policy).to("log:second");
+ }
+ });
+ context.start();
+
assertFalse(ServiceHelper.isStarted(context.getRoute("first").getConsumer()));
+
assertFalse(ServiceHelper.isStarted(context.getRoute("second").getConsumer()));
+
+ // the other node goes away, and the consumer of the first route
cannot start (a broker not reachable yet)
+ firstFails.set(true);
+ firstAttempts.set(0);
+ locks.unlock(LOCK_KEY);
+
+ // the other consumer starts, and the failed one is tried again
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ assertTrue(policy.isLeader(), "This node is not the
leader");
+
assertTrue(ServiceHelper.isStarted(context.getRoute("second").getConsumer()),
+ "The consumer of the second route did not start");
+ assertTrue(firstAttempts.get() >= 2, "The start of the
first consumer was not tried again");
+ });
+
assertFalse(ServiceHelper.isStarted(context.getRoute("first").getConsumer()));
+
+ // and starts once it can
+ firstFails.set(false);
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertTrue(ServiceHelper.isStarted(context.getRoute("first").getConsumer()),
+ "The leader does not consume from the first route"));
+ }
+
+ @Test
+ void consumerOfARouteBeingStoppedIsNotStartedAtTakeover() throws Exception
{
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ TestPolicy policy = newPolicy(TimeUnit.SECONDS.toMillis(1));
+ AtomicBoolean taken = new AtomicBoolean();
+ // listed before the Hazelcast policy, so its onStop runs while the
route stops, before the one of the policy
+ RoutePolicySupport takeoverWhileStopping = new RoutePolicySupport() {
+ @Override
+ public void onStop(Route route) {
+ if (taken.compareAndSet(false, true)) {
+ // the other node goes away now: this node takes the
leadership and starts the consumers it stopped
+ locks.unlock(LOCK_KEY);
+ await().atMost(20,
TimeUnit.SECONDS).until(policy::isLeader);
+ }
+ }
+ };
+ context = new DefaultCamelContext();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").routeId("route").routePolicy(takeoverWhileStopping,
policy).to("log:route");
+ }
+ });
+ context.start();
+ Route route = context.getRoute("route");
+ assertFalse(ServiceHelper.isStarted(route.getConsumer()));
+
+ context.getRouteController().stopRoute("route");
+
+ assertTrue(policy.isLeader());
+ await().during(500, TimeUnit.MILLISECONDS).atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertFalse(ServiceHelper.isStarted(route.getConsumer()),
+ "The consumer of the stopped route runs"));
+
+ // and it is started with the route
+ context.getRouteController().startRoute("route");
+
assertTrue(ServiceHelper.isStarted(context.getRoute("route").getConsumer()));
+ }
+
+ @Test
+ void consumerOfARouteBeingSuspendedIsNotStartedAtTakeover() throws
Exception {
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ TestPolicy policy = newPolicy(TimeUnit.SECONDS.toMillis(1));
+ AtomicBoolean taken = new AtomicBoolean();
+ // listed before the Hazelcast policy, so its onSuspend runs while the
route suspends, before the one of the
+ // policy
+ RoutePolicySupport takeoverWhileSuspending = new RoutePolicySupport() {
+ @Override
+ public void onSuspend(Route route) {
+ if (taken.compareAndSet(false, true)) {
+ locks.unlock(LOCK_KEY);
+ await().atMost(20,
TimeUnit.SECONDS).until(policy::isLeader);
+ }
+ }
+ };
+ context = new DefaultCamelContext();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:start").routeId("route").routePolicy(takeoverWhileSuspending,
policy).to("log:route");
+ }
+ });
+ context.start();
+ Route route = context.getRoute("route");
+ assertFalse(ServiceHelper.isStarted(route.getConsumer()));
+
+ context.getRouteController().suspendRoute("route");
+
+ assertTrue(policy.isLeader());
+ await().during(500, TimeUnit.MILLISECONDS).atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertFalse(ServiceHelper.isStarted(route.getConsumer()),
+ "The consumer of the suspended route runs"));
+
+ // and it is started when the route is resumed
+ context.getRouteController().resumeRoute("route");
+ assertTrue(ServiceHelper.isStarted(route.getConsumer()));
+ }
+
+ @Test
+ void consumerStartedWhileItsRouteStopsIsStoppedAgain() throws Exception {
+ IMap<String, String> locks = member.getMap(LOCK_MAP);
+ locks.lock(LOCK_KEY);
+
+ TestPolicy policy = newPolicy(TimeUnit.SECONDS.toMillis(1));
+ startContext(policy);
+ context.getRouteController().startRoute("route");
+ Route route = context.getRoute("route");
+ assertFalse(ServiceHelper.isStarted(route.getConsumer()));
+ // the route is stopped after the policy checked it and before it
starts the consumer
+ policy.beforeStart = () -> {
+ try {
+ context.getRouteController().stopRoute("route");
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ };
+
+ locks.unlock(LOCK_KEY);
+
+ await().atMost(20, TimeUnit.SECONDS).until(policy::isLeader);
+ await().during(500, TimeUnit.MILLISECONDS).atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+
assertTrue(context.getRouteController().getRouteStatus("route").isStopped(),
"The route is not stopped");
+ assertFalse(ServiceHelper.isStarted(route.getConsumer()),
"The consumer of the stopped route runs");
+ });
+ }
+
+ private TestPolicy newPolicy(long tryLockTimeoutMillis) {
+ TestPolicy policy = new TestPolicy(member);
+ policy.setLockMapName(LOCK_MAP);
+ policy.setLockKey(LOCK_KEY);
+ policy.setLockValue("this-node");
+ policy.setTryLockTimeout(tryLockTimeoutMillis, TimeUnit.MILLISECONDS);
+ return policy;
+ }
+
+ private void startContext(TestPolicy policy) throws Exception {
+ if (context == null) {
+ context = new DefaultCamelContext();
+ }
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").routeId("route").autoStartup(false).routePolicy(policy).to("log:route");
+ }
+ });
+ context.start();
+ }
+
+ private static void awaitBarrier(CyclicBarrier barrier) {
+ try {
+ barrier.await(20, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static final class TestPolicy extends HazelcastRoutePolicy {
+ // runs before a consumer is started by the policy (when this node
takes the leadership)
+ volatile Runnable beforeStart;
+
+ TestPolicy(HazelcastInstance instance) {
+ super(instance);
+ }
+
+ Lock lifecycleLock() {
+ return lock;
+ }
+
+ @Override
+ public void startConsumer(Consumer consumer) throws Exception {
+ Runnable hook = beforeStart;
+ if (hook != null) {
+ beforeStart = null;
+ hook.run();
+ }
+ super.startConsumer(consumer);
+ }
+ }
+
+ // the consumer of the endpoint "first" fails to start while firstFails is
set
+ private static final class FailingStartComponent extends DefaultComponent {
+ private final AtomicBoolean firstFails;
+ private final AtomicInteger firstAttempts;
+
+ FailingStartComponent(AtomicBoolean firstFails, AtomicInteger
firstAttempts) {
+ this.firstFails = firstFails;
+ this.firstAttempts = firstAttempts;
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ boolean first = "first".equals(remaining);
+ return new DefaultEndpoint(uri, this) {
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException("No producer");
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ return new DefaultConsumer(this, processor) {
+ @Override
+ protected void doStart() throws Exception {
+ if (first) {
+ firstAttempts.incrementAndGet();
+ if (firstFails.get()) {
+ throw new IllegalStateException("Cannot
start the consumer yet");
+ }
+ }
+ super.doStart();
+ }
+ };
+ }
+ };
+ }
+ }
+}
