From 93fa9054bafaedbcd8bc1425384dcef113a82090 Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Tue, 4 Aug 2026 20:04:34 +0200 Subject: [PATCH] CAMEL-24341: camel-google-secret-manager - fix GCP vault refresh task defects (#25326) * CAMEL-24341: camel-google-secret-manager - fix GCP vault refresh task defects The GCP secret refresh task read the AWS vault configuration, so camel.vault.gcp.secrets was never honoured. It also kept triggerReloading as receiver state, so every message following a matching secret event triggered another CamelContext reload, restarted the subscriber on every period (a Google ApiService can only be started while it is NEW), and dereferenced the secretId/eventType attributes without a null check, so a message published on the subscription by anything else failed and was redelivered forever. Also aligns the javadoc of the task and of the properties function with GCP and reuses a single ObjectMapper for secret sub-key lookups. Signed-off-by: Andrea Cosentino * CAMEL-24341: address review - AssertJ and package-private test conventions Per davsclaus's non-blocking note, PubsubReloadTriggerTaskTest now uses package-private class/@Test visibility and AssertJ assertions (assertThat, assertThatThrownBy, assertThatCode(...).doesNotThrowAnyException()). The interface-override methods (onReload/ack/nack) stay public since they override public API. Adds a test-scoped assertj-core dependency, not previously on the module's test classpath. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Andrea Cosentino --------- Signed-off-by: Andrea Cosentino Co-authored-by: Claude Opus 4.8 (1M context) --- .../camel-google-secret-manager/pom.xml | 5 + ...GoogleSecretManagerPropertiesFunction.java | 9 +- .../vault/PubsubReloadTriggerTask.java | 25 ++- .../vault/PubsubReloadTriggerTaskTest.java | 156 ++++++++++++++++++ 4 files changed, 183 insertions(+), 12 deletions(-) create mode 100644 components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java diff --git a/components/camel-google/camel-google-secret-manager/pom.xml b/components/camel-google/camel-google-secret-manager/pom.xml index 2913e0fde7e74..0edee4de651f2 100644 --- a/components/camel-google/camel-google-secret-manager/pom.xml +++ b/components/camel-google/camel-google-secret-manager/pom.xml @@ -83,5 +83,10 @@ camel-test-junit5 test + + org.assertj + assertj-core + test + diff --git a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java index 82dc1e3c29060..dc2539badc3bc 100644 --- a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java +++ b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java @@ -55,8 +55,8 @@ * Otherwise it is possible to specify the credentials as properties: * *
    - *
  • camel.vault.aws.serviceAccountKey
    • - *
  • camel.vault.aws.projectId
    • + *
  • camel.vault.gcp.serviceAccountKey
    • + *
  • camel.vault.gcp.projectId
    • *
*

* @@ -79,6 +79,8 @@ public class GoogleSecretManagerPropertiesFunction extends ServiceSupport implem private static final String CAMEL_VAULT_GCP_PROJECT_ID = "CAMEL_VAULT_GCP_PROJECT_ID"; private static final String CAMEL_VAULT_GCP_USE_DEFAULT_INSTANCE = "CAMEL_VAULT_GCP_USE_DEFAULT_INSTANCE"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + boolean useDefaultInstance; private CamelContext camelContext; @@ -207,8 +209,7 @@ private String getSecretFromSource( returnValue = response.getPayload().getData().toStringUtf8(); } if (ObjectHelper.isNotEmpty(subkey) && ObjectHelper.isNotEmpty(returnValue)) { - ObjectMapper mapper = new ObjectMapper(); - JsonNode actualObj = mapper.readTree(returnValue); + JsonNode actualObj = MAPPER.readTree(returnValue); JsonNode field = actualObj.get(subkey); if (ObjectHelper.isNotEmpty(field)) { returnValue = field.textValue(); diff --git a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java index a0ca3bcb01ce7..6edf740236fbd 100644 --- a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java +++ b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Set; +import com.google.api.core.ApiService; import com.google.api.gax.core.FixedCredentialsProvider; import com.google.auth.Credentials; import com.google.auth.oauth2.GoogleCredentials; @@ -89,7 +90,7 @@ public boolean isReloadEnabled() { } /** - * Whether Camel should be reloaded on AWS secret updated + * Whether Camel should be reloaded on GCP secret updated */ public void setReloadEnabled(boolean reloadEnabled) { this.reloadEnabled = reloadEnabled; @@ -120,6 +121,9 @@ public Instant getLastReloadTime() { protected void doStart() throws Exception { super.doStart(); + // specific secrets + secrets = camelContext.getVaultConfiguration().gcp().getSecrets(); + // auto-detect secrets in-use PropertiesComponent pc = camelContext.getPropertiesComponent(); PropertiesFunction pf = pc.getPropertiesFunction("gcp"); @@ -127,8 +131,7 @@ protected void doStart() throws Exception { propertiesFunction = (GoogleSecretManagerPropertiesFunction) pf; LOG.debug("Auto-detecting secrets from properties-function: {}", pf.getName()); } - // specific secrets - secrets = camelContext.getVaultConfiguration().aws().getSecrets(); + if (ObjectHelper.isEmpty(secrets) && propertiesFunction == null) { throw new IllegalArgumentException("Secrets must be configured on GCP vault configuration"); } @@ -187,7 +190,14 @@ protected void doShutdown() throws Exception { public void run() { lastCheckTime = Instant.now(); - subscriber.startAsync().awaitRunning(); + if (subscriber == null) { + return; + } + // the subscriber is push based, so it only has to be started once. A Google ApiService can only be + // started while it is still NEW, starting it again on every period would fail with an IllegalStateException + if (subscriber.state() == ApiService.State.NEW) { + subscriber.startAsync().awaitRunning(); + } } protected boolean matchSecret(String name) { @@ -220,14 +230,13 @@ public class FilteringEventMessageReceiver implements MessageReceiver { private static final String SECRET_UPDATE = "SECRET_UPDATE"; private static final String SECRET_VERSION_ADD = "SECRET_VERSION_ADD"; - private boolean triggerReloading; - @Override public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) { + boolean triggerReloading = false; String secretId = message.getAttributesMap().get("secretId"); String eventType = message.getAttributesMap().get("eventType"); - if (eventType.equalsIgnoreCase(SECRET_UPDATE) || eventType.equalsIgnoreCase(SECRET_VERSION_ADD)) { - if (matchSecret(secretId)) { + if (SECRET_UPDATE.equalsIgnoreCase(eventType) || SECRET_VERSION_ADD.equalsIgnoreCase(eventType)) { + if (secretId != null && matchSecret(secretId)) { int secretNameBeginInd = secretId.lastIndexOf("/") + 1; updates.put(secretId.substring(secretNameBeginInd), Instant.ofEpochSecond(message.getPublishTime().getSeconds(), message.getPublishTime().getNanos())); diff --git a/components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java b/components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java new file mode 100644 index 0000000000000..56c244fed5567 --- /dev/null +++ b/components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java @@ -0,0 +1,156 @@ +/* + * 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.google.secret.manager.vault; + +import java.lang.reflect.Field; +import java.util.concurrent.atomic.AtomicInteger; + +import com.google.cloud.pubsub.v1.AckReplyConsumer; +import com.google.pubsub.v1.PubsubMessage; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.DefaultContextReloadStrategy; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class PubsubReloadTriggerTaskTest { + + private static final class CountingReloadStrategy extends DefaultContextReloadStrategy { + private final AtomicInteger reloads = new AtomicInteger(); + + @Override + public void onReload(Object source) { + reloads.incrementAndGet(); + } + } + + private static final class RecordingAck implements AckReplyConsumer { + private boolean acked; + + @Override + public void ack() { + acked = true; + } + + @Override + public void nack() { + } + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field f = PubsubReloadTriggerTask.class.getDeclaredField(name); + f.setAccessible(true); + f.set(target, value); + } + + private static Object getField(Object target, String name) throws Exception { + Field f = PubsubReloadTriggerTask.class.getDeclaredField(name); + f.setAccessible(true); + return f.get(target); + } + + private static PubsubMessage message(String eventType, String secretId) { + PubsubMessage.Builder builder = PubsubMessage.newBuilder(); + if (eventType != null) { + builder.putAttributes("eventType", eventType); + } + if (secretId != null) { + builder.putAttributes("secretId", secretId); + } + return builder.build(); + } + + @Test + void secretsAreReadFromTheGcpVaultConfiguration() throws Exception { + DefaultCamelContext context = new DefaultCamelContext(); + context.getVaultConfiguration().gcp().setSecrets("gcp-secret"); + context.getVaultConfiguration().aws().setSecrets("aws-secret"); + + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(context); + + // no GCP credentials are configured, so starting the task fails once it reaches the gcp properties + // function. By then the secrets to watch must already have been read from the GCP vault configuration + assertThatThrownBy(task::start).isInstanceOf(RuntimeCamelException.class); + assertThat(getField(task, "secrets")).isEqualTo("gcp-secret"); + } + + @Test + void reloadIsTriggeredOnlyForMatchingSecrets() throws Exception { + DefaultCamelContext context = new DefaultCamelContext(); + CountingReloadStrategy reload = new CountingReloadStrategy(); + context.addService(reload); + context.start(); + + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(context); + setField(task, "secrets", "tracked"); + + PubsubReloadTriggerTask.FilteringEventMessageReceiver receiver = task.new FilteringEventMessageReceiver(); + + RecordingAck matching = new RecordingAck(); + receiver.receiveMessage(message("SECRET_UPDATE", "projects/p/secrets/tracked"), matching); + assertThat(reload.reloads.get()).isEqualTo(1); + assertThat(matching.acked).isTrue(); + assertThat(task.getUpdates()).containsKey("tracked"); + + // a message for a secret that is not watched must not trigger another reload + RecordingAck other = new RecordingAck(); + receiver.receiveMessage(message("SECRET_UPDATE", "projects/p/secrets/other"), other); + assertThat(reload.reloads.get()).isEqualTo(1); + assertThat(other.acked).isTrue(); + + // neither must an event type that is not a secret update + RecordingAck unrelated = new RecordingAck(); + receiver.receiveMessage(message("SECRET_DELETE", "projects/p/secrets/tracked"), unrelated); + assertThat(reload.reloads.get()).isEqualTo(1); + assertThat(unrelated.acked).isTrue(); + + context.stop(); + } + + @Test + void messagesWithoutAttributesAreAcknowledged() throws Exception { + DefaultCamelContext context = new DefaultCamelContext(); + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(context); + setField(task, "secrets", "tracked"); + + PubsubReloadTriggerTask.FilteringEventMessageReceiver receiver = task.new FilteringEventMessageReceiver(); + + // a message published on the subscription by something other than the secret manager event feed + RecordingAck noAttributes = new RecordingAck(); + assertThatCode(() -> receiver.receiveMessage(message(null, null), noAttributes)).doesNotThrowAnyException(); + assertThat(noAttributes.acked).isTrue(); + + RecordingAck noSecretId = new RecordingAck(); + assertThatCode(() -> receiver.receiveMessage(message("SECRET_UPDATE", null), noSecretId)).doesNotThrowAnyException(); + assertThat(noSecretId.acked).isTrue(); + } + + @Test + void runWithoutSubscriberDoesNotThrow() { + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(new DefaultCamelContext()); + + assertThatCode(task::run).doesNotThrowAnyException(); + assertThat(task.getLastCheckTime()).isNotNull(); + } +}