This is an automated email from the ASF dual-hosted git repository.
oscerd 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 ee9df19a53b6 CAMEL-24286: Release the latch and unregister when a
BackgroundTask supplier throws (#25217)
ee9df19a53b6 is described below
commit ee9df19a53b6472ac08d077e7e23aa13c991754e
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 30 10:14:07 2026 +0200
CAMEL-24286: Release the latch and unregister when a BackgroundTask
supplier throws (#25217)
BackgroundTask.runTaskWrapper rethrew a supplier exception on the scheduler
thread without counting down the completion latch or unregistering the task.
doRun only catches TaskRunFailureException, so any other exception reached
this
path. scheduleWithFixedDelay then suppresses all further executions, and
for a
task built with withUnlimitedDuration() waitForTaskCompletion calls
latch.await() with no timeout — so the calling thread blocked forever and
the
task leaked in the TaskManagerRegistry.
This is reachable on main via CAMEL-24272: camel-ftp SFTP (tryConnect
catches
only JSchException) and camel-mongodb-gridfs (its Mongo cursor/find calls
are
outside the inner catch) both run an unlimited-duration reconnection through
BackgroundTask.run().
Fix: in the catch, mark the task not-completed, unregister it, and count
down
the latch before rethrowing, so the blocking run() caller unblocks and
observes
the failure. The rethrow is kept so the schedule() path still surfaces the
error to the executor.
Also move removeTask + task.cancel into the finally of
waitForTaskCompletion,
so an InterruptedException from latch.await() no longer leaves the task
registered.
Add a regression test that runs an unlimited-duration task whose supplier
throws; it fails (run() hangs on the join) without the fix and passes with
it.
The runner is a daemon thread so a regression cannot wedge the JVM.
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
.../task/task/BackgroundTaskRegistryTest.java | 43 ++++++++++++++++++++++
.../apache/camel/support/task/BackgroundTask.java | 27 +++++++++-----
2 files changed, 60 insertions(+), 10 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskRegistryTest.java
b/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskRegistryTest.java
index 3f0154d49802..011228af1275 100644
---
a/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskRegistryTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/support/task/task/BackgroundTaskRegistryTest.java
@@ -128,6 +128,49 @@ class BackgroundTaskRegistryTest {
assertThat(registry.getTasks()).isEmpty();
}
+ @DisplayName("Test that run() returns and unregisters when the supplier
throws (CAMEL-24286)")
+ @Test
+ @Timeout(15)
+ void testRunReturnsAndUnregistersWhenSupplierThrows() throws Exception {
+ AtomicBoolean runReturned = new AtomicBoolean(false);
+ AtomicBoolean completed = new AtomicBoolean(true);
+
+ // an unlimited-duration task: before CAMEL-24286 a thrown supplier
exception left the latch
+ // un-counted, so run() blocked forever here and the task leaked in
the registry
+ BackgroundTask task = Tasks.backgroundTask()
+
.withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(100))
+ .withInitialDelay(Duration.ZERO)
+ .withUnlimitedDuration()
+ .build())
+ .build();
+
+ // daemon so a regression (run() hanging forever) cannot keep the JVM
alive
+ Thread runner = new Thread(() -> {
+ boolean result = task.run(camelContext, () -> {
+ throw new IllegalStateException("boom");
+ });
+ completed.set(result);
+ runReturned.set(true);
+ });
+ runner.setDaemon(true);
+ runner.start();
+
+ runner.join(TimeUnit.SECONDS.toMillis(10));
+
+ assertThat(runReturned.get())
+ .as("run() must return when the supplier throws, not block
forever")
+ .isTrue();
+ assertThat(completed.get())
+ .as("run() should report the task as not completed after a
thrown exception")
+ .isFalse();
+ await().atMost(2, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertThat(registry.getTasks())
+ .as("the failed task must not leak in the registry")
+ .isEmpty());
+ }
+
@DisplayName("Test that task is removed from registry after supplier
succeeds")
@Test
@Timeout(10)
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 5065391ca38e..c8776c352898 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
@@ -130,7 +130,14 @@ public class BackgroundTask extends AbstractTask
implements BlockingTask {
}
} 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;
}
// scheduleWithFixedDelay waits interval after this run finishes, so
compute from now
@@ -196,20 +203,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);
}
