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 daf9c53c9130 CAMEL-25522: camel-ibm-cos - Keep the in-progress
repository of the consumer accurate with fileName and on stop (#27673)
daf9c53c9130 is described below
commit daf9c53c9130cd6b7618c74135918f9c34faa4cc
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 13:25:13 2026 +0530
CAMEL-25522: camel-ibm-cos - Keep the in-progress repository of the
consumer accurate with fileName and on stop (#27673)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/ibm/cos/IBMCOSConsumer.java | 37 +++-
.../camel/component/ibm/cos/FakeS3Server.java | 206 +++++++++++++++++
.../ibm/cos/IBMCOSConsumerInProgressTest.java | 244 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 7 +
4 files changed, 490 insertions(+), 4 deletions(-)
diff --git
a/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
b/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
index 66de7917f054..4ccd1e341c9d 100644
---
a/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
+++
b/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.ibm.cos;
+import java.io.Closeable;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
@@ -66,10 +67,22 @@ public class IBMCOSConsumer extends
ScheduledBatchPollingConsumer {
Queue<Exchange> exchanges;
if (fileName != null) {
- LOG.trace("Getting object in bucket [{}] with file name [{}]...",
bucketName, fileName);
-
- S3Object s3Object = getCosClient().getObject(new
GetObjectRequest(bucketName, fileName));
- exchanges = createExchanges(s3Object, fileName);
+ if (getEndpoint().getInProgressRepository() != null
+ && !getEndpoint().getInProgressRepository().add(fileName))
{
+ LOG.trace("Object {} is already in progress", fileName);
+ exchanges = new LinkedList<>();
+ } else {
+ try {
+ LOG.trace("Getting object in bucket [{}] with file name
[{}]...", bucketName, fileName);
+
+ S3Object s3Object = getCosClient().getObject(new
GetObjectRequest(bucketName, fileName));
+ exchanges = createExchanges(s3Object, fileName);
+ } catch (Exception e) {
+ // the object is not consumed by this poll
+ removeInProgress(fileName);
+ throw e;
+ }
+ }
} else {
LOG.trace("Queueing objects in bucket [{}]...", bucketName);
@@ -136,6 +149,22 @@ public class IBMCOSConsumer extends
ScheduledBatchPollingConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+ // the remaining exchanges are not processed as the consumer is
stopping: a later poll must consume their
+ // objects again, and their on completions, which close the content
stream, never run
+ for (Object notRouted : exchanges) {
+ Exchange exchange = (Exchange) notRouted;
+ // the key the consumer stored when it created the exchange
+ String key = exchange.getProperty(IBMCOSConstants.KEY,
String.class);
+ if (key != null) {
+ removeInProgress(key);
+ }
+ if (exchange.getIn().getBody() instanceof Closeable body) {
+ IOHelper.close(body);
+ }
+ // nothing else releases an exchange that is never routed to the
exchange factory (which resets a pooled
+ // exchange, so this comes last)
+ releaseExchange(exchange, true);
+ }
return total;
}
diff --git
a/components/camel-ibm/camel-ibm-cos/src/test/java/org/apache/camel/component/ibm/cos/FakeS3Server.java
b/components/camel-ibm/camel-ibm-cos/src/test/java/org/apache/camel/component/ibm/cos/FakeS3Server.java
new file mode 100644
index 000000000000..6323ccfe701e
--- /dev/null
+++
b/components/camel-ibm/camel-ibm-cos/src/test/java/org/apache/camel/component/ibm/cos/FakeS3Server.java
@@ -0,0 +1,206 @@
+/*
+ * 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.cos;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * A minimal in-process fake of the S3 API calls the consumer makes for a
single bucket (path-style requests): head
+ * bucket, list objects (v2, paginated with max-keys and
start-after/continuation-token), head, get and delete object.
+ */
+final class FakeS3Server {
+
+ private final String bucket;
+ private final HttpServer server;
+ private final NavigableMap<String, byte[]> objects = new
ConcurrentSkipListMap<>();
+ private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
+
+ FakeS3Server(String bucket) throws IOException {
+ this.bucket = bucket;
+ this.server = HttpServer.create(new InetSocketAddress("localhost", 0),
0);
+ this.server.createContext("/", this::handle);
+ }
+
+ void start() {
+ server.start();
+ }
+
+ void stop() {
+ server.stop(0);
+ }
+
+ String endpoint() {
+ return "http://localhost:"; + server.getAddress().getPort();
+ }
+
+ void put(String key, String content) {
+ objects.put(key, content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ int gets(String key) {
+ AtomicInteger n = gets.get(key);
+ return n != null ? n.get() : 0;
+ }
+
+ private void handle(HttpExchange http) throws IOException {
+ String path = http.getRequestURI().getRawPath();
+ String key = path.length() > bucket.length() + 2 ?
decode(path.substring(bucket.length() + 2)) : null;
+ String method = http.getRequestMethod();
+ if (key == null) {
+ if ("GET".equals(method) && query(http).containsKey("list-type")) {
+ list(http, query(http));
+ } else {
+ // head bucket, location, ...
+ respond(http, 200, null, null);
+ }
+ return;
+ }
+ byte[] data = objects.get(key);
+ switch (method) {
+ case "HEAD" -> {
+ if (data == null) {
+ respond(http, 404, null, null);
+ } else {
+ objectHeaders(http, data);
+ http.sendResponseHeaders(200, -1);
+ http.close();
+ }
+ }
+ case "GET" -> {
+ if (data == null) {
+ noSuchKey(http, key);
+ } else {
+ gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ objectHeaders(http, data);
+ http.sendResponseHeaders(200, data.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(data);
+ }
+ }
+ }
+ case "DELETE" -> {
+ objects.remove(key);
+ http.sendResponseHeaders(204, -1);
+ http.close();
+ }
+ default -> respond(http, 405, null, null);
+ }
+ }
+
+ private void list(HttpExchange http, Map<String, String> query) throws
IOException {
+ int maxKeys = query.containsKey("max-keys") ?
Integer.parseInt(query.get("max-keys")) : 1000;
+ String after = query.get("continuation-token");
+ if (after == null) {
+ after = query.get("start-after");
+ }
+ NavigableMap<String, byte[]> view = after != null ?
objects.tailMap(after, false) : objects;
+ StringBuilder contents = new StringBuilder();
+ int count = 0;
+ String last = null;
+ for (Map.Entry<String, byte[]> e : view.entrySet()) {
+ if (count == maxKeys) {
+ break;
+ }
+
contents.append("<Contents><Key>").append(e.getKey()).append("</Key>")
+
.append("<LastModified>2026-10-09T00:00:00.000Z</LastModified>")
+
.append("<ETag>\"").append(md5(e.getValue())).append("\"</ETag>")
+
.append("<Size>").append(e.getValue().length).append("</Size>")
+
.append("<StorageClass>STANDARD</StorageClass></Contents>");
+ last = e.getKey();
+ count++;
+ }
+ boolean truncated = last != null && objects.higherKey(last) != null;
+ String xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+ + "<ListBucketResult
xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\";>"
+ + "<Name>" + bucket + "</Name><Prefix></Prefix>"
+ + "<KeyCount>" + count + "</KeyCount><MaxKeys>" + maxKeys
+ "</MaxKeys>"
+ + "<IsTruncated>" + truncated + "</IsTruncated>"
+ + (truncated ? "<NextContinuationToken>" + last +
"</NextContinuationToken>" : "")
+ + contents + "</ListBucketResult>";
+ respond(http, 200, "application/xml", xml);
+ }
+
+ private static void objectHeaders(HttpExchange http, byte[] data) {
+ // the client checks the content against the MD5 in the ETag
+ http.getResponseHeaders().add("ETag", "\"" + md5(data) + "\"");
+ http.getResponseHeaders().add("Last-Modified", "Fri, 09 Oct 2026
00:00:00 GMT");
+ http.getResponseHeaders().add("Content-Type", "text/plain");
+ http.getResponseHeaders().add("Content-Length",
String.valueOf(data.length));
+ }
+
+ private void noSuchKey(HttpExchange http, String key) throws IOException {
+ String xml = "<?xml version=\"1.0\"
encoding=\"UTF-8\"?><Error><Code>NoSuchKey</Code>"
+ + "<Message>The specified key does not
exist.</Message><Key>" + key + "</Key>"
+ + "<BucketName>" + bucket + "</BucketName><Resource>/" +
bucket + "/" + key + "</Resource>"
+ + "<RequestId>1</RequestId><HostId>1</HostId></Error>";
+ respond(http, 404, "application/xml", xml);
+ }
+
+ private static void respond(HttpExchange http, int status, String
contentType, String body) throws IOException {
+ if (body == null) {
+ http.sendResponseHeaders(status, -1);
+ http.close();
+ return;
+ }
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ http.getResponseHeaders().add("Content-Type", contentType);
+ http.sendResponseHeaders(status, bytes.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(bytes);
+ }
+ }
+
+ private static Map<String, String> query(HttpExchange http) {
+ Map<String, String> answer = new HashMap<>();
+ String q = http.getRequestURI().getRawQuery();
+ if (q != null) {
+ for (String part : q.split("&")) {
+ int eq = part.indexOf('=');
+ answer.put(decode(eq >= 0 ? part.substring(0, eq) : part), eq
>= 0 ? decode(part.substring(eq + 1)) : "");
+ }
+ }
+ return answer;
+ }
+
+ private static String md5(byte[] data) {
+ try {
+ return
HexFormat.of().formatHex(MessageDigest.getInstance("MD5").digest(data));
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static String decode(String s) {
+ return URLDecoder.decode(s, StandardCharsets.UTF_8);
+ }
+}
diff --git
a/components/camel-ibm/camel-ibm-cos/src/test/java/org/apache/camel/component/ibm/cos/IBMCOSConsumerInProgressTest.java
b/components/camel-ibm/camel-ibm-cos/src/test/java/org/apache/camel/component/ibm/cos/IBMCOSConsumerInProgressTest.java
new file mode 100644
index 000000000000..8363303c2a91
--- /dev/null
+++
b/components/camel-ibm/camel-ibm-cos/src/test/java/org/apache/camel/component/ibm/cos/IBMCOSConsumerInProgressTest.java
@@ -0,0 +1,244 @@
+/*
+ * 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.cos;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.ibm.cloud.objectstorage.auth.AWSStaticCredentialsProvider;
+import com.ibm.cloud.objectstorage.auth.BasicAWSCredentials;
+import com.ibm.cloud.objectstorage.client.builder.AwsClientBuilder;
+import com.ibm.cloud.objectstorage.services.s3.AmazonS3;
+import com.ibm.cloud.objectstorage.services.s3.AmazonS3ClientBuilder;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The consumer keeps the objects whose exchanges are being processed in its
in-progress repository: with fileName, an
+ * object still in flight is not consumed again by the next poll, a failed get
does not leave the object in progress,
+ * and the objects of a batch that a stopping consumer does not process are
consumed when the route is started again,
+ * and their content streams are closed.
+ */
+class IBMCOSConsumerInProgressTest extends CamelTestSupport {
+
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch failedPoll = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private FakeS3Server s3;
+ private AmazonS3 cosClient;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ s3.start();
+ cosClient = AmazonS3ClientBuilder.standard()
+ .withCredentials(new AWSStaticCredentialsProvider(new
BasicAWSCredentials("xxx", "yyy")))
+ .withEndpointConfiguration(new
AwsClientBuilder.EndpointConfiguration(s3.endpoint(), "us-south"))
+ .withPathStyleAccessEnabled(true)
+ .build();
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("cos", cosClient);
+ // counts the polls that completed (the exchanges of the poll were
handed to the route)
+ context.getRegistry().bind("countPolls", new
PollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ return true;
+ }
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ twoPolls.countDown();
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPoll.countDown();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void stopServer() {
+ release.countDown();
+ if (cosClient != null) {
+ cosClient.shutdown();
+ }
+ if (s3 != null) {
+ s3.stop();
+ }
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgainWithFileName() throws Exception {
+ s3.put("a.txt", "hello");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("ibm-cos://bkt?cosClient=#cos&delay=10&fileName=a.txt&pollStrategy=#countPolls")
+ .to("seda:hold");
+
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:hold")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ assertTrue(inRoute.await(20, TimeUnit.SECONDS), "the object was not
consumed");
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertTrue(twoPolls.await(20, TimeUnit.SECONDS), "the consumer did not
poll again");
+ int gets = s3.gets("a.txt");
+ release.countDown();
+
+ assertEquals(1, gets, "a.txt was consumed again by a later poll while
its first exchange was still in flight");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void objectWithFileNameIsConsumedOnceItExists() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("ibm-cos://bkt?cosClient=#cos&delay=10&fileName=a.txt&pollStrategy=#countPolls")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ // the get fails (NoSuchKey) while the object does not exist, which
must not leave it in progress
+ assertTrue(failedPoll.await(20, TimeUnit.SECONDS), "the get of the
missing object did not fail");
+ s3.put("a.txt", "hello");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void objectsOfAStoppedBatchAreConsumedAfterRestart() throws Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.put("c.txt", "c");
+ AtomicBoolean first = new AtomicBoolean(true);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("ibm-cos://bkt?cosClient=#cos&delay=10").routeId("cos")
+ .process(e -> {
+ if (first.compareAndSet(true, false)) {
+ // stop the route while the first exchange of
the batch is processed, and wait until
+ // the consumer does not process the rest of
the batch
+ IBMCOSConsumer consumer = (IBMCOSConsumer)
context.getRoute("cos").getConsumer();
+ CompletableFuture.runAsync(() -> {
+ try {
+
context.getRouteController().stopRoute("cos");
+ } catch (Exception ex) {
+ throw new RuntimeCamelException(ex);
+ }
+ });
+ await().atMost(20, TimeUnit.SECONDS).until(()
-> !consumer.isBatchAllowed());
+ }
+ })
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ mock.expectedBodiesReceived("a");
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() ->
context.getRouteController().getRouteStatus("cos") == ServiceStatus.Stopped);
+
+ mock.reset();
+ mock.expectedBodiesReceivedInAnyOrder("b", "c");
+ context.getRouteController().startRoute("cos");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void notProcessedExchangesCloseTheirContent() throws Exception {
+ IBMCOSEndpoint endpoint =
context.getEndpoint("ibm-cos://bkt?cosClient=#cos", IBMCOSEndpoint.class);
+ List<Boolean> closedWhenReleased = new ArrayList<>();
+ AtomicBoolean closed = new AtomicBoolean();
+ // never started, so the consumer does not process any exchange of the
batch
+ IBMCOSConsumer consumer = new IBMCOSConsumer(endpoint, exchange -> {
+ }) {
+ @Override
+ public void releaseExchange(Exchange exchange, boolean
autoRelease) {
+ closedWhenReleased.add(closed.get());
+ super.releaseExchange(exchange, autoRelease);
+ }
+ };
+ endpoint.getInProgressRepository().add("b.txt");
+ Exchange exchange = endpoint.createExchange();
+ // the consumer stores the key as exchange property, which is what it
releases, whatever the header says
+ exchange.setProperty(IBMCOSConstants.KEY, "b.txt");
+ exchange.getIn().setHeader(IBMCOSConstants.KEY, "other.txt");
+ exchange.getIn().setBody(new
ByteArrayInputStream("b".getBytes(StandardCharsets.UTF_8)) {
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ });
+ Queue<Object> batch = new LinkedList<>();
+ batch.add(exchange);
+
+ consumer.processBatch(batch);
+
+ assertFalse(endpoint.getInProgressRepository().contains("b.txt"),
"b.txt is still in progress");
+ assertTrue(closed.get(), "the content stream of b.txt was not closed");
+ // released to the exchange factory, after the stream was closed as a
pooled exchange is reset when released
+ assertEquals(List.of(true), closedWhenReleased, "the exchange of b.txt
was not released after closing its stream");
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index ae3640b4cef0..e451ae156109 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -2729,6 +2729,13 @@ breaking change for routes that reference the header by
its literal string name:
`CamelIBMCOSCacheControl`, although the change is trivial to adapt. Routes
using the
`IBMCOSConstants.CACHE_CONTROL` constant are unaffected.
+=== camel-ibm-cos - objects in progress in the consumer
+
+With `fileName` set, the consumer now skips the object while an exchange of an
earlier poll is still processing it, as
+it already did when listing the bucket (with `deleteAfterRead=false` it is
consumed again once its exchange is done).
+The objects of a poll that a stopping consumer does not process are no longer
kept in the `inProgressRepository`, so
+they are consumed when the route is started again, and their content streams
are closed.
+
=== camel-ibm-secrets-manager - a lookup without a secret group uses the
default group
The `ibm:` properties function now uses the `default` secret group when the
lookup does not specify one, such as
