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 2016/01/21 10:14:59 UTC
[6/9] camel git commit: CAMEL-9515 Add consumer handling of
ExpiredShardExceptions.
CAMEL-9515 Add consumer handling of ExpiredShardExceptions.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/9c99648c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/9c99648c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/9c99648c
Branch: refs/heads/master
Commit: 9c99648c601097bda6c5b08360b1c45c345793d8
Parents: 98a33dd
Author: Candle <ca...@candle.me.uk>
Authored: Wed Jan 20 10:27:47 2016 +0000
Committer: Claus Ibsen <da...@apache.org>
Committed: Thu Jan 21 10:10:26 2016 +0100
----------------------------------------------------------------------
.../aws/ddbstream/DdbStreamConsumer.java | 34 ++++++--
.../aws/ddbstream/ShardIteratorHandler.java | 2 +-
.../aws/ddbstream/DdbStreamConsumerTest.java | 86 ++++++++++++++------
.../aws/ddbstream/ShardIteratorHandlerTest.java | 13 ++-
4 files changed, 94 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/9c99648c/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
index c407df6..b3e451e 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Queue;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams;
+import com.amazonaws.services.dynamodbv2.model.ExpiredIteratorException;
import com.amazonaws.services.dynamodbv2.model.GetRecordsRequest;
import com.amazonaws.services.dynamodbv2.model.GetRecordsResult;
import com.amazonaws.services.dynamodbv2.model.Record;
@@ -48,17 +49,30 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer {
this.shardIteratorHandler = shardIteratorHandler;
}
+ private String lastSeenSequenceNumber;
@Override
protected int poll() throws Exception {
- GetRecordsRequest req = new GetRecordsRequest()
- .withShardIterator(shardIteratorHandler.getShardIterator())
- .withLimit(getEndpoint().getMaxResultsPerRequest());
- GetRecordsResult result = getClient().getRecords(req);
+ GetRecordsResult result;
+ try {
+ GetRecordsRequest req = new GetRecordsRequest()
+ .withShardIterator(shardIteratorHandler.getShardIterator(null))
+ .withLimit(getEndpoint().getMaxResultsPerRequest());
+ result = getClient().getRecords(req);
+ } catch (ExpiredIteratorException e) {
+ LOG.warn("Expired Shard Iterator, attempting to resume from " + lastSeenSequenceNumber, e);
+ GetRecordsRequest req = new GetRecordsRequest()
+ .withShardIterator(shardIteratorHandler.getShardIterator(lastSeenSequenceNumber))
+ .withLimit(getEndpoint().getMaxResultsPerRequest());
+ result = getClient().getRecords(req);
+ }
- Queue<Exchange> exchanges = createExchanges(result.getRecords());
+ Queue<Exchange> exchanges = createExchanges(result.getRecords(), lastSeenSequenceNumber);
int processedExchangeCount = processBatch(CastUtils.cast(exchanges));
shardIteratorHandler.updateShardIterator(result.getNextShardIterator());
+ if (!result.getRecords().isEmpty()) {
+ lastSeenSequenceNumber = result.getRecords().get((result.getRecords().size()-1)).getDynamodb().getSequenceNumber();
+ }
return processedExchangeCount;
}
@@ -90,10 +104,14 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer {
return (DdbStreamEndpoint) super.getEndpoint();
}
- private Queue<Exchange> createExchanges(List<Record> records) {
+ private Queue<Exchange> createExchanges(List<Record> records, String lastSeenSequenceNumber) {
Queue<Exchange> exchanges = new ArrayDeque<>();
- BigIntComparisons condition;
+ BigIntComparisons condition = null;
BigInteger providedSeqNum = null;
+ if (lastSeenSequenceNumber != null) {
+ providedSeqNum = new BigInteger(lastSeenSequenceNumber);
+ condition = BigIntComparisons.Conditions.LT;
+ }
switch(getEndpoint().getIteratorType()) {
case AFTER_SEQUENCE_NUMBER:
condition = BigIntComparisons.Conditions.LT;
@@ -103,8 +121,6 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer {
condition = BigIntComparisons.Conditions.LTEQ;
providedSeqNum = new BigInteger(getEndpoint().getSequenceNumberProvider().getSequenceNumber());
break;
- default:
- condition = null;
}
for (Record record : records) {
BigInteger recordSeqNum = new BigInteger(record.getDynamodb().getSequenceNumber());
http://git-wip-us.apache.org/repos/asf/camel/blob/9c99648c/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
index 8801991..685907b 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
@@ -43,7 +43,7 @@ class ShardIteratorHandler {
this.endpoint = endpoint;
}
- String getShardIterator() {
+ String getShardIterator(String resumeFromSequenceNumber) {
// either return a cached one or get a new one via a GetShardIterator request.
if (currentShardIterator == null) {
ListStreamsRequest req0 = new ListStreamsRequest()
http://git-wip-us.apache.org/repos/asf/camel/blob/9c99648c/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
index 13e6d92..c91fc86 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
@@ -25,6 +25,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams;
+import com.amazonaws.services.dynamodbv2.model.ExpiredIteratorException;
import com.amazonaws.services.dynamodbv2.model.GetRecordsRequest;
import com.amazonaws.services.dynamodbv2.model.GetRecordsResult;
import com.amazonaws.services.dynamodbv2.model.Record;
@@ -40,6 +41,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
+import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
@@ -47,6 +49,7 @@ import org.mockito.stubbing.Answer;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -62,6 +65,7 @@ public class DdbStreamConsumerTest {
private final CamelContext context = new DefaultCamelContext();
private final DdbStreamComponent component = new DdbStreamComponent(context);
private final DdbStreamEndpoint endpoint = new DdbStreamEndpoint(null, "table_name", component);
+ private GetRecordsAnswer recordsAnswer;
@Before
public void setup() throws Exception {
@@ -82,28 +86,8 @@ public class DdbStreamConsumerTest {
answers.put("shard_iterator_b_002", createRecords("14"));
answers.put("shard_iterator_d_000", createRecords("21", "25"));
answers.put("shard_iterator_d_001", createRecords("30", "35", "40"));
- when(amazonDynamoDBStreams.getRecords(any(GetRecordsRequest.class))).thenAnswer(new Answer<GetRecordsResult>() {
- @Override
- public GetRecordsResult answer(InvocationOnMock invocation) throws Throwable {
- final String shardIterator = ((GetRecordsRequest) invocation.getArguments()[0]).getShardIterator();
- // note that HashMap returns null when there is no entry in the map.
- // A null 'nextShardIterator' indicates that the shard has finished
- // and we should move onto the next shard.
- String nextShardIterator = shardIterators.get(shardIterator);
- Matcher m = Pattern.compile("shard_iterator_d_0*(\\d+)").matcher(shardIterator);
- Collection<Record> ans = answers.get(shardIterator);
- if (nextShardIterator == null && m.matches()) { // last shard iterates forever.
- Integer num = Integer.parseInt(m.group(1));
- nextShardIterator = "shard_iterator_d_" + pad(Integer.toString(num + 1), 3);
- }
- if (null == ans) { // default to an empty list of records.
- ans = createRecords();
- }
- return new GetRecordsResult()
- .withRecords(ans)
- .withNextShardIterator(nextShardIterator);
- }
- });
+ recordsAnswer = new GetRecordsAnswer(shardIterators, answers);
+ when(amazonDynamoDBStreams.getRecords(any(GetRecordsRequest.class))).thenAnswer(recordsAnswer);
}
String pad(String num, int to) {
@@ -119,10 +103,32 @@ public class DdbStreamConsumerTest {
}
@Test
+ public void itResumesFromAfterTheLastSeenSequenceNumberWhenAShardIteratorHasExpired() throws Exception {
+ endpoint.setIteratorType(ShardIteratorType.LATEST);
+ when(shardIteratorHandler.getShardIterator(anyString())).thenReturn("shard_iterator_b_000", "shard_iterator_b_001", "shard_iterator_b_001");
+ Mockito.reset(amazonDynamoDBStreams);
+ when(amazonDynamoDBStreams.getRecords(any(GetRecordsRequest.class)))
+ .thenAnswer(recordsAnswer)
+ .thenThrow(new ExpiredIteratorException("expired shard"))
+ .thenAnswer(recordsAnswer);
+
+ undertest.poll();
+ undertest.poll();
+
+ ArgumentCaptor<Exchange> exchangeCaptor = ArgumentCaptor.forClass(Exchange.class);
+ verify(processor, times(3)).process(exchangeCaptor.capture(), any(AsyncCallback.class));
+ verify(shardIteratorHandler, times(2)).getShardIterator(null); // first poll. Second poll, getRecords fails with an expired shard.
+ verify(shardIteratorHandler).getShardIterator("9"); // second poll, with a resumeFrom.
+ assertThat(exchangeCaptor.getAllValues().get(0).getIn().getBody(Record.class).getDynamodb().getSequenceNumber(), is("9"));
+ assertThat(exchangeCaptor.getAllValues().get(1).getIn().getBody(Record.class).getDynamodb().getSequenceNumber(), is("11"));
+ assertThat(exchangeCaptor.getAllValues().get(2).getIn().getBody(Record.class).getDynamodb().getSequenceNumber(), is("13"));
+ }
+
+ @Test
public void atSeqNumber35GivesFirstRecordWithSeq35() throws Exception {
endpoint.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER);
endpoint.setSequenceNumberProvider(new StaticSequenceNumberProvider("35"));
- when(shardIteratorHandler.getShardIterator()).thenReturn("shard_iterator_d_001", "shard_iterator_d_002");
+ when(shardIteratorHandler.getShardIterator(anyString())).thenReturn("shard_iterator_d_001", "shard_iterator_d_002");
for (int i = 0; i < 10; ++i) { // poll lots.
undertest.poll();
@@ -139,7 +145,7 @@ public class DdbStreamConsumerTest {
public void afterSeqNumber35GivesFirstRecordWithSeq40() throws Exception {
endpoint.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
endpoint.setSequenceNumberProvider(new StaticSequenceNumberProvider("35"));
- when(shardIteratorHandler.getShardIterator()).thenReturn("shard_iterator_d_001", "shard_iterator_d_002");
+ when(shardIteratorHandler.getShardIterator(anyString())).thenReturn("shard_iterator_d_001", "shard_iterator_d_002");
for (int i = 0; i < 10; ++i) { // poll lots.
undertest.poll();
@@ -162,5 +168,37 @@ public class DdbStreamConsumerTest {
return results;
}
+
+ private class GetRecordsAnswer implements Answer<GetRecordsResult> {
+
+ private final Map<String, String> shardIterators;
+ private final Map<String, Collection<Record>> answers;
+
+ GetRecordsAnswer(Map<String, String> shardIterators, Map<String, Collection<Record>> answers) {
+ this.shardIterators = shardIterators;
+ this.answers = answers;
+ }
+
+ @Override
+ public GetRecordsResult answer(InvocationOnMock invocation) throws Throwable {
+ final String shardIterator = ((GetRecordsRequest) invocation.getArguments()[0]).getShardIterator();
+ // note that HashMap returns null when there is no entry in the map.
+ // A null 'nextShardIterator' indicates that the shard has finished
+ // and we should move onto the next shard.
+ String nextShardIterator = shardIterators.get(shardIterator);
+ Matcher m = Pattern.compile("shard_iterator_d_0*(\\d+)").matcher(shardIterator);
+ Collection<Record> ans = answers.get(shardIterator);
+ if (nextShardIterator == null && m.matches()) { // last shard iterates forever.
+ Integer num = Integer.parseInt(m.group(1));
+ nextShardIterator = "shard_iterator_d_" + pad(Integer.toString(num + 1), 3);
+ }
+ if (null == ans) { // default to an empty list of records.
+ ans = createRecords();
+ }
+ return new GetRecordsResult()
+ .withRecords(ans)
+ .withNextShardIterator(nextShardIterator);
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/9c99648c/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandlerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandlerTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandlerTest.java
index 41b7883..c54dbef 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandlerTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandlerTest.java
@@ -94,7 +94,7 @@ public class ShardIteratorHandlerTest {
public void latestOnlyUsesTheLastShard() throws Exception {
endpoint.setIteratorType(ShardIteratorType.LATEST);
- String shardIterator = undertest.getShardIterator();
+ String shardIterator = undertest.getShardIterator(null);
ArgumentCaptor<GetShardIteratorRequest> getIteratorCaptor = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
verify(amazonDynamoDBStreams).getShardIterator(getIteratorCaptor.capture());
@@ -107,7 +107,7 @@ public class ShardIteratorHandlerTest {
endpoint.setIteratorType(ShardIteratorType.LATEST);
undertest.updateShardIterator("bar");
- String shardIterator = undertest.getShardIterator();
+ String shardIterator = undertest.getShardIterator(null);
verify(amazonDynamoDBStreams, times(0)).getShardIterator(any(GetShardIteratorRequest.class));
assertThat(shardIterator, is("bar"));
@@ -117,7 +117,7 @@ public class ShardIteratorHandlerTest {
public void trimHorizonStartsWithTheFirstShard() throws Exception {
endpoint.setIteratorType(ShardIteratorType.TRIM_HORIZON);
- String shardIterator = undertest.getShardIterator();
+ String shardIterator = undertest.getShardIterator(null);
ArgumentCaptor<GetShardIteratorRequest> getIteratorCaptor = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
verify(amazonDynamoDBStreams).getShardIterator(getIteratorCaptor.capture());
@@ -125,7 +125,6 @@ public class ShardIteratorHandlerTest {
assertThat(shardIterator, is("shard_iterator_a_000"));
}
-
@Test
public void trimHorizonWalksAllShards() throws Exception {
endpoint.setIteratorType(ShardIteratorType.TRIM_HORIZON);
@@ -133,7 +132,7 @@ public class ShardIteratorHandlerTest {
String[] shardIterators = new String[4];
for (int i = 0; i < shardIterators.length; ++i) {
- shardIterators[i] = undertest.getShardIterator();
+ shardIterators[i] = undertest.getShardIterator(null);
undertest.updateShardIterator(null);
}
@@ -152,7 +151,7 @@ public class ShardIteratorHandlerTest {
endpoint.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER);
endpoint.setSequenceNumberProvider(new StaticSequenceNumberProvider("12"));
- String shardIterator = undertest.getShardIterator();
+ String shardIterator = undertest.getShardIterator(null);
ArgumentCaptor<GetShardIteratorRequest> getIteratorCaptor = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
verify(amazonDynamoDBStreams).getShardIterator(getIteratorCaptor.capture());
@@ -165,7 +164,7 @@ public class ShardIteratorHandlerTest {
endpoint.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER);
endpoint.setSequenceNumberProvider(new StaticSequenceNumberProvider("16"));
- String shardIterator = undertest.getShardIterator();
+ String shardIterator = undertest.getShardIterator(null);
ArgumentCaptor<GetShardIteratorRequest> getIteratorCaptor = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
verify(amazonDynamoDBStreams).getShardIterator(getIteratorCaptor.capture());