This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26100-athena-output-location in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0d0abe375c0919c75db17bd53ff4ce33a337c8a0 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 16:10:19 2026 +0100 CAMEL-26100: camel-aws2-athena - result headers are not taken as the input of a later operation The CamelAwsAthenaOutputLocation header that getQueryExecution, getQueryResults with S3Pointer and startQueryExecution with waitTimeout set is the result file of that query, and a later startQueryExecution read it as the location for its results. The CamelAwsAthenaNextToken header of listQueryExecutions was sent by a later getQueryResults, and the other way around. The producer now remembers the values it set in exchange properties, and ignores the header when it still holds that value: the output location for startQueryExecution, the next token unless it is the same operation (and query). A header the route sets is used as before. A query execution without a result configuration no longer fails with a NullPointerException. The docs say that the S3Pointer location is given whatever the state of the query. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/catalog/docs/aws2-athena-component.adoc | 15 +- .../src/main/docs/aws2-athena-component.adoc | 15 +- .../component/aws2/athena/Athena2Producer.java | 97 ++++++-- .../athena/Athena2ProducerResultHeadersTest.java | 253 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++ 5 files changed, 367 insertions(+), 26 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-athena-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-athena-component.adoc index 4e2973b34b4d..ed4cd29fa3c2 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-athena-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-athena-component.adoc @@ -362,6 +362,17 @@ YAML:: Will cause the output location to be `s3://other/location/`. +The `CamelAwsAthenaOutputLocation` and `CamelAwsAthenaNextToken` headers are also results of the operations. A result +is not taken as the input of a later operation in the same route: + +* The `CamelAwsAthenaOutputLocation` header set by getQueryExecution, by getQueryResults with `S3Pointer`, and by + startQueryExecution with a `waitTimeout` is the file with the results of that query. A later startQueryExecution does + not use it, and puts its results under the `outputLocation` option instead. +* The `CamelAwsAthenaNextToken` header that listQueryExecutions sets is used only by a later listQueryExecutions, and + the one that getQueryResults sets only by a later getQueryResults for the same query. + +A header that the route sets itself is always used. + === Athena Producer Operation examples - getQueryExecution: this operation returns information about a query given its query execution ID @@ -680,7 +691,9 @@ YAML:: ==== The preceding example will return the S3 location of the results of the query as text in the body. -The location will also be set in a header (`CamelAwsAthenaOutputLocation`). +The location will also be set in a header (`CamelAwsAthenaOutputLocation`). The location is given whatever the state +of the query: the file is complete only when the `CamelAwsAthenaQueryExecutionState` header is `SUCCEEDED`. A query +without a result configuration, as with managed query results, has no location, and the body and header are null. - listQueryExecutions: this operation returns a list of query execution IDs diff --git a/components/camel-aws/camel-aws2-athena/src/main/docs/aws2-athena-component.adoc b/components/camel-aws/camel-aws2-athena/src/main/docs/aws2-athena-component.adoc index 4e2973b34b4d..ed4cd29fa3c2 100644 --- a/components/camel-aws/camel-aws2-athena/src/main/docs/aws2-athena-component.adoc +++ b/components/camel-aws/camel-aws2-athena/src/main/docs/aws2-athena-component.adoc @@ -362,6 +362,17 @@ YAML:: Will cause the output location to be `s3://other/location/`. +The `CamelAwsAthenaOutputLocation` and `CamelAwsAthenaNextToken` headers are also results of the operations. A result +is not taken as the input of a later operation in the same route: + +* The `CamelAwsAthenaOutputLocation` header set by getQueryExecution, by getQueryResults with `S3Pointer`, and by + startQueryExecution with a `waitTimeout` is the file with the results of that query. A later startQueryExecution does + not use it, and puts its results under the `outputLocation` option instead. +* The `CamelAwsAthenaNextToken` header that listQueryExecutions sets is used only by a later listQueryExecutions, and + the one that getQueryResults sets only by a later getQueryResults for the same query. + +A header that the route sets itself is always used. + === Athena Producer Operation examples - getQueryExecution: this operation returns information about a query given its query execution ID @@ -680,7 +691,9 @@ YAML:: ==== The preceding example will return the S3 location of the results of the query as text in the body. -The location will also be set in a header (`CamelAwsAthenaOutputLocation`). +The location will also be set in a header (`CamelAwsAthenaOutputLocation`). The location is given whatever the state +of the query: the file is complete only when the `CamelAwsAthenaQueryExecutionState` header is `SUCCEEDED`. A query +without a result configuration, as with managed query results, has no location, and the body and header are null. - listQueryExecutions: this operation returns a list of query execution IDs diff --git a/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2Producer.java b/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2Producer.java index ec83a1524a11..1efaad130029 100644 --- a/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2Producer.java +++ b/components/camel-aws/camel-aws2-athena/src/main/java/org/apache/camel/component/aws2/athena/Athena2Producer.java @@ -39,6 +39,7 @@ import software.amazon.awssdk.services.athena.model.GetQueryResultsRequest; import software.amazon.awssdk.services.athena.model.GetQueryResultsResponse; import software.amazon.awssdk.services.athena.model.ListQueryExecutionsRequest; import software.amazon.awssdk.services.athena.model.ListQueryExecutionsResponse; +import software.amazon.awssdk.services.athena.model.QueryExecution; import software.amazon.awssdk.services.athena.model.QueryExecutionContext; import software.amazon.awssdk.services.athena.model.ResultConfiguration; import software.amazon.awssdk.services.athena.model.StartQueryExecutionRequest; @@ -52,6 +53,16 @@ public class Athena2Producer extends DefaultProducer { private static final Logger LOG = LoggerFactory.getLogger(Athena2Producer.class); + // the output location and the next token that an operation set as its result, kept in exchange properties, so a + // later operation does not take them as its input: the output location of a query is the file of its results, and + // a next token is only valid for the operation (and query) that returned it + private static final String OUTPUT_LOCATION_RESULT = "CamelAwsAthenaOutputLocationResult"; + private static final String NEXT_TOKEN_RESULT = "CamelAwsAthenaNextTokenResult"; + private static final String LIST_QUERY_EXECUTIONS = "listQueryExecutions"; + + private record NextTokenResult(String nextToken, String readBy) { + } + private HealthCheck producerHealthCheck; private WritableHealthCheckRepository healthCheckRepository; @@ -104,8 +115,7 @@ public class Athena2Producer extends DefaultProducer { Message message = getMessageForResponse(exchange); message.setHeader(Athena2Constants.QUERY_EXECUTION_ID, getQueryExecutionResponse.queryExecution().queryExecutionId()); message.setHeader(Athena2Constants.QUERY_EXECUTION_STATE, getQueryExecutionResponse.queryExecution().status().state()); - message.setHeader(Athena2Constants.OUTPUT_LOCATION, - getQueryExecutionResponse.queryExecution().resultConfiguration().outputLocation()); + setOutputLocationResult(exchange, outputLocationOf(getQueryExecutionResponse.queryExecution())); message.setBody(getQueryExecutionResponse); } @@ -133,15 +143,14 @@ public class Athena2Producer extends DefaultProducer { } else if (outputType == Athena2OutputType.SelectList) { GetQueryResultsRequest request = doGetQueryResultsRequest(queryExecutionId, exchange).build(); GetQueryResultsResponse response = athenaClient.getQueryResults(request); - message.setHeader(Athena2Constants.NEXT_TOKEN, response.nextToken()); + setNextTokenResult(exchange, response.nextToken(), nextTokenReader(queryExecutionId)); message.setHeader(Athena2Constants.IS_TRUNCATED, ObjectHelper.isNotEmpty(response.nextToken())); message.setBody(response); } else if (outputType == Athena2OutputType.S3Pointer) { GetQueryExecutionResponse response = doGetQueryExecution(queryExecutionId, athenaClient); - String outputLocation = response.queryExecution().resultConfiguration().outputLocation(); + String outputLocation = outputLocationOf(response.queryExecution()); message.setHeader(Athena2Constants.QUERY_EXECUTION_STATE, response.queryExecution().status().state()); - message.setHeader(Athena2Constants.OUTPUT_LOCATION, - response.queryExecution().resultConfiguration().outputLocation()); + setOutputLocationResult(exchange, outputLocation); message.setBody(outputLocation); } else { throw new IllegalArgumentException( @@ -167,7 +176,7 @@ public class Athena2Producer extends DefaultProducer { Integer maxResults = determineMaxResults(exchange); request.maxResults(maxResults); - String nextToken = determineNextToken(exchange); + String nextToken = determineNextToken(exchange, nextTokenReader(queryExecutionId)); request.nextToken(nextToken); return request; @@ -182,7 +191,7 @@ public class Athena2Producer extends DefaultProducer { Integer maxResults = determineMaxResults(exchange); request.maxResults(maxResults); - String nextToken = determineNextToken(exchange); + String nextToken = determineNextToken(exchange, LIST_QUERY_EXECUTIONS); request.nextToken(nextToken); String workGroup = determineWorkGroup(exchange); @@ -190,7 +199,7 @@ public class Athena2Producer extends DefaultProducer { ListQueryExecutionsResponse response = athenaClient.listQueryExecutions(request.build()); Message message = getMessageForResponse(exchange); - message.setHeader(Athena2Constants.NEXT_TOKEN, response.nextToken()); + setNextTokenResult(exchange, response.nextToken(), LIST_QUERY_EXECUTIONS); message.setHeader(Athena2Constants.IS_TRUNCATED, ObjectHelper.isNotEmpty(response.nextToken())); message.setBody(response); } @@ -221,8 +230,8 @@ public class Athena2Producer extends DefaultProducer { message.setHeader(Athena2Constants.QUERY_EXECUTION_ID, queryExecutionId); message.setHeader(Athena2Constants.QUERY_EXECUTION_STATE, getQueryExecutionResponse == null ? null : getQueryExecutionResponse.queryExecution().status().state()); - message.setHeader(Athena2Constants.OUTPUT_LOCATION, getQueryExecutionResponse == null - ? null : getQueryExecutionResponse.queryExecution().resultConfiguration().outputLocation()); + setOutputLocationResult(exchange, getQueryExecutionResponse == null + ? null : outputLocationOf(getQueryExecutionResponse.queryExecution())); message.setHeader(Athena2Constants.START_QUERY_EXECUTION_ATTEMPTS, athena2QueryHelper.getAttempts()); message.setHeader(Athena2Constants.START_QUERY_EXECUTION_ELAPSED_MILLIS, athena2QueryHelper.getElapsedMillis()); @@ -310,13 +319,31 @@ public class Athena2Producer extends DefaultProducer { return ObjectHelper.isNotEmpty(includeTrace) ? includeTrace : false; } - private String determineNextToken(final Exchange exchange) { - return getOptionalHeader( - exchange, - Athena2Constants.NEXT_TOKEN, - String.class, - () -> getConfiguration().getNextToken(), - "next token"); + private String determineNextToken(final Exchange exchange, String reader) { + String nextToken = exchange.getIn().getHeader(Athena2Constants.NEXT_TOKEN, String.class); + NextTokenResult result = exchange.getProperty(NEXT_TOKEN_RESULT, NextTokenResult.class); + if (ObjectHelper.isNotEmpty(nextToken) && result != null && nextToken.equals(result.nextToken()) + && !reader.equals(result.readBy())) { + LOG.trace("AWS Athena next token is the result of another operation or query, not using it"); + nextToken = null; + } + if (ObjectHelper.isEmpty(nextToken)) { + nextToken = getConfiguration().getNextToken(); + LOG.trace("AWS Athena next token is missing, using default one [{}]", nextToken); + } + return nextToken; + } + + /** + * The operation (and query) that a next token is valid for. + */ + private static String nextTokenReader(String queryExecutionId) { + return "getQueryResults " + queryExecutionId; + } + + private static void setNextTokenResult(Exchange exchange, String nextToken, String readBy) { + exchange.getMessage().setHeader(Athena2Constants.NEXT_TOKEN, nextToken); + exchange.setProperty(NEXT_TOKEN_RESULT, new NextTokenResult(nextToken, readBy)); } private String determineClientRequestToken(final Exchange exchange) { @@ -394,12 +421,34 @@ public class Athena2Producer extends DefaultProducer { } private String determineOutputLocation(final Exchange exchange) { - return getRequiredHeader( - exchange, - Athena2Constants.OUTPUT_LOCATION, - String.class, - () -> getConfiguration().getOutputLocation(), - "AWS Athena output location is required."); + String outputLocation = exchange.getIn().getHeader(Athena2Constants.OUTPUT_LOCATION, String.class); + if (ObjectHelper.isNotEmpty(outputLocation) + && outputLocation.equals(exchange.getProperty(OUTPUT_LOCATION_RESULT, String.class))) { + // the file with the results of an earlier query is not where this query puts its results + LOG.trace("AWS Athena output location is the result of an earlier operation, not using it"); + outputLocation = null; + } + if (ObjectHelper.isEmpty(outputLocation)) { + outputLocation = getConfiguration().getOutputLocation(); + } + if (ObjectHelper.isEmpty(outputLocation)) { + throw new IllegalArgumentException("AWS Athena output location is required."); + } + return outputLocation; + } + + private static void setOutputLocationResult(Exchange exchange, String outputLocation) { + exchange.getMessage().setHeader(Athena2Constants.OUTPUT_LOCATION, outputLocation); + exchange.setProperty(OUTPUT_LOCATION_RESULT, outputLocation); + } + + /** + * The location of the results of a query, null when the query has no result configuration, as with managed query + * results. + */ + private static String outputLocationOf(QueryExecution queryExecution) { + ResultConfiguration resultConfiguration = queryExecution.resultConfiguration(); + return resultConfiguration != null ? resultConfiguration.outputLocation() : null; } private String determineWorkGroup(final Exchange exchange) { diff --git a/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2ProducerResultHeadersTest.java b/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2ProducerResultHeadersTest.java new file mode 100644 index 000000000000..f9131544084f --- /dev/null +++ b/components/camel-aws/camel-aws2-athena/src/test/java/org/apache/camel/component/aws2/athena/Athena2ProducerResultHeadersTest.java @@ -0,0 +1,253 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.aws2.athena; + +import java.util.ArrayList; +import java.util.LinkedList; +import java.util.List; + +import org.apache.camel.BindToRegistry; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.services.athena.model.GetQueryResultsRequest; +import software.amazon.awssdk.services.athena.model.GetQueryResultsResponse; +import software.amazon.awssdk.services.athena.model.ListQueryExecutionsRequest; +import software.amazon.awssdk.services.athena.model.ListQueryExecutionsResponse; +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 software.amazon.awssdk.services.athena.model.StartQueryExecutionRequest; +import software.amazon.awssdk.services.athena.model.StartQueryExecutionResponse; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The output location and next token headers that an operation sets as its result are not taken as the input of a later + * operation in the same route. + */ +public class Athena2ProducerResultHeadersTest extends CamelTestSupport { + + private static final String QUERY_1 = "11111111-1111-1111-1111-111111111111"; + private static final String QUERY_2 = "22222222-2222-2222-2222-222222222222"; + private static final String ATHENA = "aws2-athena://label?outputLocation=s3://bucket/path/&queryString=RAW(SELECT 1)"; + + @BindToRegistry("amazonAthenaClient") + private final RecordingAthenaClient client = new RecordingAthenaClient(); + + @Test + public void secondStartQueryExecutionUsesOutputLocationOptionNotResultFileOfFirst() { + // the first query waits for its completion, and sets the output location header to its result file + Exchange exchange = template.send("direct:startTwice", e -> { + }); + + assertThat(exchange.getException()).isNull(); + assertThat(client.startRequests).hasSize(2); + assertThat(client.startRequests.get(0).resultConfiguration().outputLocation()).isEqualTo("s3://bucket/path/"); + assertThat(client.startRequests.get(1).resultConfiguration().outputLocation()).isEqualTo("s3://bucket/path/"); + } + + @Test + public void startQueryExecutionAfterGetQueryExecutionUsesOutputLocationOption() { + Exchange exchange = template.send("direct:getThenStart", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.startRequests).hasSize(1); + assertThat(client.startRequests.get(0).resultConfiguration().outputLocation()).isEqualTo("s3://bucket/path/"); + } + + @Test + public void startQueryExecutionAfterS3PointerUsesOutputLocationOption() { + Exchange exchange = template.send("direct:s3PointerThenStart", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.startRequests).hasSize(1); + assertThat(client.startRequests.get(0).resultConfiguration().outputLocation()).isEqualTo("s3://bucket/path/"); + } + + @Test + public void outputLocationHeaderSetByRouteAfterAnOperationIsUsed() { + Exchange exchange = template.send("direct:getThenSetHeaderThenStart", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.startRequests).hasSize(1); + assertThat(client.startRequests.get(0).resultConfiguration().outputLocation()).isEqualTo("s3://other/location/"); + } + + @Test + public void getQueryResultsDoesNotUseNextTokenOfListQueryExecutions() { + Exchange exchange = template.send("direct:listThenResults", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.resultsRequests).hasSize(1); + assertThat(client.resultsRequests.get(0).nextToken()).isNull(); + } + + @Test + public void listQueryExecutionsDoesNotUseNextTokenOfGetQueryResults() { + Exchange exchange = template.send("direct:resultsThenList", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.listRequests).hasSize(1); + assertThat(client.listRequests.get(0).nextToken()).isNull(); + } + + @Test + public void getQueryResultsDoesNotUseNextTokenOfAnotherQuery() { + Exchange exchange = template.send("direct:resultsOfTwoQueries", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.resultsRequests).hasSize(2); + assertThat(client.resultsRequests.get(1).queryExecutionId()).isEqualTo(QUERY_2); + assertThat(client.resultsRequests.get(1).nextToken()).isNull(); + } + + @Test + public void getQueryResultsPagesWithItsOwnNextToken() { + Exchange exchange = template.send("direct:resultsTwice", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + + assertThat(exchange.getException()).isNull(); + assertThat(client.resultsRequests).hasSize(2); + assertThat(client.resultsRequests.get(0).nextToken()).isNull(); + assertThat(client.resultsRequests.get(1).nextToken()).isEqualTo("results-token-1"); + } + + @Test + public void listQueryExecutionsPagesWithItsOwnNextToken() { + Exchange exchange = template.send("direct:listTwice", e -> { + }); + + assertThat(exchange.getException()).isNull(); + assertThat(client.listRequests).hasSize(2); + assertThat(client.listRequests.get(0).nextToken()).isNull(); + assertThat(client.listRequests.get(1).nextToken()).isEqualTo("list-token-1"); + } + + @Test + public void queryExecutionWithoutResultConfiguration() { + QueryExecution managed = QueryExecution.builder() + .queryExecutionId(QUERY_1) + .status(QueryExecutionStatus.builder().state(QueryExecutionState.SUCCEEDED).build()) + .build(); + client.setGetQueryExecutionResults(new LinkedList<>(List.of(managed, managed, managed))); + + Exchange get = template.send("aws2-athena://label?operation=getQueryExecution", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + assertThat(get.getException()).isNull(); + assertThat(get.getMessage().getHeader(Athena2Constants.OUTPUT_LOCATION)).isNull(); + + Exchange pointer = template.send("aws2-athena://label?operation=getQueryResults&outputType=S3Pointer", + e -> e.getIn().setHeader(Athena2Constants.QUERY_EXECUTION_ID, QUERY_1)); + assertThat(pointer.getException()).isNull(); + assertThat(pointer.getMessage().getHeader(Athena2Constants.OUTPUT_LOCATION)).isNull(); + assertThat(pointer.getMessage().getBody()).isNull(); + + Exchange start = template.send(ATHENA + "&waitTimeout=60000&initialDelay=1&delay=1", e -> { + }); + assertThat(start.getException()).isNull(); + assertThat(start.getMessage().getHeader(Athena2Constants.QUERY_EXECUTION_STATE)) + .isEqualTo(QueryExecutionState.SUCCEEDED); + assertThat(start.getMessage().getHeader(Athena2Constants.OUTPUT_LOCATION)).isNull(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:startTwice") + .to(ATHENA + "&waitTimeout=60000&initialDelay=1&delay=1") + .to(ATHENA); + + from("direct:getThenStart") + .to("aws2-athena://label?operation=getQueryExecution") + .to(ATHENA); + + from("direct:s3PointerThenStart") + .to("aws2-athena://label?operation=getQueryResults&outputType=S3Pointer") + .to(ATHENA); + + from("direct:getThenSetHeaderThenStart") + .to("aws2-athena://label?operation=getQueryExecution") + .setHeader(Athena2Constants.OUTPUT_LOCATION, constant("s3://other/location/")) + .to(ATHENA); + + from("direct:listThenResults") + .to("aws2-athena://label?operation=listQueryExecutions") + .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList"); + + from("direct:resultsThenList") + .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList") + .to("aws2-athena://label?operation=listQueryExecutions"); + + from("direct:resultsOfTwoQueries") + .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList") + .setHeader(Athena2Constants.QUERY_EXECUTION_ID, constant(QUERY_2)) + .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList"); + + from("direct:resultsTwice") + .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList") + .to("aws2-athena://label?operation=getQueryResults&outputType=SelectList"); + + from("direct:listTwice") + .to("aws2-athena://label?operation=listQueryExecutions") + .to("aws2-athena://label?operation=listQueryExecutions"); + } + }; + } + + /** + * Records the requests, and answers getQueryResults and listQueryExecutions with a next token. + */ + private static final class RecordingAthenaClient extends AmazonAthenaClientMock { + + private final List<StartQueryExecutionRequest> startRequests = new ArrayList<>(); + private final List<GetQueryResultsRequest> resultsRequests = new ArrayList<>(); + private final List<ListQueryExecutionsRequest> listRequests = new ArrayList<>(); + + @Override + public StartQueryExecutionResponse startQueryExecution(StartQueryExecutionRequest request) throws SdkException { + startRequests.add(request); + return super.startQueryExecution(request); + } + + @Override + public GetQueryResultsResponse getQueryResults(GetQueryResultsRequest request) throws SdkException { + resultsRequests.add(request); + return super.getQueryResults(request).toBuilder() + .nextToken("results-token-" + resultsRequests.size()) + .build(); + } + + @Override + public ListQueryExecutionsResponse listQueryExecutions(ListQueryExecutionsRequest request) throws SdkException { + listRequests.add(request); + return super.listQueryExecutions(request).toBuilder() + .nextToken("list-token-" + listRequests.size()) + .build(); + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 1bc188419d61..fec734f4465a 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -3392,6 +3392,19 @@ With `includeTrace=true`, the URI of the from endpoint in the SQL comment is now an endpoint URI, and it is left out when the exchange has no from endpoint, which failed with a `NullPointerException` before. +=== camel-aws2-athena - result headers are not taken as the input of a later operation + +* The `CamelAwsAthenaOutputLocation` header set by getQueryExecution, by getQueryResults with `S3Pointer`, and by + startQueryExecution with a `waitTimeout` is the file with the results of that query. A later startQueryExecution on + the same message no longer uses it, and puts its results under its `outputLocation` option. Previously it put them + under the result file of the earlier query. +* The `CamelAwsAthenaNextToken` header set by listQueryExecutions is now used only by a later listQueryExecutions, and + the one set by getQueryResults only by a later getQueryResults for the same query. Previously the token was sent to + the other operation, or for another query, which Athena rejects. +* A header that the route sets itself is used as before. +* A query execution without a result configuration, as with managed query results, no longer fails with a + `NullPointerException`: the `CamelAwsAthenaOutputLocation` header is null. + === camel-protobuf - a bytes field in a Map When a protobuf message is converted to a `Map` (for example `getBody(Map.class)` after unmarshalling), the value of
