This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/25842-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 65fb7a78740d5db200fead8b7504382c0395d45a 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)); } }
