ShardConsumer implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { String nextShardItr; try { if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_LATEST_SEQUENCE_NUM.get())) { if (subscribedShard.isClosed()) { nextShardItr = null; } else { nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.LATEST.toString(), null); } } else if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())) { nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.TRIM_HORIZON.toString(), null); } else if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get())) { nextShardItr = null; } else if (lastSequenceNum.equals(SentinelSequenceNumber.SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM.get())) { nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.AT_TIMESTAMP.toString(), initTimestamp); } else { if (lastSequenceNum.isAggregated()) { String itrForLastAggregatedRecord = kinesis.getShardIterator(subscribedShard, ShardIteratorType.AT_SEQUENCE_NUMBER.toString(), lastSequenceNum.getSequenceNumber()); GetRecordsResult getRecordsResult = getRecords(itrForLastAggregatedRecord, 1); List<UserRecord> fetchedRecords = deaggregateRecords( getRecordsResult.getRecords(), subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); long lastSubSequenceNum = lastSequenceNum.getSubSequenceNumber(); for (UserRecord record : fetchedRecords) { if (record.getSubSequenceNumber() > lastSubSequenceNum) { deserializeRecordForCollectionAndUpdateState(record); } } nextShardItr = getRecordsResult.getNextShardIterator(); } else { nextShardItr = kinesis.getShardIterator(subscribedShard, ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(), lastSequenceNum.getSequenceNumber()); } } while (isRunning()) { if (nextShardItr == null) { fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); break; } else { if (fetchIntervalMillis != 0) { Thread.sleep(fetchIntervalMillis); } GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); List<UserRecord> fetchedRecords = deaggregateRecords( getRecordsResult.getRecords(), subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); for (UserRecord record : fetchedRecords) { deserializeRecordForCollectionAndUpdateState(record); } nextShardItr = getRecordsResult.getNextShardIterator(); } } } catch (Throwable t) { fetcherRef.stopWithError(t); } } ShardConsumer(KinesisDataFetcher<T> fetcherRef,
						Integer subscribedShardStateIndex,
						StreamShardHandle subscribedShard,
						SequenceNumber lastSequenceNum); protected  ShardConsumer(KinesisDataFetcher<T> fetcherRef,
							Integer subscribedShardStateIndex,
							StreamShardHandle subscribedShard,
							SequenceNumber lastSequenceNum,
							KinesisProxyInterface kinesis); @SuppressWarnings("unchecked") @Override void run();  }
@Test public void testCorrectNumOfCollectedRecordsAndUpdatedState() { StreamShardHandle fakeToBeConsumedShard = new StreamShardHandle( "fakeStream", new Shard() .withShardId(KinesisShardIdGenerator.generateFromShardOrder(0)) .withHashKeyRange( new HashKeyRange() .withStartingHashKey("0") .withEndingHashKey(new BigInteger(StringUtils.repeat("FF", 16), 16).toString()))); LinkedList<KinesisStreamShardState> subscribedShardsStateUnderTest = new LinkedList<>(); subscribedShardsStateUnderTest.add( new KinesisStreamShardState(KinesisDataFetcher.convertToStreamShardMetadata(fakeToBeConsumedShard), fakeToBeConsumedShard, new SequenceNumber("fakeStartingState"))); TestSourceContext<String> sourceContext = new TestSourceContext<>(); TestableKinesisDataFetcher<String> fetcher = new TestableKinesisDataFetcher<>( Collections.singletonList("fakeStream"), sourceContext, new Properties(), new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()), 10, 2, new AtomicReference<>(), subscribedShardsStateUnderTest, KinesisDataFetcher.createInitialSubscribedStreamsToLastDiscoveredShardsState(Collections.singletonList("fakeStream")), Mockito.mock(KinesisProxyInterface.class)); new ShardConsumer<>( fetcher, 0, subscribedShardsStateUnderTest.get(0).getStreamShardHandle(), subscribedShardsStateUnderTest.get(0).getLastProcessedSequenceNum(), FakeKinesisBehavioursFactory.totalNumOfRecordsAfterNumOfGetRecordsCalls(1000, 9)).run(); assertEquals(1000, sourceContext.getCollectedOutputs().size()); assertEquals( SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get(), subscribedShardsStateUnderTest.get(0).getLastProcessedSequenceNum()); }
@Test public void testCorrectNumOfCollectedRecordsAndUpdatedStateWithUnexpectedExpiredIterator() { StreamShardHandle fakeToBeConsumedShard = new StreamShardHandle( "fakeStream", new Shard() .withShardId(KinesisShardIdGenerator.generateFromShardOrder(0)) .withHashKeyRange( new HashKeyRange() .withStartingHashKey("0") .withEndingHashKey(new BigInteger(StringUtils.repeat("FF", 16), 16).toString()))); LinkedList<KinesisStreamShardState> subscribedShardsStateUnderTest = new LinkedList<>(); subscribedShardsStateUnderTest.add( new KinesisStreamShardState(KinesisDataFetcher.convertToStreamShardMetadata(fakeToBeConsumedShard), fakeToBeConsumedShard, new SequenceNumber("fakeStartingState"))); TestSourceContext<String> sourceContext = new TestSourceContext<>(); TestableKinesisDataFetcher<String> fetcher = new TestableKinesisDataFetcher<>( Collections.singletonList("fakeStream"), sourceContext, new Properties(), new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()), 10, 2, new AtomicReference<>(), subscribedShardsStateUnderTest, KinesisDataFetcher.createInitialSubscribedStreamsToLastDiscoveredShardsState(Collections.singletonList("fakeStream")), Mockito.mock(KinesisProxyInterface.class)); new ShardConsumer<>( fetcher, 0, subscribedShardsStateUnderTest.get(0).getStreamShardHandle(), subscribedShardsStateUnderTest.get(0).getLastProcessedSequenceNum(), FakeKinesisBehavioursFactory.totalNumOfRecordsAfterNumOfGetRecordsCallsWithUnexpectedExpiredIterator(1000, 9, 7)).run(); assertEquals(1000, sourceContext.getCollectedOutputs().size()); assertEquals( SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get(), subscribedShardsStateUnderTest.get(0).getLastProcessedSequenceNum()); }