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

Reply via email to