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 8fc0dc2f36bb CAMEL-26577: camel-support -
RebalancingCamelClusterService: release the views a rebalancing pass looks at
(#29202)
8fc0dc2f36bb is described below
commit 8fc0dc2f36bbe48257b662865b8f654e28cc8c00
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 5 17:38:02 2026 +0530
CAMEL-26577: camel-support - RebalancingCamelClusterService: release the
views a rebalancing pass looks at (#29202)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../RebalancingClusterServiceReleaseViewTest.java | 233 +++++++++++++++++++++
.../cluster/AbstractCamelClusterService.java | 29 +++
.../cluster/RebalancingCamelClusterService.java | 93 ++++++--
3 files changed, 340 insertions(+), 15 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/cluster/RebalancingClusterServiceReleaseViewTest.java
b/core/camel-core/src/test/java/org/apache/camel/cluster/RebalancingClusterServiceReleaseViewTest.java
new file mode 100644
index 000000000000..c4ea649a9bd6
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/cluster/RebalancingClusterServiceReleaseViewTest.java
@@ -0,0 +1,233 @@
+/*
+ * 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.cluster;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.cluster.ClusteredRoutePolicy;
+import org.apache.camel.support.cluster.AbstractCamelClusterService;
+import org.apache.camel.support.cluster.AbstractCamelClusterView;
+import org.apache.camel.support.cluster.RebalancingCamelClusterService;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The rebalancing cluster service looks at the views of the delegate on every
pass; it must not keep a reference to
+ * them, nor start a view again that its last user has released.
+ */
+public class RebalancingClusterServiceReleaseViewTest {
+
+ private CamelContext context;
+ private TestClusterService delegate;
+ private TestRebalancingClusterService service;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ context = new DefaultCamelContext();
+ delegate = new TestClusterService("node-1");
+ // a long period, so the only passes are the one at start-up and the
ones the tests run
+ service = new TestRebalancingClusterService(delegate);
+ context.addService(service);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ context.stop();
+ }
+
+ @Test
+ public void testRemovedRouteStopsView() throws Exception {
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace(service, "my-ns");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ context.start();
+ assertTrue(service.firstPass.await(10, TimeUnit.SECONDS));
+
+ TestClusterView view = delegate.view("my-ns");
+ assertTrue(view.isRunAllowed());
+
+ // a rebalancing pass while the route uses the view
+ service.reconcile();
+ assertTrue(view.isRunAllowed());
+
+ context.getRouteController().stopRoute("foo");
+ context.removeRoute("foo");
+
+ // no route uses the namespace any more, so this node must leave the
election
+ assertEquals(ServiceStatus.Stopped, view.getStatus());
+ }
+
+ @Test
+ public void testRebalancingDoesNotStartReleasedView() throws Exception {
+ context.start();
+ assertTrue(service.firstPass.await(10, TimeUnit.SECONDS));
+
+ CamelClusterView used = delegate.getView("my-ns");
+ TestClusterView view = delegate.view("my-ns");
+ assertTrue(view.isRunAllowed());
+ delegate.releaseView(used);
+ assertEquals(ServiceStatus.Stopped, view.getStatus());
+ int starts = view.starts.get();
+
+ service.reconcile();
+
+ // not even started and stopped again within the pass (that would take
the leadership for a moment)
+ assertEquals(ServiceStatus.Stopped, view.getStatus());
+ assertEquals(starts, view.starts.get());
+ }
+
+ @Test
+ public void testRebalancingStillOpensViewInUse() throws Exception {
+ context.start();
+ assertTrue(service.firstPass.await(10, TimeUnit.SECONDS));
+
+ CamelClusterView used = delegate.getView("my-ns");
+ TestClusterView view = delegate.view("my-ns");
+ view.setDisabled(true);
+
+ // one member and one partition that this member does not own: the
pass opens it
+ service.reconcile();
+
+ assertFalse(view.isDisabled());
+ assertTrue(view.isRunAllowed());
+ delegate.releaseView(used);
+ assertEquals(ServiceStatus.Stopped, view.getStatus());
+ }
+
+ // *********************************
+ // Helpers
+ // *********************************
+
+ private final class TestRebalancingClusterService extends
RebalancingCamelClusterService {
+ private final CountDownLatch firstPass = new CountDownLatch(1);
+
+ TestRebalancingClusterService(CamelPreemptiveClusterService delegate) {
+ super(delegate, TimeUnit.HOURS.toMillis(1));
+ }
+
+ @Override
+ protected void reconcile() {
+ try {
+ super.reconcile();
+ } finally {
+ firstPass.countDown();
+ }
+ }
+ }
+
+ private static final class TestClusterView extends
AbstractCamelClusterView implements CamelPreemptiveClusterView {
+ private final CamelClusterMember local;
+ private final AtomicInteger starts = new AtomicInteger();
+ private volatile boolean disabled;
+
+ TestClusterView(CamelClusterService cluster, String namespace) {
+ super(cluster, namespace);
+ this.local = new CamelClusterMember() {
+ @Override
+ public boolean isLeader() {
+ return false;
+ }
+
+ @Override
+ public boolean isLocal() {
+ return true;
+ }
+
+ @Override
+ public String getId() {
+ return getClusterService().getId();
+ }
+ };
+ }
+
+ @Override
+ public Optional<CamelClusterMember> getLeader() {
+ return Optional.empty();
+ }
+
+ @Override
+ public CamelClusterMember getLocalMember() {
+ return local;
+ }
+
+ @Override
+ public List<CamelClusterMember> getMembers() {
+ return List.of(local);
+ }
+
+ @Override
+ public void setDisabled(boolean disabled) {
+ this.disabled = disabled;
+ }
+
+ @Override
+ public boolean isDisabled() {
+ return disabled;
+ }
+
+ @Override
+ protected void doStart() {
+ starts.incrementAndGet();
+ }
+
+ @Override
+ protected void doStop() {
+ }
+ }
+
+ private static final class TestClusterService extends
AbstractCamelClusterService<TestClusterView>
+ implements CamelPreemptiveClusterService {
+ private final Map<String, TestClusterView> created = new
ConcurrentHashMap<>();
+
+ TestClusterService(String id) {
+ super(id);
+ }
+
+ @Override
+ protected TestClusterView createView(String namespace) {
+ return created.computeIfAbsent(namespace, ns -> new
TestClusterView(this, ns));
+ }
+
+ @Override
+ public CamelPreemptiveClusterView getView(String namespace) throws
Exception {
+ return (CamelPreemptiveClusterView) super.getView(namespace);
+ }
+
+ TestClusterView view(String namespace) {
+ return created.get(namespace);
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/cluster/AbstractCamelClusterService.java
b/core/camel-support/src/main/java/org/apache/camel/support/cluster/AbstractCamelClusterService.java
index 490c42eee837..81b495fb0b0d 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/cluster/AbstractCamelClusterService.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/cluster/AbstractCamelClusterService.java
@@ -216,6 +216,35 @@ public abstract class AbstractCamelClusterService<T
extends CamelClusterView> ex
});
}
+ /**
+ * Takes a reference to the view of the namespace only while the view is
in use (it has references), so the caller
+ * can look at it without creating it or starting it again after its last
user has released it. The caller must
+ * release the view with {@link #releaseView(CamelClusterView)}.
+ *
+ * @return the view, or {@code null} if it is not in use
+ */
+ CamelClusterView retainViewIfInUse(String namespace) {
+ return LockHelper.supplyWithWriteLock(
+ lock,
+ () -> {
+ ViewHolder<T> holder = views.get(namespace);
+ // the view is in use, so this reference does not start it
+ return holder != null && holder.count.get() > 0 ?
holder.retain() : null;
+ });
+ }
+
+ /**
+ * Whether the view of the namespace is in use (it has references).
+ */
+ boolean isViewInUse(String namespace) {
+ return LockHelper.supplyWithReadLock(
+ lock,
+ () -> {
+ ViewHolder<T> holder = views.get(namespace);
+ return holder != null && holder.count.get() > 0;
+ });
+ }
+
@Override
public boolean isLeader(String namespace) {
return LockHelper.supplyWithReadLock(
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/cluster/RebalancingCamelClusterService.java
b/core/camel-support/src/main/java/org/apache/camel/support/cluster/RebalancingCamelClusterService.java
index 40f26bc45817..71a9f62bc47e 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/cluster/RebalancingCamelClusterService.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/cluster/RebalancingCamelClusterService.java
@@ -164,10 +164,18 @@ public class RebalancingCamelClusterService extends
ServiceSupport implements Ca
protected void setDisabled(String partition, boolean disabled) {
try {
- LOG.debug("Setting partition {} to disabled={}...", partition,
disabled);
- CamelPreemptiveClusterView view = delegate.getView(partition);
- if (view.isDisabled() != disabled) {
- view.setDisabled(disabled);
+ CamelPreemptiveClusterView view = retainViewInUse(partition);
+ if (view == null) {
+ LOG.debug("Partition {} is not in use", partition);
+ return;
+ }
+ try {
+ LOG.debug("Setting partition {} to disabled={}...", partition,
disabled);
+ if (view.isDisabled() != disabled) {
+ view.setDisabled(disabled);
+ }
+ } finally {
+ releaseRetainedView(view);
}
} catch (Exception ex) {
LOG.warn("Could not get view {}", partition, ex);
@@ -178,9 +186,17 @@ public class RebalancingCamelClusterService extends
ServiceSupport implements Ca
List<String> owned = new ArrayList<>(partitions.size());
for (String partition : partitions) {
try {
- CamelPreemptiveClusterView view = delegate.getView(partition);
- if (!view.isDisabled() && view.getLocalMember().isLeader()) {
- owned.add(partition);
+ CamelPreemptiveClusterView view = retainViewInUse(partition);
+ if (view == null) {
+ // released since the partitions were listed
+ continue;
+ }
+ try {
+ if (!view.isDisabled() &&
view.getLocalMember().isLeader()) {
+ owned.add(partition);
+ }
+ } finally {
+ releaseRetainedView(view);
}
} catch (Exception ex) {
LOG.warn("Could not get view {}", partition, ex);
@@ -191,22 +207,37 @@ public class RebalancingCamelClusterService extends
ServiceSupport implements Ca
}
protected List<String> partitionList() {
- ArrayList<String> partitions = new ArrayList<>(this.getNamespaces());
+ List<String> partitions = new ArrayList<>();
+ for (String namespace : this.getNamespaces()) {
+ // a view that no route (or other user) holds any more is not a
partition to balance
+ if (isViewInUse(namespace)) {
+ partitions.add(namespace);
+ }
+ }
Collections.sort(partitions);
return partitions;
}
protected Integer members() {
Set<String> members = null;
- for (String group : this.getNamespaces()) {
+ for (String group : partitionList()) {
try {
- CamelPreemptiveClusterView view = delegate.getView(group);
- Set<String> viewMembers =
view.getMembers().stream().map(CamelClusterMember::getId).collect(Collectors.toSet());
- if (members != null && !members.equals(viewMembers)) {
- LOG.debug("View members don't match: {} vs {}", members,
viewMembers);
- return null;
+ CamelPreemptiveClusterView view = retainViewInUse(group);
+ if (view == null) {
+ // released since the partitions were listed
+ continue;
+ }
+ try {
+ Set<String> viewMembers
+ =
view.getMembers().stream().map(CamelClusterMember::getId).collect(Collectors.toSet());
+ if (members != null && !members.equals(viewMembers)) {
+ LOG.debug("View members don't match: {} vs {}",
members, viewMembers);
+ return null;
+ }
+ members = viewMembers;
+ } finally {
+ releaseRetainedView(view);
}
- members = viewMembers;
} catch (Exception ex) {
LOG.warn("Could not get view {}", group, ex);
return null;
@@ -215,6 +246,38 @@ public class RebalancingCamelClusterService extends
ServiceSupport implements Ca
return members != null ? members.size() : 0;
}
+ /**
+ * Whether the view of the partition is held by a route or another user.
The rebalancing only looks at those views,
+ * so it does not start a view again after its last user has released it.
+ */
+ private boolean isViewInUse(String partition) {
+ if (delegate instanceof AbstractCamelClusterService<?> service) {
+ return service.isViewInUse(partition);
+ }
+ return true;
+ }
+
+ /**
+ * Takes a reference to the view of the partition if it is in use, or
returns {@code null}. The caller releases it
+ * with {@link #releaseRetainedView(CamelClusterView)}, so the rebalancing
does not keep a view running after its
+ * last user has released it. A delegate that is not an {@link
AbstractCamelClusterService} has no such lookup, so
+ * its views are taken with {@code getView} (which may start a released
view until it is released again).
+ */
+ private CamelPreemptiveClusterView retainViewInUse(String partition)
throws Exception {
+ if (delegate instanceof AbstractCamelClusterService<?> service) {
+ return (CamelPreemptiveClusterView)
service.retainViewIfInUse(partition);
+ }
+ return delegate.getView(partition);
+ }
+
+ private void releaseRetainedView(CamelClusterView view) {
+ try {
+ delegate.releaseView(view);
+ } catch (Exception ex) {
+ LOG.warn("Could not release view {}", view.getNamespace(), ex);
+ }
+ }
+
private void rescheduleAfterDelay() {
this.serializedExecutor.schedule(this::reconcile,
this.periodMillis,
