You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2026/07/23 13:25:33 UTC
(camel) branch camel-4.18.x updated: CAMEL-24242: camel-aws2-athena - stop ignoring thread interruption while polling a query (#25032)
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 <an...@gmail.com>
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 <an...@gmail.com>
Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
---
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();