This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/26112-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit b2701152de5c4e75d2c53b07af5cfd7d4f91c6d9 Author: Guillaume Nodet <[email protected]> AuthorDate: Fri Sep 11 14:21:04 2026 +0000 [backport camel-4.18.x] CAMEL-24626: camel-master - leadership gets its own lock, and cancelled start tasks leave the task registry --- .../src/main/docs/master-component.adoc | 11 + .../camel/component/master/MasterConsumer.java | 237 ++++++++--- .../master/MasterConsumerLeadershipTest.java | 459 +++++++++++++++++++++ .../consumer/SimpleMessageListenerContainer.java | 7 +- .../support/task/task/BackgroundTaskTest.java | 158 +++++++ .../apache/camel/support/task/BackgroundTask.java | 128 +++++- .../java/org/apache/camel/support/task/Task.java | 8 + 7 files changed, 935 insertions(+), 73 deletions(-) diff --git a/components/camel-master/src/main/docs/master-component.adoc b/components/camel-master/src/main/docs/master-component.adoc index df71353940e2..0ddeda042cbb 100644 --- a/components/camel-master/src/main/docs/master-component.adoc +++ b/components/camel-master/src/main/docs/master-component.adoc @@ -37,6 +37,17 @@ include::partial$component-endpoint-headers.adoc[] == Usage +=== Starting the delegated consumer + +When a node takes the leadership, the delegated consumer is started in the background, and the start is +retried when it fails, for instance because the back end it consumes from is not reachable yet. The retries +are configured on the component with `backOffDelay` (the delay between attempts, 5000 millis by default) +and `backOffMaxAttempts` (the number of attempts, 10 by default). + +Once the attempts are used up, the node keeps the leadership but consumes nothing until the leadership +changes again, which is logged at ERROR level. Set `backOffMaxAttempts` to 0 to keep retrying for as long as +the node is the leader instead. + === Using the master endpoint Prefix any camel endpoint with **master:someName:** where _someName_ is a logical name and is diff --git a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java index 5dfebcb3e6de..f9d9f79685f2 100644 --- a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java +++ b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java @@ -17,7 +17,11 @@ package org.apache.camel.component.master; import java.time.Duration; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.camel.Consumer; import org.apache.camel.Endpoint; @@ -59,6 +63,15 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum private volatile CamelClusterView view; private ResumeStrategy resumeStrategy; private ScheduledExecutorService leaderPool; + // leadership state and the pending start task are guarded by leadershipLock. This is deliberately not + // the lock of BaseService: the cluster view dispatches events while holding its own read lock and then + // needs this lock, while doStop holds the service lock and needs the write lock of the view to remove + // the listener. Guarding the leadership with the service lock closes that into a lock cycle, and it also + // makes every leadership event and start attempt wait for whatever lifecycle operation is in progress + private final Lock leadershipLock = new ReentrantLock(); + private boolean leadershipTaken; + private BackgroundTask leaderTask; + private Future<?> leaderTaskFuture; public MasterConsumer(MasterEndpoint masterEndpoint, Processor processor, CamelClusterService clusterService) { super(masterEndpoint, processor); @@ -103,6 +116,20 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum protected void doStop() throws Exception { super.doStop(); + leadershipLock.lock(); + try { + // a start can still be pending, cancel it first so it cannot start the delegated consumer + // after this consumer has been stopped + leadershipTaken = false; + cancelLeaderTask(true); + } finally { + leadershipLock.unlock(); + } + + // note: removeEventListener below needs the write lock of the cluster view, while an event dispatch + // takes the read lock of the view and then leadershipLock. This thread must not hold leadershipLock + // here, or the two orders deadlock + if (view != null) { view.removeEventListener(leadershipListener); clusterService.releaseView(view); @@ -145,74 +172,146 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum .withBudget(Budgets.iterationTimeBudget() .withInterval(Duration.ofMillis(masterEndpoint.getComponent().getBackOffDelay())) .withInitialDelay(Duration.ofSeconds(1)) + // 0 or less leaves the unlimited default of the builder in place .withMaxIterations(masterEndpoint.getComponent().getBackOffMaxAttempts()) + // the attempts are bounded by backOffMaxAttempts, not by the 5s default duration of + // the builder, which would otherwise end the task before the second attempt + .withUnlimitedDuration() .build()) .withName("Leadership") .build(); } - private void onLeadershipTaken() throws Exception { - lock.lock(); + private void onLeadershipTaken() { + leadershipLock.lock(); try { if (!isRunAllowed()) { return; } - if (delegatedConsumer != null) { + leadershipTaken = true; + + if (delegatedConsumer != null || isStartPending()) { return; } - final BackgroundTask leaderTask = createTask(); - leaderTask.schedule(getEndpoint().getCamelContext(), () -> { - if (!isRunAllowed()) { - return false; - } - LOG.info("Leadership taken. Attempt #{} to start consumer: {}", leaderTask.iteration(), delegatedEndpoint); - - Exception cause = null; - try { - if (delegatedConsumer == null) { - delegatedConsumer = delegatedEndpoint.createConsumer(processor); - if (delegatedConsumer instanceof StartupListener) { - getEndpoint().getCamelContext().addStartupListener((StartupListener) delegatedConsumer); - } - if (delegatedConsumer instanceof ResumeAware resumeAwareConsumer && resumeStrategy != null) { - LOG.debug("Setting up the resume adapter for the resume strategy in consumer"); - ResumeAdapter resumeAdapter - = AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer, - resumeStrategy); - resumeStrategy.setAdapter(resumeAdapter); - - LOG.debug("Setting up the resume strategy for consumer"); - resumeAwareConsumer.setResumeStrategy(resumeStrategy); - } - } - ServiceHelper.startService(delegatedEndpoint, delegatedConsumer); + // a task from a previous leadership term may still be scheduled, drop it + cancelLeaderTask(false); + + final BackgroundTask task = createTask(); + // the consumer is created once and re-used by the start attempts of this task + final AtomicReference<Consumer> attempt = new AtomicReference<>(); + leaderTask = task; + leaderTaskFuture = task.schedule(getEndpoint().getCamelContext(), () -> startDelegatedConsumer(task, attempt)); + } finally { + leadershipLock.unlock(); + } + } - } catch (Exception e) { - cause = e; + private boolean startDelegatedConsumer(BackgroundTask task, AtomicReference<Consumer> attempt) { + leadershipLock.lock(); + try { + if (!isRunAllowed()) { + return false; + } + + if (!leadershipTaken) { + // leadership was lost while this start was pending. Starting now would run the consumer on a + // node that is not the leader, and no further leadership event is coming to stop it again + LOG.debug("Leadership lost while the start was pending. Not starting consumer: {}", delegatedEndpoint); + return true; // no more attempts + } + + if (delegatedConsumer != null) { + return true; // no more attempts + } + } finally { + leadershipLock.unlock(); + } + + LOG.info("Leadership taken. Attempt #{} to start consumer: {}", task.iteration(), delegatedEndpoint); + + // the delegate is created and started without holding the lock. It can block for a long time, and the + // lock is taken by the cluster view event dispatch and by doStop, which must not wait for a broker + // connect. The leadership is re-checked below before the consumer is published + Consumer consumer = attempt.get(); + Exception cause = null; + try { + if (consumer == null) { + consumer = delegatedEndpoint.createConsumer(processor); + // held for the attempts of this task, so the startup listener and the resume strategy are + // wired once and a retry only starts the consumer again + attempt.set(consumer); + if (consumer instanceof StartupListener startupListener) { + getEndpoint().getCamelContext().addStartupListener(startupListener); + } + if (consumer instanceof ResumeAware resumeAwareConsumer && resumeStrategy != null) { + LOG.debug("Setting up the resume adapter for the resume strategy in consumer"); + ResumeAdapter resumeAdapter + = AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer, + resumeStrategy); + resumeStrategy.setAdapter(resumeAdapter); + + LOG.debug("Setting up the resume strategy for consumer"); + resumeAwareConsumer.setResumeStrategy(resumeStrategy); } + } + ServiceHelper.startService(delegatedEndpoint, consumer); + } catch (Exception e) { + cause = e; + } - if (cause != null) { - String message = "Leadership taken. Attempt #" + leaderTask.iteration() - + " failed to start consumer due to: " + cause.getMessage(); - getExceptionHandler().handleException(message, cause); - // make the task runner aware of the exception (will retry) - throw new TaskRunFailureException(message, cause); + leadershipLock.lock(); + try { + if (cause != null) { + // the consumer is kept for the next attempt. It is not stopped here: a consumer that failed to + // start was already stopped by its own start(), and shutting it down would also shut down the + // processor of the route, which the next attempt and this consumer still need + String message = "Leadership taken. Attempt #" + task.iteration() + + " failed to start consumer due to: " + cause.getMessage(); + getExceptionHandler().handleException(message, cause); + int maxAttempts = masterEndpoint.getComponent().getBackOffMaxAttempts(); + if (maxAttempts > 0 && task.iteration() >= maxAttempts) { + LOG.error("Leadership taken. Giving up after {} attempts to start consumer: {}." + + " This node holds the leadership but is not consuming, until the leadership changes again.", + task.iteration(), delegatedEndpoint); } + // make the task runner aware of the exception (will retry) + throw new TaskRunFailureException(message, cause); + } - LOG.info("Leadership taken. Attempt #{} success. Consumer started: {}", leaderTask.iteration(), - delegatedEndpoint); + if (!leadershipTaken || !isRunAllowed()) { + // the leadership went away while the consumer was starting, so stop what was just started + // instead of publishing it. No leadership event is going to do it, delegatedConsumer is unset + LOG.info("Leadership lost while the consumer was starting. Stopping consumer: {}", delegatedEndpoint); + ServiceHelper.stopAndShutdownServices(consumer, delegatedEndpoint); + attempt.set(null); return true; // no more attempts - }); + } + + delegatedConsumer = consumer; + LOG.info("Leadership taken. Attempt #{} success. Consumer started: {}", task.iteration(), + delegatedEndpoint); + // release the task, a later leadership term schedules a new one + cancelLeaderTask(false); + return true; // no more attempts } finally { - lock.unlock(); + leadershipLock.unlock(); } } private void onLeadershipLost() { - lock.lock(); + leadershipLock.lock(); try { + leadershipTaken = false; + // a start scheduled by the leadership taken event may not have run yet, cancel it so it + // cannot start the consumer on a node that is no longer the leader + cancelLeaderTask(false); + + if (delegatedConsumer == null) { + return; + } + LOG.debug("Leadership lost. Stopping consumer: {}", delegatedEndpoint); try { ServiceHelper.stopAndShutdownServices(delegatedConsumer, delegatedEndpoint); @@ -221,7 +320,22 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum } LOG.info("Leadership lost. Consumer stopped: {}", delegatedEndpoint); } finally { - lock.unlock(); + leadershipLock.unlock(); + } + } + + private boolean isStartPending() { + return leaderTaskFuture != null && !leaderTaskFuture.isDone(); + } + + private void cancelLeaderTask(boolean mayInterruptIfRunning) { + if (leaderTask != null) { + // cancelled through the task and not through its future, so the task also leaves the + // TaskManagerRegistry. Only a run of the task removes it from there, and once the schedule + // is cancelled no run is coming + leaderTask.cancel(mayInterruptIfRunning); + leaderTask = null; + leaderTaskFuture = null; } } @@ -233,22 +347,37 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum @Override public void leadershipChanged(CamelClusterView view, CamelClusterMember leader) { if (!isRunAllowed()) { + // this runs on the dispatch thread of the cluster view, holding the lock of that view, so + // do no work at all for a consumer that is stopping return; } - if (view.getLocalMember().isLeader()) { - try { - onLeadershipTaken(); - } catch (Exception e) { - getExceptionHandler().handleException("Error starting consumer while taking leadership", e); + leadershipLock.lock(); + try { + if (!isRunAllowed()) { + return; } - } else if (delegatedConsumer != null) { - try { - onLeadershipLost(); - } catch (Exception e) { - getExceptionHandler() - .handleException("Error stopping consumer while loosing leadership. This exception is ignored.", e); + + // the leadership is read under the same lock that applies it, so that two events + // dispatched concurrently cannot be applied in the wrong order + if (view.getLocalMember().isLeader()) { + try { + onLeadershipTaken(); + } catch (Exception e) { + getExceptionHandler().handleException("Error starting consumer while taking leadership", e); + } + } else { + // dispatched even when there is no consumer yet, as a start may be pending + try { + onLeadershipLost(); + } catch (Exception e) { + getExceptionHandler() + .handleException("Error stopping consumer while loosing leadership. This exception is ignored.", + e); + } } + } finally { + leadershipLock.unlock(); } } } diff --git a/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java b/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java new file mode 100644 index 000000000000..f39510bfbf14 --- /dev/null +++ b/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java @@ -0,0 +1,459 @@ +/* + * 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.master; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.Consumer; +import org.apache.camel.Endpoint; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.cluster.CamelClusterMember; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.DefaultComponent; +import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.support.DefaultEndpoint; +import org.apache.camel.support.PluginHelper; +import org.apache.camel.support.cluster.AbstractCamelClusterService; +import org.apache.camel.support.cluster.AbstractCamelClusterView; +import org.apache.camel.support.task.TaskManagerRegistry; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +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.assertTrue; + +/** + * Verifies that the delegated consumer only ever runs while this node holds the leadership, also when the leadership + * changes while the start of the delegated consumer is still pending. + */ +public class MasterConsumerLeadershipTest { + + private DefaultCamelContext context; + private TestClusterService clusterService; + private ProbeComponent probe; + + @BeforeEach + void setUp() throws Exception { + probe = new ProbeComponent(); + clusterService = new TestClusterService(); + + context = new DefaultCamelContext(); + context.disableJMX(); + context.addService(clusterService); + context.addComponent("probe", probe); + + MasterComponent master = context.getComponent("master", MasterComponent.class); + // keep the retries short so an exhausted start does not dominate the test time + master.setBackOffDelay(200); + master.setBackOffMaxAttempts(2); + + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("master:ns:probe:test").routeId("master-route").to("mock:result"); + } + }); + + context.start(); + } + + @AfterEach + void tearDown() { + // a test that fails while the delegate is parked on a gate would otherwise wedge the stop below + probe.releaseGates(); + if (context != null) { + context.stop(); + } + } + + @Test + @Timeout(60) + void testLeadershipLostWhilePendingStartDoesNotStartConsumer() { + TestClusterView view = clusterService.getTestView(); + + view.setLeader(true); + // the start is scheduled with an initial delay, so it is still pending here + view.setLeader(false); + + // outlast the initial delay of the start task and verify the consumer was never even created + await().during(3, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, probe.created.get())); + + // taking the leadership again must still work, which also proves the events did reach the consumer + view.setLeader(true); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.started.get())); + assertEquals(1, probe.created.get(), "The cancelled start must not have created a second consumer"); + } + + @Test + @Timeout(60) + void testLeadershipTakenStartsConsumerAndLostStopsIt() { + TestClusterView view = clusterService.getTestView(); + + view.setLeader(true); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.started.get())); + + view.setLeader(false); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.stopped.get())); + } + + @Test + @Timeout(60) + void testConsumerIsRestartedWhenLeadershipFlapsAfterASuccessfulStart() { + TestClusterView view = clusterService.getTestView(); + + view.setLeader(true); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.started.get())); + + // the membership flap seen in production: the consumer must stop and then come back + view.setLeader(false); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.stopped.get())); + + view.setLeader(true); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(2, probe.started.get())); + assertEquals(1, probe.stopped.get()); + } + + @Test + @Timeout(60) + void testLeadershipLostWhileStartIsInProgressStopsTheConsumer() throws Exception { + TestClusterView view = clusterService.getTestView(); + CountDownLatch startGate = new CountDownLatch(1); + probe.startGate.set(startGate); + + view.setLeader(true); + // wait until the start is running and blocked inside the delegated consumer + await().atMost(10, TimeUnit.SECONDS).until(() -> probe.startAttempts.get() == 1); + + // the leadership is lost while the start is in progress, this must not be able to interleave + Thread loser = new Thread(() -> view.setLeader(false), "leadership-lost"); + loser.start(); + startGate.countDown(); + loser.join(TimeUnit.SECONDS.toMillis(20)); + + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(1, probe.started.get()); + assertEquals(1, probe.stopped.get(), "The consumer started on a node that lost the leadership must be stopped"); + }); + } + + @Test + @Timeout(60) + void testRepeatedLeadershipTakenStartsOnlyOneConsumer() { + TestClusterView view = clusterService.getTestView(); + + view.setLeader(true); + view.setLeader(true); + + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.started.get())); + await().during(2, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, probe.created.get())); + } + + @Test + @Timeout(60) + void testStoppingTheConsumerCancelsAPendingStart() { + TestClusterView view = clusterService.getTestView(); + + view.setLeader(true); + // the start is still pending, stopping must cancel it instead of letting it start afterwards + context.stop(); + + await().during(3, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, probe.created.get())); + } + + @Test + @Timeout(60) + void testConsumerStartsAfterLeadershipIsTakenAgainWhenAnEarlierStartFailed() { + TestClusterView view = clusterService.getTestView(); + + probe.failStart.set(true); + view.setLeader(true); + + // every start attempt fails, then the task runs out of budget and stops attempting + await().atMost(20, TimeUnit.SECONDS).until(() -> probe.startAttempts.get() == 2); + await().during(1, TimeUnit.SECONDS).atMost(20, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(2, probe.startAttempts.get())); + assertEquals(0, probe.started.get()); + + // a failed start must not leave state behind that makes a later leadership event a no-op, + // not even without an intervening leadership lost event + probe.failStart.set(false); + view.setLeader(true); + + await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.started.get())); + } + + @Test + @Timeout(60) + void testAllConfiguredStartAttemptsAreMade() { + TestClusterView view = clusterService.getTestView(); + MasterComponent master = context.getComponent("master", MasterComponent.class); + // the attempts have to outlast the default 5s duration of the iteration time budget + master.setBackOffDelay(3000); + master.setBackOffMaxAttempts(3); + + probe.failStart.set(true); + view.setLeader(true); + + // every configured attempt must be made, the task must not end on a time budget of its own + await().atMost(30, TimeUnit.SECONDS).until(() -> probe.startAttempts.get() == 3); + assertEquals(0, probe.started.get()); + } + + @Test + @Timeout(60) + void testCancellingAPendingStartRemovesTheTaskFromTheRegistry() { + TestClusterView view = clusterService.getTestView(); + TaskManagerRegistry registry = PluginHelper.getTaskManagerRegistry(context.getCamelContextExtension()); + MasterComponent master = context.getComponent("master", MasterComponent.class); + // the task must still be retrying when the leadership is lost below, not exhausted by then + master.setBackOffMaxAttempts(1000); + + probe.failStart.set(true); + view.setLeader(true); + + // the task adds itself to the registry from its first run + await().atMost(20, TimeUnit.SECONDS).until(() -> probe.startAttempts.get() >= 1); + await().atMost(20, TimeUnit.SECONDS).until(() -> hasLeadershipTask(registry)); + + view.setLeader(false); + + // only a run of the task removes it from the registry, and after the cancel no run is coming + await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertFalse(hasLeadershipTask(registry), + "The cancelled start task must not stay in the task registry")); + } + + @Test + @Timeout(60) + void testEventDispatchIsNotBlockedByALifecycleOperation() throws Exception { + TestClusterView view = clusterService.getTestView(); + + view.setLeader(true); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, probe.started.get())); + + CountDownLatch suspendEntered = new CountDownLatch(1); + CountDownLatch suspendGate = new CountDownLatch(1); + probe.suspendEntered.set(suspendEntered); + probe.suspendGate.set(suspendGate); + + // suspending holds the service lock of the master consumer for as long as the delegate takes + MasterConsumer consumer = (MasterConsumer) context.getRoute("master-route").getConsumer(); + Thread suspender = new Thread(consumer::suspend, "suspend"); + suspender.start(); + assertTrue(suspendEntered.await(20, TimeUnit.SECONDS), "The suspend of the delegate should have started"); + + // the cluster view dispatches its events while holding its own lock, and needs that same lock again + // to remove the listener when the consumer stops. An event that waits here for the service lock of + // the consumer is what closes that into a deadlock, so the leadership must not be guarded by it + Thread dispatcher = new Thread(() -> view.setLeader(true), "leadership-taken"); + dispatcher.start(); + try { + dispatcher.join(TimeUnit.SECONDS.toMillis(20)); + assertFalse(dispatcher.isAlive(), "An event dispatch must not wait for the service lock of the consumer"); + } finally { + suspendGate.countDown(); + suspender.join(TimeUnit.SECONDS.toMillis(20)); + } + } + + private static boolean hasLeadershipTask(TaskManagerRegistry registry) { + return registry.getTasks().stream().anyMatch(task -> "Leadership".equals(task.getName())); + } + + // ************************************ + // Delegated endpoint under observation + // ************************************ + + private static final class ProbeComponent extends DefaultComponent { + private final AtomicInteger created = new AtomicInteger(); + private final AtomicInteger startAttempts = new AtomicInteger(); + private final AtomicInteger started = new AtomicInteger(); + private final AtomicInteger stopped = new AtomicInteger(); + private final AtomicBoolean failStart = new AtomicBoolean(); + private final AtomicReference<CountDownLatch> startGate = new AtomicReference<>(); + private final AtomicReference<CountDownLatch> suspendEntered = new AtomicReference<>(); + private final AtomicReference<CountDownLatch> suspendGate = new AtomicReference<>(); + + @Override + protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) { + return new ProbeEndpoint(uri, this); + } + + void releaseGates() { + List.of(startGate, suspendGate).forEach(gate -> { + CountDownLatch latch = gate.get(); + if (latch != null) { + latch.countDown(); + } + }); + } + } + + private static final class ProbeEndpoint extends DefaultEndpoint { + private final ProbeComponent component; + + ProbeEndpoint(String uri, ProbeComponent component) { + super(uri, component); + this.component = component; + } + + @Override + public Producer createProducer() { + throw new UnsupportedOperationException("Cannot produce from this endpoint"); + } + + @Override + public Consumer createConsumer(Processor processor) { + component.created.incrementAndGet(); + return new ProbeConsumer(this, processor, component); + } + + @Override + public boolean isSingleton() { + return true; + } + } + + private static final class ProbeConsumer extends DefaultConsumer { + private final ProbeComponent component; + + ProbeConsumer(Endpoint endpoint, Processor processor, ProbeComponent component) { + super(endpoint, processor); + this.component = component; + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + // counted before the failure flag is read, so a test can await an attempt that has made its decision + component.startAttempts.incrementAndGet(); + CountDownLatch gate = component.startGate.get(); + if (gate != null) { + gate.await(); + } + if (component.failStart.get()) { + throw new IllegalStateException("Simulated failure to start the delegated consumer"); + } + component.started.incrementAndGet(); + } + + @Override + protected void doStop() throws Exception { + super.doStop(); + component.stopped.incrementAndGet(); + } + + @Override + protected void doSuspend() throws Exception { + super.doSuspend(); + CountDownLatch entered = component.suspendEntered.get(); + if (entered != null) { + entered.countDown(); + } + CountDownLatch gate = component.suspendGate.get(); + if (gate != null) { + gate.await(); + } + } + } + + // ************************************ + // Cluster with a leadership we control + // ************************************ + + private static final class TestClusterService extends AbstractCamelClusterService<TestClusterView> { + private volatile TestClusterView view; + + TestClusterService() { + super("test-cluster-service"); + } + + TestClusterView getTestView() { + return view; + } + + @Override + protected TestClusterView createView(String namespace) { + view = new TestClusterView(this, namespace); + return view; + } + } + + private static final class TestClusterView extends AbstractCamelClusterView { + private final TestClusterMember localMember = new TestClusterMember(); + + TestClusterView(TestClusterService clusterService, String namespace) { + super(clusterService, namespace); + } + + void setLeader(boolean leader) { + localMember.leader = leader; + fireLeadershipChangedEvent(leader ? localMember : null); + } + + @Override + public Optional<CamelClusterMember> getLeader() { + return localMember.isLeader() ? Optional.of(localMember) : Optional.empty(); + } + + @Override + public CamelClusterMember getLocalMember() { + return localMember; + } + + @Override + public List<CamelClusterMember> getMembers() { + return List.of(localMember); + } + } + + private static final class TestClusterMember implements CamelClusterMember { + private final String id = UUID.randomUUID().toString(); + private volatile boolean leader; + + @Override + public boolean isLeader() { + return leader; + } + + @Override + public boolean isLocal() { + return true; + } + + @Override + public String getId() { + return id; + } + } +} diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java index 932e6e8f3512..2aa2b821637c 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java @@ -269,8 +269,11 @@ public class SimpleMessageListenerContainer extends ServiceSupport endpoint.getCamelContext().getExecutorServiceManager().shutdown(recoverPool); recoverPool = null; } - if (recoverFuture != null && recoverTask != null && recoverTask.isRunning()) { - recoverFuture.cancel(true); + if (recoverTask != null && recoverTask.isRunning()) { + // cancelled through the task and not through its future, so the task also leaves the + // TaskManagerRegistry. Only a run of the task removes it from there, and once the schedule + // is cancelled no run is coming + recoverTask.cancel(true); recoverTask = null; recoverFuture = null; } diff --git a/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskTest.java b/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskTest.java index e61ff91c32f7..039405fb07bf 100644 --- a/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskTest.java @@ -18,14 +18,21 @@ package org.apache.camel.support.task.task; import java.time.Duration; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.camel.support.PluginHelper; import org.apache.camel.support.task.BackgroundTask; +import org.apache.camel.support.task.Task; +import org.apache.camel.support.task.TaskManagerRegistry; import org.apache.camel.support.task.Tasks; import org.apache.camel.support.task.budget.Budgets; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +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.assertNotNull; @@ -208,4 +215,155 @@ public class BackgroundTaskTest extends TaskTestSupport { assertTrue(duration.getSeconds() <= 5); assertFalse(completed, "The task did not complete because of timeout, the return should be false"); } + + @DisplayName("Test that a scheduled task is unscheduled once it has completed") + @Test + @Timeout(10) + void testScheduleStopsWhenCompleted() { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + try { + BackgroundTask task = Tasks.backgroundTask() + .withScheduledExecutor(executor) + .withBudget(Budgets.iterationTimeBudget() + .withInterval(Duration.ofMillis(100)) + .withInitialDelay(Duration.ZERO) + .withMaxIterations(maxIterations) + .build()) + .build(); + + Future<?> future = task.schedule(camelContext, () -> { + taskCount.increment(); + return true; + }); + + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertTrue(future.isCancelled(), + "A completed task should not stay scheduled")); + assertEquals(1, taskCount.intValue(), "The supplier should have run exactly once"); + assertEquals(Task.Status.Completed, task.getStatus()); + } finally { + executor.shutdownNow(); + } + } + + @DisplayName("Test that a scheduled task is unscheduled once it runs out of budget") + @Test + @Timeout(10) + void testScheduleStopsWhenExhausted() { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + try { + BackgroundTask task = Tasks.backgroundTask() + .withScheduledExecutor(executor) + .withBudget(Budgets.iterationTimeBudget() + .withInterval(Duration.ofMillis(100)) + .withInitialDelay(Duration.ZERO) + .withMaxIterations(maxIterations) + .build()) + .build(); + + Future<?> future = task.schedule(camelContext, this::booleanSupplier); + + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertTrue(future.isCancelled(), + "An exhausted task should not stay scheduled")); + assertEquals(maxIterations, taskCount.intValue()); + assertEquals(Task.Status.Exhausted, task.getStatus()); + } finally { + executor.shutdownNow(); + } + } + + @DisplayName("Test that a cancelled task is unscheduled and leaves the task registry") + @Test + @Timeout(20) + void testCancelUnschedulesAndDeregisters() { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + try { + BackgroundTask task = Tasks.backgroundTask() + .withScheduledExecutor(executor) + .withBudget(Budgets.iterationTimeBudget() + .withInterval(Duration.ofMillis(100)) + .withInitialDelay(Duration.ZERO) + .withUnlimitedDuration() + .build()) + .withName("cancelled") + .build(); + + TaskManagerRegistry registry = PluginHelper.getTaskManagerRegistry(camelContext.getCamelContextExtension()); + Future<?> future = task.schedule(camelContext, this::booleanSupplier); + await().atMost(5, TimeUnit.SECONDS).until(() -> registry.getTasks().contains(task)); + + task.cancel(false); + + assertTrue(future.isCancelled(), "A cancelled task should not stay scheduled"); + // a run that had already started may still have re-added itself, it then removes itself again + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertFalse(registry.getTasks().contains(task), + "A cancelled task should not stay in the task registry")); + assertEquals(Task.Status.Inactive, task.getStatus()); + assertFalse(task.isRunning(), "A cancelled task should not report itself as running"); + + int attempts = taskCount.intValue(); + await().pollDelay(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS).untilAsserted( + () -> assertEquals(attempts, taskCount.intValue(), "A cancelled task should not run again")); + } finally { + executor.shutdownNow(); + } + } + + @DisplayName("Test that cancelling a task before its first run leaves nothing behind") + @Test + @Timeout(20) + void testCancelBeforeTheFirstRun() { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + try { + BackgroundTask task = Tasks.backgroundTask() + .withScheduledExecutor(executor) + .withBudget(Budgets.iterationTimeBudget() + .withInterval(Duration.ofMillis(100)) + // long enough that the cancel below lands before the first run + .withInitialDelay(Duration.ofSeconds(3)) + .withUnlimitedDuration() + .build()) + .withName("cancelled-before-first-run") + .build(); + + TaskManagerRegistry registry = PluginHelper.getTaskManagerRegistry(camelContext.getCamelContextExtension()); + Future<?> future = task.schedule(camelContext, this::booleanSupplier); + + task.cancel(false); + + assertTrue(future.isCancelled(), "A cancelled task should not stay scheduled"); + assertFalse(registry.getTasks().contains(task), "A cancelled task should not stay in the task registry"); + await().pollDelay(1, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(0, + taskCount.intValue(), "The supplier of a task cancelled before its first run should never run")); + } finally { + executor.shutdownNow(); + } + } + + @DisplayName("Test that cancelling a task that already completed keeps the outcome of its last run") + @Test + @Timeout(20) + void testCancelKeepsTheOutcomeOfACompletedTask() { + ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + try { + BackgroundTask task = Tasks.backgroundTask() + .withScheduledExecutor(executor) + .withBudget(Budgets.iterationTimeBudget() + .withInterval(Duration.ofMillis(100)) + .withInitialDelay(Duration.ZERO) + .withUnlimitedDuration() + .build()) + .withName("completed-then-cancelled") + .build(); + + task.schedule(camelContext, () -> true); + await().atMost(5, TimeUnit.SECONDS).until(() -> task.getStatus() == Task.Status.Completed); + + // a caller that cancels defensively must not undo the success of the task + task.cancel(false); + + assertEquals(Task.Status.Completed, task.getStatus()); + } finally { + executor.shutdownNow(); + } + } } diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java index c49b1584a6f3..41fade98bf95 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java @@ -24,6 +24,7 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.BooleanSupplier; import org.apache.camel.CamelContext; @@ -81,6 +82,12 @@ public class BackgroundTask extends AbstractTask implements BlockingTask { private Duration elapsed = Duration.ZERO; private final AtomicBoolean running = new AtomicBoolean(); private final AtomicBoolean completed = new AtomicBoolean(); + // only set when scheduled via schedule(), run() cancels the future it owns itself + private final AtomicReference<Future<?>> scheduledFuture = new AtomicReference<>(); + // the context the schedule was made with, so cancel() can deregister without being handed it again + private final AtomicReference<CamelContext> scheduledContext = new AtomicReference<>(); + private volatile boolean registeredByRun; + private volatile boolean attempting; BackgroundTask(TimeBudget budget, ScheduledExecutorService service, String name) { super(name); @@ -91,22 +98,33 @@ public class BackgroundTask extends AbstractTask implements BlockingTask { private void runTaskWrapper(CamelContext camelContext, BooleanSupplier supplier) { LOG.trace("Current latch value: {}", latch.getCount()); if (latch.getCount() == 0) { + // the task is done and every further run is a no-op, so stop being rescheduled + unschedule(false); return; } TaskManagerRegistry registry = null; if (camelContext != null) { registry = PluginHelper.getTaskManagerRegistry(camelContext.getCamelContextExtension()); - registry.addTask(this); + if (!registeredByRun) { + registry.addTask(this); + if (latch.getCount() == 0) { + // cancelled while this run was starting up, so undo the registration just made + registry.removeTask(this); + unschedule(false); + return; + } + } } if (!budget.next()) { LOG.warn("The task {} does not have more budget to continue running", getName()); status = Status.Exhausted; completed.set(false); - if (registry != null) { + if (!registeredByRun && registry != null) { registry.removeTask(this); } latch.countDown(); + unschedule(false); return; } @@ -118,18 +136,27 @@ public class BackgroundTask extends AbstractTask implements BlockingTask { if (doRun(supplier)) { status = Status.Completed; completed.set(true); - if (registry != null) { + if (!registeredByRun && registry != null) { registry.removeTask(this); } latch.countDown(); + unschedule(false); LOG.trace("Task {} succeeded and the current task is unscheduled: {}", getName(), latch.getCount()); } } catch (Exception e) { status = Status.Failed; + completed.set(false); cause = e; + // release the blocking run() caller and unregister; without this a task built with + // withUnlimitedDuration() would await() forever and stay in the registry (CAMEL-24286) + if (!registeredByRun && registry != null) { + registry.removeTask(this); + } + latch.countDown(); throw e; } - nextAttemptTime = lastAttemptTime + budget.interval(); + // scheduleWithFixedDelay waits interval after this run finishes, so compute from now + nextAttemptTime = System.currentTimeMillis() + budget.interval(); } /** @@ -142,28 +169,95 @@ public class BackgroundTask extends AbstractTask implements BlockingTask { */ public Future<?> schedule(CamelContext camelContext, BooleanSupplier supplier) { running.set(true); - return service.scheduleWithFixedDelay(() -> runTaskWrapper(camelContext, supplier), budget.initialDelay(), - budget.interval(), TimeUnit.MILLISECONDS); + Future<?> future = service.scheduleWithFixedDelay(() -> runTaskWrapper(camelContext, supplier), + budget.initialDelay(), budget.interval(), TimeUnit.MILLISECONDS); + scheduledContext.set(camelContext); + scheduledFuture.set(future); + if (latch.getCount() == 0) { + // the task already finished before the future was published, so it could not unschedule itself + unschedule(false); + } + return future; + } + + /** + * Cancels a task scheduled with {@link #schedule(CamelContext, BooleanSupplier)} that is no longer needed, and + * removes it from the {@link TaskManagerRegistry}. A scheduled task deregisters itself from one of its runs, which + * is not going to happen once the schedule is cancelled, so cancelling the returned {@link Future} directly leaves + * the task behind in the registry. + * <p/> + * This does not wait for an attempt that is already running: with {@code mayInterruptIfRunning} false, a supplier + * call that is in progress runs to completion after this method returns. {@link #isRunning()} answers for the + * schedule and turns false here even then, so {@link #isAttempting()} is the one to ask whether an attempt is still + * in flight. + * <p/> + * A task that already completed, failed or exhausted its budget keeps the outcome of its last run. Only the + * schedule of a task that is still {@link Status#Active} is cancelled, which turns it {@link Status#Inactive}. + * + * @param mayInterruptIfRunning whether the thread of an attempt that is currently running should be interrupted + */ + public void cancel(boolean mayInterruptIfRunning) { + // any run that has not started yet becomes a no-op + latch.countDown(); + unschedule(mayInterruptIfRunning); + if (status == Status.Active) { + status = Status.Inactive; + completed.set(false); + } + deregister(); + running.set(false); + } + + /** + * Cancels the repeating schedule created by {@link #schedule(CamelContext, BooleanSupplier)}, so a task that has + * nothing left to do does not keep occupying the scheduler for the lifetime of its executor. + */ + private void unschedule(boolean mayInterruptIfRunning) { + Future<?> future = scheduledFuture.getAndSet(null); + if (future != null) { + future.cancel(mayInterruptIfRunning); + } + } + + private void deregister() { + CamelContext context = scheduledContext.getAndSet(null); + if (context != null) { + TaskManagerRegistry registry = PluginHelper.getTaskManagerRegistry(context.getCamelContextExtension()); + if (registry != null) { + registry.removeTask(this); + } + } } @Override public boolean run(CamelContext camelContext, BooleanSupplier supplier) { running.set(true); + registeredByRun = true; + if (camelContext != null) { + PluginHelper.getTaskManagerRegistry(camelContext.getCamelContextExtension()).addTask(this); + } Future<?> task = service.scheduleWithFixedDelay(() -> runTaskWrapper(camelContext, supplier), budget.initialDelay(), budget.interval(), TimeUnit.MILLISECONDS); waitForTaskCompletion(camelContext, task); return completed.get(); } + @Override + public boolean isAttempting() { + return attempting; + } + protected boolean doRun(BooleanSupplier supplier) { try { - cause = null; + attempting = true; return supplier.getAsBoolean(); } catch (TaskRunFailureException e) { LOG.debug("Task {} failed at {} iterations and will attempt again on next interval: {}", getName(), budget.iteration(), e.getMessage()); cause = e; return false; + } finally { + attempting = false; } } @@ -180,20 +274,20 @@ public class BackgroundTask extends AbstractTask implements BlockingTask { LOG.debug("The task has finished the execution and it is ready to continue"); } } - - TaskManagerRegistry registry = null; - if (camelContext != null) { - registry = PluginHelper.getTaskManagerRegistry(camelContext.getCamelContextExtension()); - } - if (registry != null) { - registry.removeTask(this); - } - - task.cancel(true); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for the repeatable task to execute: {}", e.getMessage(), e); Thread.currentThread().interrupt(); } finally { + // unregister and cancel even if the await was interrupted, otherwise the task leaks in + // the registry and the scheduled future keeps running (CAMEL-24286) + if (camelContext != null) { + TaskManagerRegistry registry + = PluginHelper.getTaskManagerRegistry(camelContext.getCamelContextExtension()); + if (registry != null) { + registry.removeTask(this); + } + } + task.cancel(true); elapsed = budget.elapsed(); running.set(false); } diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java b/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java index 5f74bf5b0f4d..72cfa586bdf4 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java @@ -81,6 +81,14 @@ public interface Task { */ long getNextAttemptTime(); + /** + * Whether the task is currently attempting to run its supplier (true), or waiting for the next scheduled tick + * (false). + */ + default boolean isAttempting() { + return false; + } + /** * The task failed for some un-expected exception */
