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 ba83ac200916 CAMEL-24242: camel-aws2-athena - stop ignoring thread
interruption while polling a query (#25029)
ba83ac200916 is described below
commit ba83ac2009168c1d8a214dd738d9478300e23b51
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 23 15:24:54 2026 +0200
CAMEL-24242: camel-aws2-athena - stop ignoring thread interruption while
polling a query (#25029)
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.
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 b9cf0dd87c1d..1048efd4f173 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();
