This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 4fdb9f62d8b8 CAMEL-24545: Camel Shutdown Locked State With
ZookeeprCluster Service
4fdb9f62d8b8 is described below
commit 4fdb9f62d8b8ef42010382e71c519df6fc653474
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Aug 31 22:23:23 2026 +0200
CAMEL-24545: Camel Shutdown Locked State With ZookeeprCluster Service
Backport of #25842. During shutdown, ClusteredRoutePolicy.onRemove and
the Curator leader selector thread could acquire the same two locks in
opposite order (AB-BA deadlock), because ZooKeeperClusterView.take
Leadership kept firing a leadership-changed event while the view was
stopping/stopped, even though the listener had already been removed
and no longer needed notifying.
Guard the event with isStoppingOrStopped() so it only fires during
normal leadership loss, not during shutdown. Add an integration test
asserting the exact number of leadership-changed events pushed around
a ZooKeeper outage and shutdown.
Co-authored-by: Mustafa Kamal ALmonayer
<[email protected]>
Closes #25980
---
.../zookeeper/cluster/ZooKeeperClusterView.java | 4 ++-
.../ZooKeeperClusterViewLeadershipLostIT.java | 33 ++++++++++++++++++----
2 files changed, 31 insertions(+), 6 deletions(-)
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
index 594134bb4207..d2d9f0676ec2 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
@@ -151,7 +151,9 @@ final class ZooKeeperClusterView extends
AbstractCamelClusterView {
task.run(getCamelContext(), () -> !isRunAllowed());
} finally {
leader = false;
- fireLeadershipChangedEvent((CamelClusterMember) null);
+ if (!isStoppingOrStopped()) {
+ fireLeadershipChangedEvent((CamelClusterMember) null);
+ }
}
}
}
diff --git
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
index f23b136da743..69ba30dc3b2d 100644
---
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
+++
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
@@ -17,8 +17,10 @@
package org.apache.camel.component.zookeeper.cluster.integration;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.cluster.CamelClusterEventListener;
import org.apache.camel.cluster.CamelClusterView;
import org.apache.camel.component.zookeeper.cluster.ZooKeeperClusterService;
import org.apache.camel.impl.DefaultCamelContext;
@@ -38,12 +40,14 @@ class ZooKeeperClusterViewLeadershipLostIT {
private static final String NAMESPACE = "my-ns";
private static final String ROUTE_ID = "clustered-route";
+ private static final int EXPECTED_PUSHED_EVENTS_NUM = 3;
@RegisterExtension
static ZooKeeperService service = ZooKeeperServiceFactory.createService();
@Test
void leadershipIsReleasedAndReacquiredAroundAZooKeeperOutage() throws
Exception {
+ AtomicInteger numberOfLeadershipChangedPushed = new AtomicInteger();
GenericContainer<?> zooKeeper = zooKeeperContainer();
try (DefaultCamelContext context = new DefaultCamelContext()) {
@@ -66,14 +70,16 @@ class ZooKeeperClusterViewLeadershipLostIT {
}
});
- context.start();
+ CamelClusterView clusterView = clusterService.getView(NAMESPACE);
+
clusterView.addEventListener((CamelClusterEventListener.Leadership) (
+ view, leader) ->
numberOfLeadershipChangedPushed.incrementAndGet());
- CamelClusterView view = clusterService.getView(NAMESPACE);
+ context.start();
await().atMost(1, TimeUnit.MINUTES)
.untilAsserted(() -> {
assertEquals(true,
- view.getLocalMember().isLeader(),
+ clusterView.getLocalMember().isLeader(),
"the only node of the cluster must be the
leader");
assertEquals(true,
context.getRouteController().getRouteStatus(ROUTE_ID).isStarted(),
@@ -86,7 +92,7 @@ class ZooKeeperClusterViewLeadershipLostIT {
await().atMost(1, TimeUnit.MINUTES)
.untilAsserted(() -> {
assertEquals(false,
- view.getLocalMember().isLeader(),
+ clusterView.getLocalMember().isLeader(),
"the leadership must be given up once
ZooKeeper is no longer reachable");
assertEquals(false,
context.getRouteController().getRouteStatus(ROUTE_ID).isStarted(),
@@ -99,12 +105,29 @@ class ZooKeeperClusterViewLeadershipLostIT {
await().atMost(1, TimeUnit.MINUTES)
.untilAsserted(() -> {
assertEquals(true,
- view.getLocalMember().isLeader(),
+ clusterView.getLocalMember().isLeader(),
"the node must re-enter the election once
ZooKeeper is reachable again");
assertEquals(true,
context.getRouteController().getRouteStatus(ROUTE_ID).isStarted(),
"the clustered route must be restarted once
the leadership is taken back");
});
+ clusterView.stop();
+
+ /*
+ Give some time so the event can be consumed
+ (the correct behavior is that an event shouldn't be pushed)
+ this is just a safeguard so that if an event is pushed it has some
time to be consumed
+ */
+ await()
+ .pollDelay(1, TimeUnit.SECONDS)
+ .atLeast(1, TimeUnit.SECONDS)
+ .atMost(2, TimeUnit.SECONDS)
+ .until(() -> true);
+
+ assertEquals(EXPECTED_PUSHED_EVENTS_NUM,
+ numberOfLeadershipChangedPushed.get(),
+ "the pushed Leadership Changed event must be %d otherwise
a push happened on stop view"
+ .formatted(EXPECTED_PUSHED_EVENTS_NUM));
}
}
