luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixing the tests for the Shutdown fix.

Browse source
This commit is contained in:
Sahil Palvia 2017-11-09 14:35:41 -08:00
parent f34d6de4f2
commit 5dac42039b
1 changed files with 34 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java
View file

@ -105,6 +105,8 @@ public class ShardConsumerTest {
private GetRecordsCache getRecordsCache; private GetRecordsCache getRecordsCache;
private KinesisDataFetcher dataFetcher;
@Mock @Mock
private IRecordProcessor processor; private IRecordProcessor processor;
@Mock @Mock
@ -121,6 +123,7 @@ public class ShardConsumerTest {
@Before @Before
public void setup() { public void setup() {
getRecordsCache = null; getRecordsCache = null;
dataFetcher = null;
recordsFetcherFactory = spy(new SimpleRecordsFetcherFactory(maxRecords)); recordsFetcherFactory = spy(new SimpleRecordsFetcherFactory(maxRecords));
when(config.getRecordsFetcherFactory()).thenReturn(recordsFetcherFactory); when(config.getRecordsFetcherFactory()).thenReturn(recordsFetcherFactory);
@ -342,7 +345,7 @@ public class ShardConsumerTest {
) )
); );
KinesisDataFetcher dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo); dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo);
getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords, getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords,
new SynchronousGetRecordsRetrievalStrategy(dataFetcher))); new SynchronousGetRecordsRetrievalStrategy(dataFetcher)));
@ -474,18 +477,42 @@ public class ShardConsumerTest {
skipCheckpointValidationValue, INITIAL_POSITION_LATEST); skipCheckpointValidationValue, INITIAL_POSITION_LATEST);
ShardInfo shardInfo = new ShardInfo(streamShardId, testConcurrencyToken, null, null); ShardInfo shardInfo = new ShardInfo(streamShardId, testConcurrencyToken, null, null);
dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo);
getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords,
new SynchronousGetRecordsRetrievalStrategy(dataFetcher)));
when(recordsFetcherFactory.createRecordsFetcher(any(GetRecordsRetrievalStrategy.class), anyString(),
any(IMetricsFactory.class)))
.thenReturn(getRecordsCache);
RecordProcessorCheckpointer recordProcessorCheckpointer = new RecordProcessorCheckpointer(
shardInfo,
checkpoint,
new SequenceNumberValidator(
streamConfig.getStreamProxy(),
shardInfo.getShardId(),
streamConfig.shouldValidateSequenceNumberBeforeCheckpointing()
)
);
ShardConsumer consumer = ShardConsumer consumer =
new ShardConsumer(shardInfo, new ShardConsumer(shardInfo,
streamConfig, streamConfig,
checkpoint, checkpoint,
processor, processor,
recordProcessorCheckpointer,
leaseManager, leaseManager,
parentShardPollIntervalMillis, parentShardPollIntervalMillis,
cleanupLeasesOfCompletedShards, cleanupLeasesOfCompletedShards,
executorService, executorService,
metricsFactory, metricsFactory,
taskBackoffTimeMillis, taskBackoffTimeMillis,
KinesisClientLibConfiguration.DEFAULT_SKIP_SHARD_SYNC_AT_STARTUP_IF_LEASES_EXIST); KinesisClientLibConfiguration.DEFAULT_SKIP_SHARD_SYNC_AT_STARTUP_IF_LEASES_EXIST,
dataFetcher,
Optional.empty(),
Optional.empty(),
config);
assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.WAITING_ON_PARENT_SHARDS))); assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.WAITING_ON_PARENT_SHARDS)));
consumer.consumeShard(); // check on parent shards consumer.consumeShard(); // check on parent shards
@ -494,12 +521,13 @@ public class ShardConsumerTest {
assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.INITIALIZING))); assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.INITIALIZING)));
consumer.consumeShard(); // initialize consumer.consumeShard(); // initialize
processor.getInitializeLatch().await(5, TimeUnit.SECONDS); processor.getInitializeLatch().await(5, TimeUnit.SECONDS);
verify(getRecordsCache).start();
// We expect to process all records in numRecs calls // We expect to process all records in numRecs calls
for (int i = 0; i < numRecs;) { for (int i = 0; i < numRecs;) {
boolean newTaskSubmitted = consumer.consumeShard(); boolean newTaskSubmitted = consumer.consumeShard();
if (newTaskSubmitted) { if (newTaskSubmitted) {
LOG.debug("New processing task was submitted, call # " + i); LOG.info("New processing task was submitted, call # " + i);
assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.PROCESSING))); assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.PROCESSING)));
// CHECKSTYLE:IGNORE ModifiedControlVariable FOR NEXT 1 LINES // CHECKSTYLE:IGNORE ModifiedControlVariable FOR NEXT 1 LINES
i += maxRecords; i += maxRecords;
@ -537,6 +565,8 @@ public class ShardConsumerTest {
assertThat(processor.getShutdownReason(), is(equalTo(ShutdownReason.TERMINATE))); assertThat(processor.getShutdownReason(), is(equalTo(ShutdownReason.TERMINATE)));
verify(getRecordsCache).shutdown();
executorService.shutdown(); executorService.shutdown();
executorService.awaitTermination(60, TimeUnit.SECONDS); executorService.awaitTermination(60, TimeUnit.SECONDS);
@ -594,7 +624,7 @@ public class ShardConsumerTest {
) )
); );
KinesisDataFetcher dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo); dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo);
getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords, getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords,
new SynchronousGetRecordsRetrievalStrategy(dataFetcher))); new SynchronousGetRecordsRetrievalStrategy(dataFetcher)));