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

Addressed comments

Browse source
This commit is contained in:
Wei 2017-10-06 13:34:57 -07:00
parent a2fd07c7b0
commit be34a50f59
13 changed files with 23 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/BlockingGetRecordsCache.java
View file

@ -49,11 +49,6 @@ public class BlockingGetRecordsCache implements GetRecordsCache {
//
}
@Override
public void setMetricOperation(String operation){
}
@Override
public ProcessRecordsInput getNextResult() {
sleepBeforeNextCall();

2
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/GetRecordsCache.java
View file

@ -41,6 +41,4 @@ public interface GetRecordsCache {
* This method calls the shutdown behavior on the cache, if available.
*/
void shutdown();
void setMetricOperation(String operation);
}

1
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/InitializeTask.java
View file

@ -83,7 +83,6 @@ class InitializeTask implements ITask {
ExtendedSequenceNumber initialCheckpoint = initialCheckpointObject.getCheckpoint();
dataFetcher.initialize(initialCheckpoint.getSequenceNumber(), streamConfig.getInitialPositionInStream());
getRecordsCache.setMetricOperation("InitializeTask");
getRecordsCache.start();
recordProcessorCheckpointer.setLargestPermittedCheckpointValue(initialCheckpoint);
recordProcessorCheckpointer.setInitialCheckpointValue(initialCheckpoint);

8
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/PrefetchGetRecordsCache.java
View file

@ -78,7 +78,8 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
@NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
@NonNull final ExecutorService executorService,
long idleMillisBetweenCalls,
@NonNull final IMetricsFactory metricsFactory) {
@NonNull final IMetricsFactory metricsFactory,
@NonNull String operation) {
this.getRecordsRetrievalStrategy = getRecordsRetrievalStrategy;
this.maxRecordsPerCall = maxRecordsPerCall;
this.maxPendingProcessRecordsInput = maxPendingProcessRecordsInput;
@ -90,6 +91,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
this.metricsFactory = new ThreadSafeMetricsDelegatingFactory(metricsFactory);
this.idleMillisBetweenCalls = idleMillisBetweenCalls;
this.defaultGetRecordsCacheDaemon = new DefaultGetRecordsCacheDaemon();
this.operation = operation;
}
@Override
@ -129,10 +131,6 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
return getRecordsRetrievalStrategy;
}
@Override
public void setMetricOperation(String operation) {
this.operation = operation;
}
@Override
public void shutdown() {
defaultGetRecordsCacheDaemon.isShutdown = true;

1
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ProcessTask.java
View file

@ -148,7 +148,6 @@ class ProcessTask implements ITask {
scope.addDimension(MetricsHelper.SHARD_ID_DIMENSION_NAME, shardInfo.getShardId());
scope.addData(RECORDS_PROCESSED_METRIC, 0, StandardUnit.Count, MetricsLevel.SUMMARY);
scope.addData(DATA_BYTES_PROCESSED_METRIC, 0, StandardUnit.Bytes, MetricsLevel.SUMMARY);
getRecordsCache.setMetricOperation("ProcessTask");
Exception exception = null;
try {

4
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordsFetcherFactory.java
View file

@ -25,10 +25,12 @@ public interface RecordsFetcherFactory {
*
* @param getRecordsRetrievalStrategy GetRecordsRetrievalStrategy to be used with the GetRecordsCache
* @param shardId ShardId of the shard that the fetcher will retrieve records for
* @param metricsFactory MetricsFactory used to create metricScope
* @param operation operation used to set metricScope
*
* @return GetRecordsCache used to get records from Kinesis.
*/
GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, String shardId, IMetricsFactory metricsFactory);
GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, String shardId, IMetricsFactory metricsFactory, String operation);
/**
* Sets the maximum number of ProcessRecordsInput objects the GetRecordsCache can hold, before further requests are

2
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java
View file

@ -235,7 +235,7 @@ class ShardConsumer {
this.dataFetcher = kinesisDataFetcher;
this.getRecordsCache = config.getRecordsFetcherFactory().createRecordsFetcher(
makeStrategy(this.dataFetcher, retryGetRecordsInSeconds, maxGetRecordsThreadPool, this.shardInfo),
this.getShardInfo().getShardId(), metricsFactory);
this.getShardInfo().getShardId(), metricsFactory, "ProcessTask");
}
/**

4
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/SimpleRecordsFetcherFactory.java
View file

@ -35,7 +35,7 @@ public class SimpleRecordsFetcherFactory implements RecordsFetcherFactory {
}
@Override
public GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, String shardId, IMetricsFactory metricsFactory) {
public GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, String shardId, IMetricsFactory metricsFactory, String operation) {
if(dataFetchingStrategy.equals(DataFetchingStrategy.DEFAULT)) {
return new BlockingGetRecordsCache(maxRecords, getRecordsRetrievalStrategy, idleMillisBetweenCalls);
} else {
@ -45,7 +45,7 @@ public class SimpleRecordsFetcherFactory implements RecordsFetcherFactory {
.setDaemon(true)
.setNameFormat("prefetch-cache-" + shardId + "-%04d")
.build()),
idleMillisBetweenCalls, metricsFactory);
idleMillisBetweenCalls, metricsFactory, operation);
}
}

7
src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/PrefetchGetRecordsCacheIntegrationTest.java
View file

@ -66,6 +66,7 @@ public class PrefetchGetRecordsCacheIntegrationTest {
private KinesisDataFetcher dataFetcher;
private ExecutorService executorService;
private List<Record> records;
private String operation = "ProcessTask";
@Mock
private IKinesisProxy proxy;
@ -87,7 +88,8 @@ public class PrefetchGetRecordsCacheIntegrationTest {
getRecordsRetrievalStrategy,
executorService,
IDLE_MILLIS_BETWEEN_CALLS,
new NullMetricsFactory());
new NullMetricsFactory(),
operation);
}
@Test
@ -132,7 +134,8 @@ public class PrefetchGetRecordsCacheIntegrationTest {
getRecordsRetrievalStrategy2,
executorService2,
IDLE_MILLIS_BETWEEN_CALLS,
new NullMetricsFactory());
new NullMetricsFactory(),
operation);
getRecordsCache.start();
sleep(IDLE_MILLIS_BETWEEN_CALLS);

4
src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/PrefetchGetRecordsCacheTest.java
View file

@ -71,6 +71,7 @@ public class PrefetchGetRecordsCacheTest {
private ExecutorService executorService;
private LinkedBlockingQueue<ProcessRecordsInput> spyQueue;
private PrefetchGetRecordsCache getRecordsCache;
private String operation = "ProcessTask";
@Before
public void setup() {
@ -83,7 +84,8 @@ public class PrefetchGetRecordsCacheTest {
getRecordsRetrievalStrategy,
executorService,
IDLE_MILLIS_BETWEEN_CALLS,
new NullMetricsFactory());
new NullMetricsFactory(),
operation);
spyQueue = spy(getRecordsCache.getRecordsResultQueue);
records = spy(new ArrayList<>());

5
src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordsFetcherFactoryTest.java
View file

@ -13,6 +13,7 @@ import org.mockito.MockitoAnnotations;
public class RecordsFetcherFactoryTest {
private String shardId = "TestShard";
private RecordsFetcherFactory recordsFetcherFactory;
private String operation = "ProcessTask";
@Mock
private GetRecordsRetrievalStrategy getRecordsRetrievalStrategy;
@ -29,7 +30,7 @@ public class RecordsFetcherFactoryTest {
@Test
public void createDefaultRecordsFetcherTest() {
GetRecordsCache recordsCache = recordsFetcherFactory.createRecordsFetcher(getRecordsRetrievalStrategy, shardId,
metricsFactory);
metricsFactory, operation);
assertThat(recordsCache, instanceOf(BlockingGetRecordsCache.class));
}
@ -37,7 +38,7 @@ public class RecordsFetcherFactoryTest {
public void createPrefetchRecordsFetcherTest() {
recordsFetcherFactory.setDataFetchingStrategy(DataFetchingStrategy.PREFETCH_CACHED);
GetRecordsCache recordsCache = recordsFetcherFactory.createRecordsFetcher(getRecordsRetrievalStrategy, shardId,
metricsFactory);
metricsFactory, operation);
assertThat(recordsCache, instanceOf(PrefetchGetRecordsCache.class));
}

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

@ -341,7 +341,7 @@ public class ShardConsumerTest {
getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords,
new SynchronousGetRecordsRetrievalStrategy(dataFetcher),
0L));
when(recordsFetcherFactory.createRecordsFetcher(any(), anyString(),any())).thenReturn(getRecordsCache);
when(recordsFetcherFactory.createRecordsFetcher(any(), anyString(),any(),anyString())).thenReturn(getRecordsCache);
ShardConsumer consumer =
new ShardConsumer(shardInfo,
@ -471,7 +471,7 @@ public class ShardConsumerTest {
getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords,
new SynchronousGetRecordsRetrievalStrategy(dataFetcher),
0L));
when(recordsFetcherFactory.createRecordsFetcher(any(), anyString(),any())).thenReturn(getRecordsCache);
when(recordsFetcherFactory.createRecordsFetcher(any(), anyString(),any(),anyString())).thenReturn(getRecordsCache);
ShardConsumer consumer =
new ShardConsumer(shardInfo,

2
src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/WorkerTest.java
View file

@ -621,7 +621,7 @@ public class WorkerTest {
RecordsFetcherFactory recordsFetcherFactory = mock(RecordsFetcherFactory.class);
GetRecordsCache getRecordsCache = mock(GetRecordsCache.class);
when(config.getRecordsFetcherFactory()).thenReturn(recordsFetcherFactory);
when(recordsFetcherFactory.createRecordsFetcher(any(), anyString(),any())).thenReturn(getRecordsCache);
when(recordsFetcherFactory.createRecordsFetcher(any(), anyString(),any(), anyString())).thenReturn(getRecordsCache);
when(getRecordsCache.getNextResult()).thenReturn(new ProcessRecordsInput().withRecords(Collections.emptyList()).withMillisBehindLatest(0L));
WorkerThread workerThread = runWorker(shardList,