CAMEL-20250: resume after restart of Kinesis consumer

components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/kinesis-resume-strategy

@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.kinesis.consumer.KinesisResumeStrategy

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java

@@ -161,7 +161,7 @@ private void fetchAndPrepareRecordsForCamel(
         }
 
         try {
-            Queue<Exchange> exchanges = createExchanges(result.records());
+            Queue<Exchange> exchanges = createExchanges(shard, result.records());
             processedExchangeCount.getAndSet(processBatch(CastUtils.cast(exchanges)));
         } catch (Exception e) {
             throw new RuntimeException(e);
@@ -220,7 +220,7 @@ private String getShardIterator(
                 request.startingSequenceNumber(getEndpoint().getConfiguration().getSequenceNumber());
             }
 
-            resume(request);
+            resume(shardId, request);
 
             GetShardIteratorResponse result;
             if (getEndpoint().getConfiguration().isAsyncClient()) {
@@ -265,7 +265,7 @@ private void handleClosedShard(String shardId) {
         }
     }
 
-    private void resume(GetShardIteratorRequest.Builder req) {
+    private void resume(String shardId, GetShardIteratorRequest.Builder req) {
         if (resumeStrategy == null) {
             return;
         }
@@ -277,26 +277,25 @@ private void resume(GetShardIteratorRequest.Builder req) {
             return;
         }
 
-        adapter.setRequestBuilder(req);
-        adapter.setStreamName(getEndpoint().getConfiguration().getStreamName());
-        adapter.resume();
+        adapter.configureGetShardIteratorRequest(req, getEndpoint().getConfiguration().getStreamName(), shardId);
     }
 
-    private Queue<Exchange> createExchanges(List<Record> records) {
+    private Queue<Exchange> createExchanges(Shard shard, List<Record> records) {
         Queue<Exchange> exchanges = new ArrayDeque<>();
         for (Record dataRecord : records) {
-            exchanges.add(createExchange(dataRecord));
+            exchanges.add(createExchange(shard, dataRecord));
         }
         return exchanges;
     }
 
-    protected Exchange createExchange(Record dataRecord) {
+    protected Exchange createExchange(Shard shard, Record dataRecord) {
         LOG.debug("Received Kinesis record with partition_key={}", dataRecord.partitionKey());
         Exchange exchange = createExchange(true);
         exchange.getIn().setBody(dataRecord.data().asInputStream());
         exchange.getIn().setHeader(Kinesis2Constants.APPROX_ARRIVAL_TIME, dataRecord.approximateArrivalTimestamp());
         exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, dataRecord.partitionKey());
         exchange.getIn().setHeader(Kinesis2Constants.SEQUENCE_NUMBER, dataRecord.sequenceNumber());
+        exchange.getIn().setHeader(Kinesis2Constants.SHARD_ID, shard.shardId());
         if (dataRecord.approximateArrivalTimestamp() != null) {
             long ts = dataRecord.approximateArrivalTimestamp().getEpochSecond() * 1000;
             exchange.getIn().setHeader(Kinesis2Constants.MESSAGE_TIMESTAMP, ts);

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisConsumerOffsetProcessor.java

@@ -0,0 +1,34 @@
+/*
+ * 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.aws2.kinesis.consumer;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Constants;
+import org.apache.camel.support.resume.Resumables;
+
+public class KinesisConsumerOffsetProcessor implements Processor {
+    @Override
+    public void process(Exchange exchange) throws Exception {
+        exchange.getMessage().setHeader(
+                Exchange.OFFSET,
+                Resumables.of(
+                        exchange.getMessage().getHeader(Kinesis2Constants.SHARD_ID),
+                        exchange.getMessage().getHeader(Kinesis2Constants.SEQUENCE_NUMBER)));
+    }
+}

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisDefaultResumeAdapter.java

@@ -17,10 +17,7 @@
 
 package org.apache.camel.component.aws2.kinesis.consumer;
 
-import java.nio.ByteBuffer;
-
 import org.apache.camel.resume.Cacheable;
-import org.apache.camel.resume.Deserializable;
 import org.apache.camel.resume.Offset;
 import org.apache.camel.resume.OffsetKey;
 import org.apache.camel.resume.ResumeAdapter;
@@ -32,35 +29,14 @@
 import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
 
 @JdkService(ResumeAdapter.RESUME_ADAPTER_FACTORY)
-public class KinesisDefaultResumeAdapter implements KinesisResumeAdapter, Cacheable, Deserializable {
+public class KinesisDefaultResumeAdapter implements KinesisResumeAdapter, Cacheable {
     private static final Logger LOG = LoggerFactory.getLogger(KinesisDefaultResumeAdapter.class);
 
     private ResumeCache<String> cache;
 
-    private GetShardIteratorRequest.Builder resumable;
-    private String streamName;
-
-    public void setRequestBuilder(GetShardIteratorRequest.Builder resumable) {
-        this.resumable = resumable;
-    }
-
     @Override
     public void resume() {
-        assert streamName != null;
-        assert resumable != null;
-
-        KinesisOffset offset = cache.get(streamName, KinesisOffset.class);
-
-        if (offset == null) {
-            LOG.info("There is no offset for the stream {}", streamName);
-            return;
-        }
-
-        final String sequenceNumber = offset.getValue();
-        LOG.info("Resuming from offset {} for key {}", sequenceNumber, streamName);
-
-        resumable.shardIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
-        resumable.startingSequenceNumber(sequenceNumber);
+        throw new UnsupportedOperationException();
     }
 
     private void add(Object key, Object offset) {
@@ -87,17 +63,19 @@ public ResumeCache<?> getCache() {
     }
 
     @Override
-    public boolean deserialize(ByteBuffer keyBuffer, ByteBuffer valueBuffer) {
-        Object keyObj = deserializeKey(keyBuffer);
-        Object valueObj = deserializeValue(valueBuffer);
+    public void configureGetShardIteratorRequest(GetShardIteratorRequest.Builder builder, String streamName, String shardId) {
+        KinesisOffset offset = cache.get(shardId, KinesisOffset.class);
 
-        add(keyObj, valueObj);
+        if (offset == null) {
+            LOG.info("There is no offset for the stream {}", streamName);
+            return;
+        }
 
-        return true;
-    }
+        final String sequenceNumber = offset.getValue();
+        LOG.info("Resuming from offset {} for key {}", sequenceNumber, streamName);
 
-    @Override
-    public void setStreamName(String streamName) {
-        this.streamName = streamName;
+        builder.shardId(shardId);
+        builder.shardIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
+        builder.startingSequenceNumber(sequenceNumber);
     }
 }

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeAdapter.java

@@ -24,17 +24,10 @@
  * The resume adapter for Kinesis
  */
 public interface KinesisResumeAdapter extends ResumeAdapter {
-    /**
-     * Sets the shard iterator request builder that can be used to customize the call and set the exact resume point
-     *
-     * @param builder the builder instance
+    /*
+        When consuming from multiple shards the KinesisResumeAdapter is potentially accessed by multiple threads.
+        To avoid any concurrency issues the configuration of the GetShardIteratorRequest should be done in one operation
+        and not using multiple calls like in the previous version of this interface
      */
-    void setRequestBuilder(GetShardIteratorRequest.Builder builder);
-
-    /**
-     * Sets the stream name being worked on
-     *
-     * @param streamName the stream name
-     */
-    void setStreamName(String streamName);
+    void configureGetShardIteratorRequest(GetShardIteratorRequest.Builder builder, String streamName, String shardId);
 }

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeStrategy.java

@@ -0,0 +1,89 @@
+/*
+ * 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.aws2.kinesis.consumer;
+
+import org.apache.camel.resume.Offset;
+import org.apache.camel.resume.OffsetKey;
+import org.apache.camel.resume.Resumable;
+import org.apache.camel.resume.ResumeAdapter;
+import org.apache.camel.resume.ResumeStrategy;
+import org.apache.camel.resume.ResumeStrategyConfiguration;
+import org.apache.camel.resume.cache.ResumeCache;
+import org.apache.camel.spi.annotations.JdkService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@JdkService("kinesis-resume-strategy")
+public class KinesisResumeStrategy implements ResumeStrategy {
+    private static final Logger LOG = LoggerFactory.getLogger(KinesisResumeStrategy.class);
+
+    private ResumeStrategyConfiguration configuration = KinesisResumeStrategyConfiguration.builder().build();
+    private ResumeCache resumeCache;
+    private ResumeAdapter adapter;
+
+    @Override
+    public void start() {
+        LOG.info("start");
+        this.resumeCache = configuration.getResumeCache();
+    }
+
+    @Override
+    public void stop() {
+        LOG.info("stop");
+    }
+
+    @Override
+    public void setAdapter(ResumeAdapter adapter) {
+        this.adapter = adapter;
+    }
+
+    @Override
+    public ResumeAdapter getAdapter() {
+        return adapter;
+    }
+
+    @Override
+    public <T extends Resumable> void updateLastOffset(T offset) {
+        resumeCache.add(offset.getOffsetKey().getValue().toString(),
+                new KinesisOffset(offset.getLastOffset().getValue(String.class)));
+    }
+
+    @Override
+    public <T extends Resumable> void updateLastOffset(T offset, UpdateCallBack updateCallBack) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?> offsetValue) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void updateLastOffset(OffsetKey<?> offsetKey, Offset<?> offset, UpdateCallBack updateCallBack) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void setResumeStrategyConfiguration(ResumeStrategyConfiguration resumeStrategyConfiguration) {
+        this.configuration = resumeStrategyConfiguration;
+    }
+
+    @Override
+    public ResumeStrategyConfiguration getResumeStrategyConfiguration() {
+        return configuration;
+    }
+}

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/consumer/KinesisResumeStrategyConfiguration.java

@@ -0,0 +1,68 @@
+/*
+ * 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.aws2.kinesis.consumer;
+
+import org.apache.camel.resume.Cacheable;
+import org.apache.camel.resume.ResumeStrategyConfiguration;
+import org.apache.camel.resume.ResumeStrategyConfigurationBuilder;
+import org.apache.camel.resume.cache.ResumeCache;
+
+public class KinesisResumeStrategyConfiguration extends ResumeStrategyConfiguration {
+
+    private KinesisResumeStrategyConfiguration() {
+    }
+
+    @Override
+    public String resumeStrategyService() {
+        return "kinesis-resume-strategy";
+    }
+
+    public static KinesisResumeStrategyConfigurationBuilder builder() {
+        return new KinesisResumeStrategyConfigurationBuilder();
+    }
+
+    public static class KinesisResumeStrategyConfigurationBuilder
+            implements
+            ResumeStrategyConfigurationBuilder<KinesisResumeStrategyConfigurationBuilder, KinesisResumeStrategyConfiguration> {
+
+        private KinesisResumeStrategyConfigurationBuilder() {
+        }
+
+        private Cacheable.FillPolicy fillPolicy = Cacheable.FillPolicy.MAXIMIZING;
+        private ResumeCache resumeCache;
+
+        @Override
+        public KinesisResumeStrategyConfigurationBuilder withCacheFillPolicy(Cacheable.FillPolicy cacheFillPolicy) {
+            this.fillPolicy = cacheFillPolicy;
+            return this;
+        }
+
+        @Override
+        public KinesisResumeStrategyConfigurationBuilder withResumeCache(ResumeCache<?> resumeCache) {
+            this.resumeCache = resumeCache;
+            return this;
+        }
+
+        @Override
+        public KinesisResumeStrategyConfiguration build() {
+            KinesisResumeStrategyConfiguration result = new KinesisResumeStrategyConfiguration();
+            result.setResumeCache(resumeCache);
+            result.setCacheFillPolicy(fillPolicy);
+            return result;
+        }
+    }
+}

components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerIT.java

@@ -54,6 +54,7 @@ public class KinesisConsumerIT extends CamelTestSupport {
     private static class KinesisData {
         private String partition;
         private String body;
+        private String shardId;
 
         @Override
         public String toString() {
@@ -97,6 +98,7 @@ public void configure() {
                             if (message != null) {
                                 data.body = message.getBody(String.class);
                                 data.partition = message.getHeader(Kinesis2Constants.PARTITION_KEY, String.class);
+                                data.shardId = message.getHeader(Kinesis2Constants.SHARD_ID, String.class);
                             }
 
                             receivedMessages.add(data);
@@ -135,6 +137,7 @@ void testProduceMessages() {
             assertTrue(data.partition.endsWith(data.body), "The data/partition mismatch for record: " + data);
             assertNotEquals(partitionKey, data.partition);
             partitionKey = data.partition;
+            assertNotNull(data.shardId);
         }
     }
 }