You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2024/08/20 09:56:44 UTC
(camel) branch kin-shard created (now 0768aa7ba10)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch kin-shard
in repository https://gitbox.apache.org/repos/asf/camel.git
at 0768aa7ba10 CAMEL-20958: camel-aws - Kinesis consumer loops on closed shards
This branch includes the following new commits:
new 0768aa7ba10 CAMEL-20958: camel-aws - Kinesis consumer loops on closed shards
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-20958: camel-aws - Kinesis consumer loops on closed shards
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch kin-shard
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0768aa7ba10e04b33a3204f1cdb0a67c41c3fb38
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Aug 20 11:56:27 2024 +0200
CAMEL-20958: camel-aws - Kinesis consumer loops on closed shards
---
.../camel/component/aws2/kinesis/aws2-kinesis.json | 4 +--
.../aws2/kinesis/Kinesis2Configuration.java | 6 ++--
.../component/aws2/kinesis/Kinesis2Consumer.java | 24 ++++++++++---
.../KinesisConsumerClosedShardWithSilentTest.java | 39 +++++++++++++---------
4 files changed, 48 insertions(+), 25 deletions(-)
diff --git a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
index 90713f67676..5f46322e0b6 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
+++ b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
@@ -32,7 +32,7 @@
"iteratorType": { "index": 6, "kind": "property", "displayName": "Iterator Type", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.kinesis.model.ShardIteratorType", "enum": [ "AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "TRIM_HORIZON", "LATEST", "AT_TIMESTAMP", "null" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "TRIM_HORIZON", "configurationClass": "org.apache.camel.component [...]
"maxResultsPerRequest": { "index": 7, "kind": "property", "displayName": "Max Results Per Request", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Maximum number of records that will be fetched in each poll" },
"sequenceNumber": { "index": 8, "kind": "property", "displayName": "Sequence Number", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "The sequence number to start polling from. Required if iteratorType is set to AFTER_SEQUENC [...]
- "shardClosed": { "index": 9, "kind": "property", "displayName": "Shard Closed", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum", "enum": [ "ignore", "fail", "silent" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ignore", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configur [...]
+ "shardClosed": { "index": 9, "kind": "property", "displayName": "Shard Closed", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum", "enum": [ "ignore", "fail", "silent" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ignore", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configur [...]
"shardId": { "index": 10, "kind": "property", "displayName": "Shard Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Defines which shardId in the Kinesis stream to get records from" },
"shardMonitorInterval": { "index": 11, "kind": "property", "displayName": "Shard Monitor Interval", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10000, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "The interval in milliseconds to wait between [...]
"lazyStartProducer": { "index": 12, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fai [...]
@@ -74,7 +74,7 @@
"maxResultsPerRequest": { "index": 6, "kind": "parameter", "displayName": "Max Results Per Request", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Maximum number of records that will be fetched in each poll" },
"sendEmptyMessageWhenIdle": { "index": 7, "kind": "parameter", "displayName": "Send Empty Message When Idle", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead." },
"sequenceNumber": { "index": 8, "kind": "parameter", "displayName": "Sequence Number", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "The sequence number to start polling from. Required if iteratorType is set to AFTER_SEQUEN [...]
- "shardClosed": { "index": 9, "kind": "parameter", "displayName": "Shard Closed", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum", "enum": [ "ignore", "fail", "silent" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ignore", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configu [...]
+ "shardClosed": { "index": 9, "kind": "parameter", "displayName": "Shard Closed", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum", "enum": [ "ignore", "fail", "silent" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ignore", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configu [...]
"shardId": { "index": 10, "kind": "parameter", "displayName": "Shard Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Defines which shardId in the Kinesis stream to get records from" },
"bridgeErrorHandler": { "index": 11, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming [...]
"exceptionHandler": { "index": 12, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By de [...]
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java
index 6ba5d24d338..ae2b1ed1682 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java
@@ -60,9 +60,9 @@ public class Kinesis2Configuration implements Cloneable {
private String sequenceNumber = "";
@UriParam(label = "consumer", defaultValue = "ignore",
description = "Define what will be the behavior in case of shard closed. Possible value are ignore, silent and fail."
- + " In case of ignore a message will be logged and the consumer will restart from the beginning,"
- + "in case of silent there will be no logging and the consumer will start from the beginning,"
- + "in case of fail a ReachedClosedStateException will be raised")
+ + " In case of ignore a WARN message will be logged once and the consumer will not process new messages until restarted,"
+ + "in case of silent there will be no logging and the consumer will not process new messages until restarted,"
+ + "in case of fail a ReachedClosedStateException will be thrown")
private Kinesis2ShardClosedStrategyEnum shardClosed;
@UriParam(label = "proxy", enums = "HTTP,HTTPS", defaultValue = "HTTPS",
description = "To define a proxy protocol when instantiating the Kinesis client")
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index c33cedab3ba..85fa04ecc25 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -53,6 +53,7 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
private ResumeStrategy resumeStrategy;
private final Map<String, String> currentShardIterators = new java.util.HashMap<>();
+ private final Set<String> warnLogged = new HashSet<>();
private volatile List<Shard> currentShardList = List.of();
private static final String SHARD_MONITOR_EXECUTOR_NAME = "Kinesis_shard_monitor";
@@ -71,11 +72,23 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
this.connection = connection;
}
+ public boolean isShardClosed(String shardId) {
+ return currentShardIterators.get(shardId) == null && currentShardIterators.containsKey(shardId);
+ }
+
@Override
protected int poll() throws Exception {
var processedExchangeCount = new AtomicInteger(0);
- if (!getEndpoint().getConfiguration().getShardId().isEmpty()) {
+ String shardId = getEndpoint().getConfiguration().getShardId();
+ if (!shardId.isEmpty()) {
+ // skip if the shard is closed
+ if (isShardClosed(shardId)) {
+ // There was previously a shardIterator but shard is now closed
+ handleClosedShard(shardId);
+ return 0;
+ }
+
var request = DescribeStreamRequest
.builder()
.streamName(getEndpoint().getConfiguration().getStreamName())
@@ -129,7 +142,7 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
final Shard shard,
final KinesisConnection kinesisConnection,
AtomicInteger processedExchangeCount) {
- String shardIterator = null;
+ String shardIterator;
try {
shardIterator = getShardIterator(shard, kinesisConnection);
} catch (InterruptedException e) {
@@ -222,6 +235,7 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
if (currentShardIterators.containsKey(shardId)) {
// There was previously a shardIterator but shard is now closed
handleClosedShard(shardId);
+ return null; // we cannot get the shard again as its closed
}
GetShardIteratorRequest.Builder request = GetShardIteratorRequest.builder()
@@ -264,8 +278,10 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
private void handleClosedShard(String shardId) {
switch (getEndpoint().getConfiguration().getShardClosed()) {
case ignore:
- LOG.warn("The shard with id={} on stream {} reached CLOSE status",
- shardId, getEndpoint().getConfiguration().getStreamName());
+ if (warnLogged.add(shardId)) {
+ LOG.warn("The shard with id={} on stream {} reached CLOSE status",
+ shardId, getEndpoint().getConfiguration().getStreamName());
+ }
break;
case silent:
break;
diff --git a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
index 643bb199ad9..3c7303b8b2a 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -70,22 +71,23 @@ public class KinesisConsumerClosedShardWithSilentTest {
@BeforeEach
public void setup() {
- SequenceNumberRange range = SequenceNumberRange.builder().endingSequenceNumber("20").build();
+ SequenceNumberRange range = SequenceNumberRange.builder().endingSequenceNumber("2").build();
Shard shard = Shard.builder().shardId("shardId").sequenceNumberRange(range).build();
ArrayList<Shard> shardList = new ArrayList<>();
shardList.add(shard);
+ var r1 = GetRecordsResponse.builder()
+ .nextShardIterator(null)
+ .records(
+ Record.builder().sequenceNumber("1")
+ .data(SdkBytes.fromString("Hello", Charset.defaultCharset()))
+ .build(),
+ Record.builder().sequenceNumber("2")
+ .data(SdkBytes.fromString("Bye", Charset.defaultCharset()))
+ .build())
+ .build();
when(kinesisClient
- .getRecords(any(GetRecordsRequest.class))).thenReturn(GetRecordsResponse.builder()
- .nextShardIterator("nextShardIterator")
- .records(
- Record.builder().sequenceNumber("1")
- .data(SdkBytes.fromString("Hello", Charset.defaultCharset()))
- .build(),
- Record.builder().sequenceNumber("2")
- .data(SdkBytes.fromString("Hello", Charset.defaultCharset()))
- .build())
- .build());
+ .getRecords(any(GetRecordsRequest.class))).thenReturn(r1);
when(kinesisClient
.getShardIterator(any(GetShardIteratorRequest.class)))
.thenReturn(GetShardIteratorResponse.builder().shardIterator("shardIterator").build());
@@ -179,15 +181,20 @@ public class KinesisConsumerClosedShardWithSilentTest {
}
@Test
- public void itUsesTheShardIteratorOnSubsiquentPolls() throws Exception {
- underTest.poll();
- underTest.poll();
+ public void itUsesTheShardIteratorOnSubsequentPolls() throws Exception {
+ // should not be closed on start
+ Assertions.assertFalse(underTest.isShardClosed("shardId"));
+
+ underTest.poll(); // pull records and reached EOL of shard
+ underTest.poll(); // the shard is now closed
final ArgumentCaptor<GetRecordsRequest> getRecordsReqCap = ArgumentCaptor.forClass(GetRecordsRequest.class);
// On second call it uses the one returned from the first call
verify(kinesisClient, times(1)).getShardIterator(any(GetShardIteratorRequest.class));
- verify(kinesisClient, times(2)).getRecords(getRecordsReqCap.capture());
+ verify(kinesisClient, times(1)).getRecords(getRecordsReqCap.capture());
assertThat(getRecordsReqCap.getAllValues().get(0).shardIterator(), is("shardIterator"));
- assertThat(getRecordsReqCap.getAllValues().get(1).shardIterator(), is("nextShardIterator"));
+
+ // should be closed
+ Assertions.assertTrue(underTest.isShardClosed("shardId"));
}
}