This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 55d4f9e20a41 CAMEL-25702: camel-ibm-secrets-manager - skip an Event
Streams record that cannot be read instead of ending the refresh task (#27987)
55d4f9e20a41 is described below
commit 55d4f9e20a413f7e78130f0cc0e6439277217c79
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 14:43:17 2026 +0530
CAMEL-25702: camel-ibm-secrets-manager - skip an Event Streams record that
cannot be read instead of ending the refresh task (#27987)
IBMEventStreamReloadTriggerTask is scheduled once
(PeriodTaskScheduler.scheduledTask) and runs for as
long as Camel runs. Nothing in its loop caught an exception, so the first
record it could not read (a
value that is not JSON, or JSON without data.event_type, secrets,
secret_name or event_time) ended the
task. The exception went into a Future that nobody reads: nothing was
logged, and no later secret
rotation reloaded the CamelContext until the application was restarted.
The handling of one record moves unchanged into onRecord(), and run() calls
it in a try per record: a
record that cannot be processed is skipped with a WARN (stack trace at
DEBUG) and the task goes on.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../vault/IBMEventStreamReloadTriggerTask.java | 73 ++++++-----
.../vault/IBMEventStreamReloadTriggerTaskTest.java | 139 +++++++++++++++++++++
2 files changed, 178 insertions(+), 34 deletions(-)
diff --git
a/components/camel-ibm/camel-ibm-secrets-manager/src/main/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTask.java
b/components/camel-ibm/camel-ibm-secrets-manager/src/main/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTask.java
index 276e1df3f7ba..f5e3ed39ef98 100644
---
a/components/camel-ibm/camel-ibm-secrets-manager/src/main/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTask.java
+++
b/components/camel-ibm/camel-ibm-secrets-manager/src/main/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTask.java
@@ -173,56 +173,61 @@ public class IBMEventStreamReloadTriggerTask extends
ServiceSupport implements C
public void run() {
lastCheckTime = Instant.now();
- boolean triggerReloading = false;
ObjectMapper mapper = new ObjectMapper();
while (isRunAllowed()) {
ConsumerRecords<String, String> records =
kafkaConsumer.poll(Duration.ofMillis(pollTimeout));
for (ConsumerRecord<String, String> record : records) {
- JsonNode recordJson;
- String secretType;
try {
- recordJson = mapper.readTree(record.value());
- } catch (JsonProcessingException e) {
- throw new RuntimeException(e);
+ onRecord(mapper, record);
+ } catch (Exception e) {
+ // this task runs only once (until Camel stops), so a
record it cannot read must not end it: no later
+ // secret rotation would trigger a reload any more
+ LOG.warn("Skipping the IBM Event Streams record at offset
{} of {}, as it cannot be processed: {}",
+ record.offset(), record.topic(), e.getMessage());
+ LOG.debug("Cannot process the IBM Event Streams record at
offset {} of {}", record.offset(),
+ record.topic(), e);
}
- JsonNode payload = recordJson.get("data");
- if (payload != null) {
- secretType = payload.get("event_type").asText();
- if (secretType != null) {
- if
(secretType.equalsIgnoreCase(IBM_SECRETS_MANAGER_SECRET_ROTATED_EVENT)) {
- ArrayNode secretsRotated = (ArrayNode)
payload.get("secrets");
- for (JsonNode secret : secretsRotated) {
- if (secret != null) {
- String name =
secret.get("secret_name").asText();
- if (matchSecret(name)) {
- updates.put(name,
Instant.parse(secret.get("event_time").asText()));
- if (isReloadEnabled()) {
- LOG.info("Update for IBM secret:
{} detected, triggering CamelContext reload",
- name);
- triggerReloading = true;
- }
- if (triggerReloading) {
- ContextReloadStrategy reload
- =
camelContext.hasService(ContextReloadStrategy.class);
- if (reload != null) {
- // trigger reload
- lastReloadTime = Instant.now();
- reload.onReload(this);
- }
- triggerReloading = false;
- }
- break;
+ }
+ }
+
+ }
+
+ private void onRecord(ObjectMapper mapper, ConsumerRecord<String, String>
record) throws JsonProcessingException {
+ boolean triggerReloading = false;
+ JsonNode recordJson = mapper.readTree(record.value());
+ JsonNode payload = recordJson.get("data");
+ if (payload != null) {
+ String secretType = payload.get("event_type").asText();
+ if (secretType != null) {
+ if
(secretType.equalsIgnoreCase(IBM_SECRETS_MANAGER_SECRET_ROTATED_EVENT)) {
+ ArrayNode secretsRotated = (ArrayNode)
payload.get("secrets");
+ for (JsonNode secret : secretsRotated) {
+ if (secret != null) {
+ String name = secret.get("secret_name").asText();
+ if (matchSecret(name)) {
+ updates.put(name,
Instant.parse(secret.get("event_time").asText()));
+ if (isReloadEnabled()) {
+ LOG.info("Update for IBM secret: {}
detected, triggering CamelContext reload", name);
+ triggerReloading = true;
+ }
+ if (triggerReloading) {
+ ContextReloadStrategy reload =
camelContext.hasService(ContextReloadStrategy.class);
+ if (reload != null) {
+ // trigger reload
+ lastReloadTime = Instant.now();
+ reload.onReload(this);
}
+ triggerReloading = false;
}
+ break;
}
}
}
}
}
}
-
}
@Override
diff --git
a/components/camel-ibm/camel-ibm-secrets-manager/src/test/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTaskTest.java
b/components/camel-ibm/camel-ibm-secrets-manager/src/test/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTaskTest.java
new file mode 100644
index 000000000000..e3338be49344
--- /dev/null
+++
b/components/camel-ibm/camel-ibm-secrets-manager/src/test/java/org/apache/camel/component/ibm/secrets/manager/vault/IBMEventStreamReloadTriggerTaskTest.java
@@ -0,0 +1,139 @@
+/*
+ * 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.ibm.secrets.manager.vault;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultContextReloadStrategy;
+import org.apache.camel.vault.IBMSecretsManagerVaultConfiguration;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class IBMEventStreamReloadTriggerTaskTest {
+
+ private static final String TOPIC = "secrets-events";
+ private static final String ROTATED =
"{\"data\":{\"event_type\":\"secret_rotated\",\"secrets\":"
+ +
"[{\"secret_name\":\"database-password\",\"event_time\":\"2026-10-16T10:00:00Z\"}]}}";
+
+ @ParameterizedTest
+ @ValueSource(strings = {
+ "not json",
+ "{\"data\":{\"secrets\":[]}}",
+ "{\"data\":{\"event_type\":\"secret_rotated\"}}" })
+ void rotationAfterARecordThatCannotBeReadStillTriggersTheReload(String
unreadable) throws Exception {
+ CountDownLatch reloaded = new CountDownLatch(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ context.addService(new DefaultContextReloadStrategy() {
+ @Override
+ public void onReload(Object source) {
+ reloaded.countDown();
+ }
+ });
+ IBMSecretsManagerVaultConfiguration ibm =
context.getVaultConfiguration().ibmSecretsManager();
+ // the reload trigger task looks up the IBM properties function,
which needs these
+ ibm.setToken("api-key");
+ ibm.setServiceUrl("https://localhost:9443";);
+ ibm.setSecrets("database-password");
+ ibm.setEventStreamBootstrapServers("localhost:9092");
+ ibm.setEventStreamTopic(TOPIC);
+ ibm.setEventStreamGroupId("camel");
+ ibm.setEventStreamPassword("secret");
+ ibm.setEventStreamConsumerPollTimeout(100);
+ context.start();
+
+ IBMEventStreamReloadTriggerTask task = new
IBMEventStreamReloadTriggerTask();
+ task.setCamelContext(context);
+ task.start();
+ // replace the client that doStart created (it never connected) by
one that returns the given records
+ KafkaConsumer<String, String> created = task.kafkaConsumer;
+ ScriptedConsumer scripted = new ScriptedConsumer();
+ task.kafkaConsumer = scripted;
+ created.close(Duration.ZERO);
+
+ // an event that the task cannot read, then the rotation of a
secret that the task watches
+ scripted.batches.add(unreadable);
+ scripted.batches.add(ROTATED);
+
+ // the task runs once, on its own thread, for as long as Camel
runs (as the period task scheduler runs it)
+ Future<?> running = executor.submit(task);
+ boolean reloadTriggered = reloaded.await(10, TimeUnit.SECONDS);
+ boolean taskEnded = running.isDone();
+
+ task.stop();
+ assertTrue(reloadTriggered,
+ "the rotation of a watched secret did not trigger a reload
after the record [" + unreadable + "]");
+ assertFalse(taskEnded, "the reload trigger task ended while Camel
was running");
+ } finally {
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS), "the
reload trigger task did not end");
+ }
+ }
+
+ /**
+ * A Kafka consumer that is never connected: each poll returns the next
queued record value, or no record when none
+ * is queued within the poll timeout.
+ */
+ private static final class ScriptedConsumer extends KafkaConsumer<String,
String> {
+
+ private final BlockingQueue<String> batches = new
LinkedBlockingQueue<>();
+ private long offset;
+
+ ScriptedConsumer() {
+ super(Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9092",
+ ConsumerConfig.GROUP_ID_CONFIG, "camel",
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class));
+ }
+
+ @Override
+ public ConsumerRecords<String, String> poll(Duration timeout) {
+ String value;
+ try {
+ value = batches.poll(timeout.toMillis(),
TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return ConsumerRecords.empty();
+ }
+ if (value == null) {
+ return ConsumerRecords.empty();
+ }
+ TopicPartition partition = new TopicPartition(TOPIC, 0);
+ return new ConsumerRecords<>(
+ Map.of(partition, List.of(new ConsumerRecord<>(TOPIC, 0,
offset++, "key", value))), Map.of());
+ }
+ }
+}
