This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 632c06923ca9 CAMEL-24242: camel-aws2-athena - stop ignoring thread
interruption while polling a query (#25032)
632c06923ca9 is described below
commit 632c06923ca9c24d4d559b4d5d6a4cea4a7fb90b
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 23 15:25:25 2026 +0200
CAMEL-24242: camel-aws2-athena - stop ignoring thread interruption while
polling a query (#25032)
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.
(cherry picked from commit f7c0fe466c4c7048258878f480035695d15b6a47)
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
components/camel-aws/camel-aws2-athena/pom.xml | 5 ++++
.../component/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 5291ecd77164..e7551ad19614 100644
--- a/components/camel-aws/camel-aws2-athena/pom.xml
+++ b/components/camel-aws/camel-aws2-athena/pom.xml
@@ -76,6 +76,11 @@
<version>${mockito-version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
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 f4cc50e95c06..9bfa3c48f051 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 @@ class Athena2QueryHelper {
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 @@ class Athena2QueryHelper {
.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 db81f368106c..2d668168f85d 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.QueryExecution;
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 class Athena2QueryHelperTest {
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();
