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 647f10537076 CAMEL-24545: Camel Shutdown Locked State With
ZookeeprCluster Service
647f10537076 is described below
commit 647f10537076d236f46cf0336d2d7b17d2e38ed3
Author: Mustafa Kamal ALmonayer
<[email protected]>
AuthorDate: Mon Aug 31 22:37:19 2026 +0300
CAMEL-24545: Camel Shutdown Locked State With ZookeeprCluster Service
Fix a deadlock during Camel shutdown when using camel-zookeeper cluster
service. ClusteredRoutePolicy.onRemove and the Curator leader selector
thread could acquire the same two locks in opposite order (AB-BA
deadlock) because ZooKeeperClusterView.takeLeadership still fired 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, and add an
integration test asserting the exact number of leadership-changed
events pushed around a ZooKeeper outage and shutdown.
Closes #25842
---
.../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));
}
}
