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 b6ea388b59dc CAMEL-24193: camel-aws-secrets-manager - fix
SQS-notification secret refresh NPE and queue draining (#24847)
b6ea388b59dc is described below
commit b6ea388b59dc0a0259058bd84f354ad19fbbcc6b
Author: Andrea Cosentino <[email protected]>
AuthorDate: Fri Jul 17 15:04:35 2026 +0200
CAMEL-24193: camel-aws-secrets-manager - fix SQS-notification secret
refresh NPE and queue draining (#24847)
In the SQS-notification branch of CloudTrailReloadTriggerTask:
- A QueueDoesNotExistException from receiveMessage was caught and only
logged,
leaving the result null, which was then dereferenced by the message loop
(NPE on every poll). The loop is now guarded against a null receive
result.
- deleteMessage ran only for the first matching message when reload was
enabled, so non-matching messages (and everything when reload was
disabled)
were never deleted and were redelivered on every poll. Every examined
message
is now deleted so the queue is drained; a malformed message body is logged
and discarded instead of aborting the whole batch.
- The per-message ObjectMapper allocation is replaced with a single reused
instance.
Adds CloudTrailReloadTriggerTaskSqsTest covering the drain behaviour and the
missing-queue path.
Co-Authored-By: Claude Fable 5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel-aws/camel-aws-secrets-manager/pom.xml | 6 ++
.../vault/CloudTrailReloadTriggerTask.java | 41 +++++----
.../vault/CloudTrailReloadTriggerTaskSqsTest.java | 98 ++++++++++++++++++++++
3 files changed, 123 insertions(+), 22 deletions(-)
diff --git a/components/camel-aws/camel-aws-secrets-manager/pom.xml
b/components/camel-aws/camel-aws-secrets-manager/pom.xml
index 96ddacfba160..43aaed0b9341 100644
--- a/components/camel-aws/camel-aws-secrets-manager/pom.xml
+++ b/components/camel-aws/camel-aws-secrets-manager/pom.xml
@@ -90,6 +90,12 @@
<artifactId>camel-test-spring-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
diff --git
a/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
b/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
index 5c194115f0ea..0795267005bc 100644
---
a/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
+++
b/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
@@ -92,6 +92,7 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
private static final String CAMEL_AWS_VAULT_URI_ENDPOINT_OVERRIDE =
"CAMEL_AWS_VAULT_URI_ENDPOINT_OVERRIDE";
private static final Logger LOG =
LoggerFactory.getLogger(CloudTrailReloadTriggerTask.class);
+ private static final ObjectMapper MAPPER = new ObjectMapper();
private static final String SECRETSMANAGER_AMAZONAWS_COM =
"secretsmanager.amazonaws.com";
private static final String SECRETSMANAGER_UPDATE_EVENT = "PutSecretValue";
@@ -329,38 +330,34 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
LOG.info("Queue does not exist.");
}
- for (Message message : messageResult.messages()) {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode event = null;
- try {
- event = mapper.readTree(message.body());
- } catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- if (ObjectHelper.isNotEmpty(event.get("detail"))) {
- JsonNode innerDetail = event.get("detail");
- if
(innerDetail.get("eventSource").asText().equalsIgnoreCase(SECRETSMANAGER_AMAZONAWS_COM))
{
- if
(innerDetail.get("eventName").asText().equalsIgnoreCase(SECRETSMANAGER_UPDATE_EVENT)
- || innerDetail.get("eventName").asText()
-
.equalsIgnoreCase(SECRETSMANAGER_UPDATE_SECRET_EVENT)) {
+ if (messageResult != null) {
+ for (Message message : messageResult.messages()) {
+ try {
+ JsonNode event = MAPPER.readTree(message.body());
+ JsonNode innerDetail = event.get("detail");
+ if (ObjectHelper.isNotEmpty(innerDetail)
+ &&
innerDetail.get("eventSource").asText().equalsIgnoreCase(SECRETSMANAGER_AMAZONAWS_COM)
+ &&
(innerDetail.get("eventName").asText().equalsIgnoreCase(SECRETSMANAGER_UPDATE_EVENT)
+ ||
innerDetail.get("eventName").asText()
+
.equalsIgnoreCase(SECRETSMANAGER_UPDATE_SECRET_EVENT))) {
String name =
innerDetail.get("requestParameters").get("secretId").asText();
if (matchSecret(name)) {
updates.put(name,
Instant.parse(innerDetail.get("eventTime").asText()));
if (isReloadEnabled()) {
LOG.info("Update for AWS secret: {}
detected, triggering CamelContext reload", name);
triggerReloading = true;
- DeleteMessageRequest.Builder
deleteRequest
- =
DeleteMessageRequest.builder().queueUrl(queueUrl)
-
.receiptHandle(message.receiptHandle());
-
- LOG.trace("Deleting message with
receipt handle {}...", message.receiptHandle());
-
-
sqsClient.deleteMessage(deleteRequest.build());
}
- break;
}
}
+ } catch (JsonProcessingException e) {
+ LOG.warn("Unable to parse SQS message body,
discarding message: {}", e.getMessage());
}
+ // Always delete the examined message so the queue is
drained rather than redelivering it
+ // on every poll (matched -> reload recorded above;
unmatched or malformed -> discarded).
+ LOG.trace("Deleting message with receipt handle
{}...", message.receiptHandle());
+ sqsClient.deleteMessage(
+
DeleteMessageRequest.builder().queueUrl(queueUrl)
+
.receiptHandle(message.receiptHandle()).build());
}
}
} catch (Exception e) {
diff --git
a/components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskSqsTest.java
b/components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskSqsTest.java
new file mode 100644
index 000000000000..3ae4f097ddfc
--- /dev/null
+++
b/components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskSqsTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.aws.secretsmanager.vault;
+
+import java.lang.reflect.Field;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.QueueDoesNotExistException;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class CloudTrailReloadTriggerTaskSqsTest {
+
+ @Mock
+ private SqsClient sqsClient;
+
+ private static String event(String secretId) {
+ return
"{\"detail\":{\"eventSource\":\"secretsmanager.amazonaws.com\",\"eventName\":\"PutSecretValue\","
+ + "\"requestParameters\":{\"secretId\":\"" + secretId +
"\"},\"eventTime\":\"2026-01-01T00:00:00Z\"}}";
+ }
+
+ private CloudTrailReloadTriggerTask task(String secrets) throws Exception {
+ CloudTrailReloadTriggerTask task = new CloudTrailReloadTriggerTask();
+ task.setCamelContext(new DefaultCamelContext());
+ setField(task, "useSqsNotification", true);
+ setField(task, "queueUrl", "http://localhost/queue";);
+ setField(task, "sqsClient", sqsClient);
+ setField(task, "secrets", secrets);
+ return task;
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Field f = CloudTrailReloadTriggerTask.class.getDeclaredField(name);
+ f.setAccessible(true);
+ f.set(target, value);
+ }
+
+ private static Message message(String secretId, String receiptHandle) {
+ return
Message.builder().body(event(secretId)).receiptHandle(receiptHandle).build();
+ }
+
+ @Test
+ public void drainsEveryExaminedMessageIncludingNonMatching() throws
Exception {
+ CloudTrailReloadTriggerTask task = task("tracked-secret");
+ // One message matches the tracked secret, one does not. Both must be
deleted so the queue is drained.
+ when(sqsClient.receiveMessage(any(ReceiveMessageRequest.class)))
+ .thenReturn(ReceiveMessageResponse.builder()
+ .messages(message("other-secret", "rh1"),
message("tracked-secret", "rh2"))
+ .build());
+
+ task.run();
+
+ verify(sqsClient,
times(2)).deleteMessage(any(DeleteMessageRequest.class));
+ }
+
+ @Test
+ public void doesNotThrowOrDeleteWhenQueueDoesNotExist() throws Exception {
+ CloudTrailReloadTriggerTask task = task("tracked-secret");
+ when(sqsClient.receiveMessage(any(ReceiveMessageRequest.class)))
+
.thenThrow(QueueDoesNotExistException.builder().message("missing").build());
+
+ // A missing queue must not lead to a NullPointerException on the
(null) receive result.
+ assertDoesNotThrow(task::run);
+ verify(sqsClient,
never()).deleteMessage(any(DeleteMessageRequest.class));
+ }
+}
