gnodet commented on code in PR #26112:
URL: https://github.com/apache/camel/pull/26112#discussion_r3940597962
##########
core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java:
##########
@@ -163,22 +171,53 @@ public Future<?> schedule(CamelContext camelContext,
BooleanSupplier supplier) {
running.set(true);
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();
+ 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.
+ *
+ * @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();
Review Comment:
💡 **Observation (low):** `cancel()` sets `running.set(false)` at the end,
but a concurrent `runTaskWrapper()` that is already past the latch check and
executing the supplier will still be running. After `cancel()` returns,
`isRunning()` returns `false` while an attempt may still be in progress (with
`mayInterruptIfRunning=false`).
This is probably fine in practice — callers that need to wait for the
in-flight attempt to finish would use `mayInterruptIfRunning=true` or await the
future — but it means `isRunning()` can briefly lie after `cancel(false)`.
Worth a comment on the Javadoc noting that `cancel(false)` does not wait for an
in-flight supplier call to complete.
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -315,8 +329,12 @@ private boolean isStartPending() {
}
private void cancelLeaderTask(boolean mayInterruptIfRunning) {
- if (leaderTaskFuture != null) {
- leaderTaskFuture.cancel(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;
Review Comment:
💡 **Observation (medium):** The `cancelLeaderTask` method nulls `leaderTask`
and `leaderTaskFuture` but these fields are not volatile. They're always
accessed under `leadershipLock`, so this is safe — but it's worth noting this
invariant somewhere since the fields are declared next to the volatile
`delegatedConsumer` and `view` at lines 64-65, and a future reader might wonder
why some fields are volatile and others aren't.
The `leaderTask` field could arguably be volatile for consistency with the
surrounding declarations, but since all accesses are lock-guarded it's
unnecessary overhead.
##########
components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java:
##########
@@ -224,6 +230,66 @@ void testAllConfiguredStartAttemptsAreMade() {
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));
Review Comment:
💡 **Nice test.** `testEventDispatchIsNotBlockedByALifecycleOperation` is a
well-designed deadlock detection test — holding the service lock via a blocking
suspend, then verifying the dispatch thread completes without waiting. The
`join(20s)` timeout with the alive check is the right pattern for this.
##########
core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java:
##########
@@ -163,22 +171,53 @@ public Future<?> schedule(CamelContext camelContext,
BooleanSupplier supplier) {
running.set(true);
Future<?> future = service.scheduleWithFixedDelay(() ->
runTaskWrapper(camelContext, supplier),
budget.initialDelay(), budget.interval(),
TimeUnit.MILLISECONDS);
+ scheduledContext.set(camelContext);
Review Comment:
💡 **Observation (low):** `scheduledContext` is set after `scheduledFuture`
(line 174 vs 173), but `cancel()` reads `scheduledContext` in `deregister()`
and `scheduledFuture` in `unschedule()` independently. There's a theoretical
window where `cancel()` is called between the two `set()` calls — it would
`unschedule` the future but `deregister()` would see a null context and skip
the registry removal.
The race guard in `runTaskWrapper` (lines 108-113) catches this: a run that
started before the cancel saw the latch will deregister on its next check. And
the existing `if (latch.getCount() == 0) { unschedule(false); }` at line 176
catches the reverse. So the window is covered by defense-in-depth, but it might
be slightly cleaner to set both atomically (or at least set `scheduledContext`
first, before `scheduledFuture`, since `cancel()` checks the future first).
Looking again — `scheduledContext` IS set before `scheduledFuture` at lines
173-174. So if `cancel()` fires between them, it sees the context (deregisters)
but doesn't see the future yet (skips unschedule). Then line 174 publishes the
future, and the final `latch.getCount() == 0` check at line 176 catches it. 👍
Order is correct.
##########
components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java:
##########
@@ -109,14 +116,19 @@ protected void doStart() throws Exception {
protected void doStop() throws Exception {
super.doStop();
- // 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);
+ 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 cluster view lock while
this thread holds the lock of
- // this service, which is the opposite order of an event dispatch.
Nothing that runs under this lock
- // may wait for the view, and the listener bails out before locking
once this consumer is stopping
+ // 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
Review Comment:
💡 **Observation (low):** `doStop()` acquires `leadershipLock`, clears state,
releases the lock, then calls `view.removeEventListener()`. Between the lock
release and the `removeEventListener`, a leadership event could arrive, pass
the `isRunAllowed()` fast check (which returns `false` since `super.doStop()`
ran first), and bail out — correct. But the comment on line 128-131 explaining
this ordering is valuable. It might be worth adding that `super.doStop()` has
already run at this point, so `isRunAllowed()` is the first gate the listener
hits.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
