From f7c0fe466c4c7048258878f480035695d15b6a47 Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Wed, 22 Jul 2026 14:57:48 +0200 Subject: [PATCH] CAMEL-24242: camel-aws2-athena - stop ignoring thread interruption while polling a query Athena2QueryHelper.interrupted gates both shouldAttempt() and shouldWait(), but nothing assigned it any more, so both guards were dead code. CAMEL-20297 had added interrupt handling in doWait(); the CAMEL-22949 migration to Camel's Task API (commit 1b0fca17c0) replaced the Thread.sleep() block along with the catch clause that was its only writer, silently reverting it. ForegroundTask.run() does restore the interrupt status and return false, but doWait() discarded that result. The consequence was worse than a missed shutdown signal: once the interrupt status is set, every subsequent Thread.sleep() inside the task returns immediately, so the polling loop spun with no delay, issuing GetQueryExecution calls as fast as the API allowed until waitTimeout elapsed. Capture the return value and record the interruption, restoring CAMEL-20297's behaviour on top of the Task API. Co-authored-by: Claude Opus 4.8 Signed-off-by: Andrea Cosentino --- .../camel-aws/camel-aws2-athena/pom.xml | 5 +++ .../aws2/athena/Athena2QueryHelper.java | 10 +++++- .../aws2/athena/Athena2QueryHelperTest.java | 31 +++++++++++++++++++ 3 files changed, 45 insertions(+), 1 deletion(-) diff --git a/components/camel-aws/camel-aws2-athena/pom.xml b/components/camel-aws/camel-aws2-athena/pom.xml index b9cf0dd87c1de..1048efd4f173f 100644 --- a/components/camel-aws/camel-aws2-athena/pom.xml +++ b/components/camel-aws/camel-aws2-athena/pom.xml @@ -76,6 +76,11 @@ ${mockito-version} test + + org.assertj + assertj-core + test + diff --git a/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2QueryHelper.java b/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2QueryHelper.java index f4cc50e95c062..9bfa3c48f0519 100644 --- a/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2QueryHelper.java +++ b/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2QueryHelper.java @@ -169,7 +169,7 @@ boolean shouldWait() { void doWait() { // Use Camel's task API for polling delay instead of Thread.sleep() // We use initialDelay for the actual delay, and maxIterations(1) to run once - Tasks.foregroundTask() + boolean completed = Tasks.foregroundTask() .withBudget(Budgets.iterationBudget() .withMaxIterations(1) .withInitialDelay(Duration.ofMillis(this.currentDelay)) @@ -179,6 +179,14 @@ void doWait() { .build() .run(exchange.getContext(), () -> true); + if (!completed && Thread.currentThread().isInterrupted()) { + // the task API already restored the interrupt status, so only record it here to let the + // attempt and wait loops bail out at the earliest opportunity + this.interrupted = true; + LOG.trace( + "AWS Athena start query execution wait thread was interrupted; will return at earliest opportunity"); + } + this.currentDelay = this.delay; } diff --git a/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2QueryHelperTest.java b/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2QueryHelperTest.java index db81f368106c0..2d668168f85d6 100644 --- a/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2QueryHelperTest.java +++ b/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2QueryHelperTest.java @@ -28,6 +28,7 @@ import software.amazon.awssdk.services.athena.model.QueryExecutionState; import software.amazon.awssdk.services.athena.model.QueryExecutionStatus; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -155,6 +156,36 @@ public void testStartQueryExecutionHappyPath() { assertFalse(helper.shouldAttempt()); } + @Test + void doWaitRecordsThreadInterruptionSoTheLoopsStop() { + Athena2Configuration configuration = new Athena2Configuration(); + configuration.setMaxAttempts(3); + configuration.setWaitTimeout(60_000); + configuration.setInitialDelay(10_000); + configuration.setDelay(10_000); + + Athena2QueryHelper helper = new Athena2QueryHelper( + new DefaultExchange(new DefaultCamelContext()), + configuration); + + helper.markAttempt(); + assertThat(helper.shouldWait()).isTrue(); + + Thread.currentThread().interrupt(); + try { + helper.doWait(); + + // without this, the wait loop keeps polling Athena with no delay at all, because every + // subsequent sleep returns immediately while the interrupt status is still set + assertThat(helper.isInterrupted()).isTrue(); + assertThat(helper.shouldWait()).isFalse(); + assertThat(helper.shouldAttempt()).isFalse(); + } finally { + // clear the interrupt status so it does not leak into the following tests + Thread.interrupted(); + } + } + @Test public void isComplete() { Athena2QueryHelper helper = defaultAthena2QueryHelper();