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 4fe61e034b8b CAMEL-26784: camel-hazelcast - HazelcastRoutePolicy
releases the leadership when its route is removed and keeps a follower stopped
on resume (#29532)
4fe61e034b8b is described below
commit 4fe61e034b8ba7fa2161fc71af9e7163f0719137
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 10 18:39:21 2026 +0530
CAMEL-26784: camel-hazelcast - HazelcastRoutePolicy releases the leadership
when its route is removed and keeps a follower stopped on resume (#29532)
HazelcastRoutePolicy did not override onRemove, so a route removed or
reloaded on the leader kept the Hazelcast lock and the route then ran
nowhere
in the cluster. It did not override onResume either, so a resume on a
follower started its consumer next to the leader's. The policy now counts
its
routes and the removal of the last one ends the leadership (the thread waits
on a latch and releases the lock without being interrupted, so a client
instance can unlock), onInit starts a leadership thread for a route added
again, and onResume keeps a follower's consumer stopped. "Leadership lost"
is
now logged (the condition was inverted).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../hazelcast/policy/HazelcastRoutePolicy.java | 199 ++++++++++--
.../HazelcastRoutePolicyRouteLifecycleTest.java | 360 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
3 files changed, 538 insertions(+), 29 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 0956bcb8abac..9bf14d299004 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
@@ -16,8 +16,11 @@
*/
package org.apache.camel.component.hazelcast.policy;
+import java.util.ArrayList;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -43,8 +46,13 @@ public class HazelcastRoutePolicy extends RoutePolicySupport
implements CamelCon
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;
- private Route route;
+ // a route of the policy, for JMX
+ private volatile Route route;
private CamelContext camelContext;
private ExecutorService executorService;
private HazelcastInstance instance;
@@ -55,6 +63,8 @@ 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;
private boolean shouldStopConsumer;
public HazelcastRoutePolicy() {
@@ -69,6 +79,9 @@ public class HazelcastRoutePolicy extends RoutePolicySupport
implements CamelCon
this.instance = instance;
this.managedInstance = managedInstance;
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;
@@ -93,7 +106,34 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
@Override
public void onInit(Route route) {
super.onInit(route);
- this.route = route;
+
+ lock.lock();
+ try {
+ addRoute(route);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void onRemove(Route route) {
+ lock.lock();
+ try {
+ suspendedRoutes.remove(route);
+ suspendedWhileStopped.remove(route);
+ if (routes.remove(route)) {
+ if (this.route == route) {
+ this.route = routes.isEmpty() ? null :
routes.iterator().next();
+ }
+ if (routes.isEmpty()) {
+ // no route uses this policy any more: give up the
leadership, so another node runs the route. The
+ // thread is only interrupted when the CamelContext stops
(as the stop of the policy does)
+ stopLeadership(camelContext != null &&
camelContext.isStopping());
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
}
@Override
@@ -108,6 +148,7 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
lock.lock();
try {
suspendedRoutes.remove(route);
+ suspendedWhileStopped.remove(route);
} finally {
lock.unlock();
}
@@ -117,7 +158,31 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
public void onSuspend(Route route) {
lock.lock();
try {
- suspendedRoutes.remove(route);
+ if (suspendedRoutes.remove(route)) {
+ suspendedWhileStopped.add(route);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void onResume(Route route) {
+ lock.lock();
+ try {
+ boolean stoppedByPolicy = suspendedWhileStopped.remove(route);
+ if (!leader.get() && shouldStopConsumer) {
+ // resuming the route resumed its consumer, which a follower
keeps stopped (as when the route starts)
+ stopConsumer(route);
+ } else if (stoppedByPolicy) {
+ // 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());
+ stopConsumer(route.getConsumer());
+ startConsumer(route.getConsumer());
+ }
+ } catch (Exception e) {
+ handleException(e);
} finally {
lock.unlock();
}
@@ -133,17 +198,19 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
executorService =
getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
"HazelcastRoutePolicy");
locks = instance.getMap(lockMapName);
- future = executorService.submit(this::acquireLeadership);
+ // under the lock: BaseService.start() holds it (the same lock as the
route callbacks of the policy)
+ if (!routes.isEmpty()) {
+ startLeadership();
+ }
super.doStart();
}
@Override
protected void doStop() throws Exception {
- if (future != null) {
- future.cancel(true);
- future = null;
- }
+ // 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);
if (managedInstance) {
instance.shutdown();
@@ -157,6 +224,49 @@ 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
+ 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
+ private void stopLeadership(boolean interrupt) {
+ if (leadershipRelease != null) {
+ leadershipRelease.countDown();
+ leadershipRelease = null;
+ }
+ if (future != null) {
+ // a thread that waits for the lock ends after the tryLock
timeout, or (on a Hazelcast client) when it is
+ // interrupted
+ releasedFutures.add(future);
+ future = null;
+ }
+ releasedFutures.removeIf(Future::isDone);
+ if (interrupt) {
+ releasedFutures.forEach(f -> f.cancel(true));
+ releasedFutures.clear();
+ }
+ setLeader(false);
+ }
+
protected void setLeader(boolean isLeader) {
if (isLeader && leader.compareAndSet(false, isLeader)) {
LOGGER.info("Leadership taken (map={}, key={}, val={})",
@@ -165,13 +275,11 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
lockValue);
startAllStoppedConsumers();
- } else {
- if (!leader.getAndSet(isLeader) && isLeader) {
- LOGGER.info("Leadership lost (map={}, key={} val={})",
- lockMapName,
- lockKey,
- lockValue);
- }
+ } else if (!isLeader && leader.getAndSet(false)) {
+ LOGGER.info("Leadership lost (map={}, key={}, val={})",
+ lockMapName,
+ lockKey,
+ lockValue);
}
}
@@ -212,16 +320,18 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
@ManagedAttribute(description = "The route id")
public String getRouteId() {
- if (route != null) {
- return route.getId();
+ Route r = route;
+ if (r != null) {
+ return r.getId();
}
return null;
}
@ManagedAttribute(description = "The consumer endpoint", mask = true)
public String getEndpointUrl() {
- if (route != null && route.getConsumer() != null &&
route.getConsumer().getEndpoint() != null) {
- return route.getConsumer().getEndpoint().toString();
+ Route r = route;
+ if (r != null && r.getConsumer() != null &&
r.getConsumer().getEndpoint() != null) {
+ return r.getConsumer().getEndpoint().toString();
}
return null;
}
@@ -294,18 +404,18 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
//
//
*************************************************************************
- private Void acquireLeadership() throws Exception {
+ private Void acquireLeadership(CountDownLatch release) {
boolean locked = false;
- while (isRunAllowed()) {
+ while (isRunAllowed() && release.getCount() > 0) {
try {
locked = locks.tryLock(lockKey, tryLockTimeout,
tryLockTimeoutUnit);
- if (locked) {
+ if (locked && release.getCount() > 0) {
locks.put(lockKey, lockValue);
- setLeader(true);
-
- // Wait almost forever
- Thread.sleep(Long.MAX_VALUE);
- } else {
+ if (takeLeadership(release)) {
+ // hold the lock until the policy stops or its routes
are removed
+ release.await();
+ }
+ } else if (!locked) {
LOGGER.debug("Failed to acquire lock (map={}, key={},
val={}) after {} {}",
lockMapName,
lockKey,
@@ -319,8 +429,7 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
getExceptionHandler().handleException(e);
} finally {
if (locked) {
- locks.remove(lockKey);
- locks.unlock(lockKey);
+ releaseLock();
locked = false;
}
@@ -330,4 +439,36 @@ public class HazelcastRoutePolicy extends
RoutePolicySupport implements CamelCon
return null;
}
+
+ 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();
+ }
+ }
+
+ private void releaseLock() {
+ // with the interrupt flag cleared: on a client instance an
interrupted thread fails to remove and unlock
+ boolean interrupted = Thread.interrupted();
+ try {
+ try {
+ locks.remove(lockKey);
+ } finally {
+ locks.unlock(lockKey);
+ }
+ } catch (Exception e) {
+ getExceptionHandler().handleException(e);
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
}
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicyRouteLifecycleTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicyRouteLifecycleTest.java
new file mode 100644
index 000000000000..44ae27d53825
--- /dev/null
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/policy/HazelcastRoutePolicyRouteLifecycleTest.java
@@ -0,0 +1,360 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.hazelcast.policy;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import com.hazelcast.client.HazelcastClient;
+import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.cluster.Address;
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+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.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * {@link HazelcastRoutePolicy} on two nodes (two CamelContexts on one
embedded Hazelcast member, the lock is held by
+ * the leadership thread of a node) when a route is removed, added again,
suspended and resumed.
+ */
+class HazelcastRoutePolicyRouteLifecycleTest {
+
+ private static final String ROUTE_ID = "clustered";
+
+ private final List<CamelContext> contexts = new ArrayList<>();
+ private HazelcastInstance member;
+ private HazelcastInstance client;
+
+ @BeforeEach
+ void startMember() {
+ member = Hazelcast.newHazelcastInstance(memberConfig("route-policy-" +
UUID.randomUUID()));
+ }
+
+ @AfterEach
+ void stopNodes() {
+ contexts.forEach(CamelContext::stop);
+ if (client != null) {
+ client.shutdown();
+ }
+ if (member != null) {
+ member.shutdown();
+ }
+ }
+
+ @Test
+ void removedRouteGivesUpTheLeadership() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ a.getRouteController().stopRoute(ROUTE_ID);
+ a.removeRoute(ROUTE_ID);
+
+ // the route runs nowhere on node a any more, so node b must take over
+ awaitLeader(b, policyB);
+ assertFalse(policyA.isLeader(), "The policy of the removed route is
still the leader");
+ assertConsumes(b);
+ }
+
+ @Test
+ void removedRouteGivesUpTheLeadershipWithAClientInstance() throws
Exception {
+ // node a uses a Hazelcast client: its thread must release the lock
(an interrupted client call fails)
+ HazelcastRoutePolicy policyA = newPolicy(newClient(), "a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ a.getRouteController().stopRoute(ROUTE_ID);
+ a.removeRoute(ROUTE_ID);
+
+ awaitLeader(b, policyB);
+ assertFalse(policyA.isLeader(), "The policy of the removed route is
still the leader");
+ assertConsumes(b);
+ }
+
+ @Test
+ void policySharedByTwoRoutesKeepsTheLeadershipUntilBothAreRemoved() throws
Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ a.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:other").routeId("other").routePolicy(policyA).to("mock:other");
+ }
+ });
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ // the policy still has a route on node a
+ a.getRouteController().stopRoute("other");
+ a.removeRoute("other");
+ // JMX shows a route that still uses the policy
+ assertEquals(ROUTE_ID, policyA.getRouteId());
+ await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ assertTrue(policyA.isLeader(), "Node a gave up the
leadership while a route still uses the policy");
+ assertFalse(policyB.isLeader(), "Node b took the
leadership while node a has a route");
+ });
+ assertConsumes(a);
+
+ a.getRouteController().stopRoute(ROUTE_ID);
+ a.removeRoute(ROUTE_ID);
+ assertNull(policyA.getRouteId());
+ awaitLeader(b, policyB);
+ assertConsumes(b);
+ }
+
+ @Test
+ void routeReloadedWithANewPolicyRunsOnOneNode() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ // a reload on the leader: the route is removed and added again, with
a new policy
+ a.getRouteController().stopRoute(ROUTE_ID);
+ a.removeRoute(ROUTE_ID);
+ HazelcastRoutePolicy policyA2 = newPolicy("a");
+ a.addRoutes(route(policyA2));
+
+ awaitOneLeader(a, policyA2, b, policyB);
+ assertFalse(policyA.isLeader(), "The policy of the removed route is
still the leader");
+ }
+
+ @Test
+ void routeAddedAgainWithTheSamePolicyRunsOnOneNode() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ // the same policy instance (for example a bean referenced by the
route) for the route added again
+ a.getRouteController().stopRoute(ROUTE_ID);
+ a.removeRoute(ROUTE_ID);
+ a.addRoutes(route(policyA));
+
+ awaitOneLeader(a, policyA, b, policyB);
+ }
+
+ @Test
+ void resumedRouteOnAFollowerKeepsItsConsumerStopped() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ b.getRouteController().suspendRoute(ROUTE_ID);
+ b.getRouteController().resumeRoute(ROUTE_ID);
+
+ assertFalse(consumerStarted(b), "The consumer of the follower runs
after the route was resumed");
+ assertFalse(policyB.isLeader());
+
+ // and the follower still starts its consumer when it takes over
+ a.stop();
+ awaitLeader(b, policyB);
+ assertConsumes(b);
+ }
+
+ @Test
+ void routeResumedAfterTheFollowerTookOverConsumes() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ // the route of the follower is suspended when the follower takes over
+ b.getRouteController().suspendRoute(ROUTE_ID);
+ a.stop();
+ await().atMost(20, TimeUnit.SECONDS).until(policyB::isLeader);
+
+ b.getRouteController().resumeRoute(ROUTE_ID);
+ assertTrue(consumerStarted(b), "The consumer of the leader is not
started after the route was resumed");
+ assertConsumes(b);
+ }
+
+ @Test
+ void restartedNodeCompetesForTheLeadershipAgain() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ HazelcastRoutePolicy policyB = newPolicy("b");
+ CamelContext b = startNode("b", policyB);
+ assertFollower(b, policyB);
+
+ // the stop of the CamelContext removes the route from the policy
(onRemove), and the start gives it back
+ a.stop();
+ awaitLeader(b, policyB);
+ a.start();
+ assertFollower(a, policyA);
+
+ // node a waits for the lock again, so it takes over when node b stops
+ b.stop();
+ awaitLeader(a, policyA);
+ assertConsumes(a);
+ }
+
+ @Test
+ void stopAfterTheRouteWasRemovedDoesNotWaitForTheLock() throws Exception {
+ HazelcastRoutePolicy policyA = newPolicy("a");
+ CamelContext a = startNode("a", policyA);
+ awaitLeader(a, policyA);
+ // node b uses a Hazelcast client, whose wait in tryLock ends when the
thread is interrupted (on a member it
+ // ends after the tryLock timeout)
+ HazelcastRoutePolicy policyB = newPolicy(newClient(), "b");
+ // the follower waits a long time for the lock, longer than the
graceful shutdown of its thread pool
+ policyB.setTryLockTimeout(60, TimeUnit.SECONDS);
+ CamelContext b = startNode("b", policyB);
+
b.getExecutorServiceManager().setShutdownAwaitTermination(TimeUnit.SECONDS.toMillis(30));
+ assertFollower(b, policyB);
+
+ // the route is removed while the leadership thread of node b waits
for the lock
+ b.getRouteController().stopRoute(ROUTE_ID);
+ b.removeRoute(ROUTE_ID);
+
+ // the stop of the policy ends that thread too, instead of waiting for
the end of its tryLock
+ long start = System.nanoTime();
+ b.stop();
+ long took = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertTrue(took < TimeUnit.SECONDS.toMillis(10), "The stop of node b
took " + took + " ms");
+ }
+
+ private HazelcastRoutePolicy newPolicy(String node) {
+ return newPolicy(member, node);
+ }
+
+ private static HazelcastRoutePolicy newPolicy(HazelcastInstance instance,
String node) {
+ HazelcastRoutePolicy policy = new HazelcastRoutePolicy(instance);
+ policy.setLockMapName("camel-route-policy");
+ policy.setLockKey("route-lock");
+ policy.setLockValue(node);
+ policy.setTryLockTimeout(1, TimeUnit.SECONDS);
+ return policy;
+ }
+
+ private CamelContext startNode(String node, HazelcastRoutePolicy policy)
throws Exception {
+ DefaultCamelContext context = new DefaultCamelContext();
+ context.disableJMX();
+ context.getCamelContextExtension().setName(node);
+ contexts.add(context);
+ context.addRoutes(route(policy));
+ context.start();
+ return context;
+ }
+
+ private static RouteBuilder route(HazelcastRoutePolicy policy) {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:in").routeId(ROUTE_ID).routePolicy(policy).to("mock:out");
+ }
+ };
+ }
+
+ private static boolean consumerStarted(CamelContext context) {
+ Route route = context.getRoute(ROUTE_ID);
+ return route != null && ServiceHelper.isStarted(route.getConsumer());
+ }
+
+ private static void awaitLeader(CamelContext context, HazelcastRoutePolicy
policy) {
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ assertTrue(policy.isLeader(), "Node " + context.getName()
+ " is not the leader");
+ assertTrue(consumerStarted(context), "The consumer of the
leader " + context.getName() + " is not started");
+ });
+ }
+
+ private static void assertFollower(CamelContext context,
HazelcastRoutePolicy policy) {
+ assertFalse(policy.isLeader(), "Node " + context.getName() + " is the
leader");
+ assertFalse(consumerStarted(context), "The consumer of the follower "
+ context.getName() + " is started");
+ }
+
+ private static void awaitOneLeader(
+ CamelContext a, HazelcastRoutePolicy policyA, CamelContext b,
HazelcastRoutePolicy policyB)
+ throws InterruptedException {
+ await().atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ assertNotEquals(policyA.isLeader(), policyB.isLeader(),
"Exactly one node must be the leader");
+ assertTrue(consumerStarted(policyA.isLeader() ? a : b),
"The consumer of the leader is not started");
+ assertFalse(consumerStarted(policyA.isLeader() ? b : a),
"The consumer of the follower is started");
+ });
+ assertConsumes(policyA.isLeader() ? a : b);
+ }
+
+ private static void assertConsumes(CamelContext context) throws
InterruptedException {
+ MockEndpoint mock = context.getEndpoint("mock:out",
MockEndpoint.class);
+ mock.reset();
+ mock.setResultWaitTime(TimeUnit.SECONDS.toMillis(20));
+ mock.expectedBodiesReceived("hello");
+
context.createFluentProducerTemplate().to("seda:in").withBody("hello").send();
+ mock.assertIsSatisfied();
+ }
+
+ private HazelcastInstance newClient() {
+ Address address = member.getCluster().getLocalMember().getAddress();
+ ClientConfig config = new ClientConfig();
+ config.setClusterName(member.getConfig().getClusterName());
+ config.getNetworkConfig().addAddress(address.getHost() + ":" +
address.getPort());
+
config.getConnectionStrategyConfig().getConnectionRetryConfig().setClusterConnectTimeoutMillis(30000);
+ client = HazelcastClient.newHazelcastClient(config);
+ return client;
+ }
+
+ private static Config memberConfig(String clusterName) {
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ // a client of the test connects to the member on the loopback address
+ config.setProperty("hazelcast.local.localAddress", "127.0.0.1");
+
config.getNetworkConfig().getInterfaces().setEnabled(true).addInterface("127.0.0.1");
+ config.getMetricsConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+
config.getNetworkConfig().getJoin().getAutoDetectionConfig().setEnabled(false);
+ config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(false);
+ return config;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index bc04938d7a7a..f96092f0c261 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2895,6 +2895,14 @@ for another reason, such as a destroyed topic, the
consumer reports it to its ex
default); restart the route to receive messages again. A listener cancelled
because the Hazelcast instance or the
CamelContext is shutting down is only logged at DEBUG.
+=== camel-hazelcast - HazelcastRoutePolicy gives up the leadership when its
route is removed
+
+A `HazelcastRoutePolicy` kept the Hazelcast lock until the CamelContext
stopped, also after its route was removed (with
+`removeRoute`, or by a route reload), so no other node could become the leader
and the route ran nowhere in the
+cluster. The policy now releases the lock when the last route that uses it is
removed, so another node takes over, and
+it competes for the lock again when a route that uses it is added again (with
the same or a new policy instance). So a
+route reloaded on the leader can now move to another node.
+
=== camel-sql, camel-mybatis, camel-jooq - an exchange marked rollback only
last is not consumed either
The SQL, MyBatis and jOOQ consumers already left a message unconsumed when the
route marked the exchange rollback
