Addressed comments

This commit is contained in:
Wei 2017-10-03 13:39:29 -07:00
parent 2fc4267b83
commit 192aef6619
9 changed files with 16 additions and 32 deletions

View file

@ -28,7 +28,7 @@ public interface RecordsFetcherFactory {
*
* @return Returns a record fetcher object
*/
GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy);
GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, IMetricsFactory metricsFactory);
void setMaxSize(int maxSize);
@ -40,6 +40,4 @@ public interface RecordsFetcherFactory {
void setIdleMillisBetweenCalls(long idleMillisBetweenCalls);
void setMetricsFactory(IMetricsFactory metricsFactory);
}

View file

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

View file

@ -34,7 +34,7 @@ public class SimpleRecordsFetcherFactory implements RecordsFetcherFactory {
}
@Override
public GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy) {
public GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, IMetricsFactory metricsFactory) {
if(dataFetchingStrategy.equals(DataFetchingStrategy.DEFAULT)) {
return new BlockingGetRecordsCache(maxRecords, getRecordsRetrievalStrategy, idleMillisBetweenCalls);
} else {
@ -63,10 +63,6 @@ public class SimpleRecordsFetcherFactory implements RecordsFetcherFactory {
this.dataFetchingStrategy = dataFetchingStrategy;
}
public void setMetricsFactory(IMetricsFactory metricsFactory) {
this.metricsFactory = metricsFactory;
}
public void setIdleMillisBetweenCalls(final long idleMillisBetweenCalls) {
this.idleMillisBetweenCalls = idleMillisBetweenCalls;
}

View file

@ -415,7 +415,6 @@ public class Worker implements Runnable {
this.shardPrioritization = shardPrioritization;
this.retryGetRecordsInSeconds = retryGetRecordsInSeconds;
this.maxGetRecordsThreadPool = maxGetRecordsThreadPool;
this.config.getRecordsFetcherFactory().setMetricsFactory(metricsFactory);
}
/**

View file

@ -31,12 +31,10 @@ import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
<<<<<<< HEAD
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.metrics.impl.NullMetricsFactory;
import com.amazonaws.services.kinesis.model.Record;
=======
>>>>>>> upstream/prefetch
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@ -88,11 +86,8 @@ public class PrefetchGetRecordsCacheIntegrationTest {
MAX_RECORDS_PER_CALL,
getRecordsRetrievalStrategy,
executorService,
<<<<<<< HEAD
new NullMetricsFactory());
=======
new NullMetricsFactory(),
IDLE_MILLIS_BETWEEN_CALLS);
>>>>>>> upstream/prefetch
}
@Test
@ -136,12 +131,8 @@ public class PrefetchGetRecordsCacheIntegrationTest {
MAX_RECORDS_PER_CALL,
getRecordsRetrievalStrategy2,
executorService2,
<<<<<<< HEAD
new NullMetricsFactory()
);
=======
new NullMetricsFactory(),
IDLE_MILLIS_BETWEEN_CALLS);
>>>>>>> upstream/prefetch
getRecordsCache.start();
sleep(IDLE_MILLIS_BETWEEN_CALLS);

View file

@ -82,11 +82,8 @@ public class PrefetchGetRecordsCacheTest {
MAX_RECORDS_PER_CALL,
getRecordsRetrievalStrategy,
executorService,
<<<<<<< HEAD
new NullMetricsFactory());
=======
new NullMetricsFactory(),
IDLE_MILLIS_BETWEEN_CALLS);
>>>>>>> upstream/prefetch
spyQueue = spy(getRecordsCache.getRecordsResultQueue);
records = spy(new ArrayList<>());

View file

@ -1,6 +1,7 @@
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
import com.amazonaws.services.kinesis.metrics.impl.NullMetricsFactory;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
@ -20,23 +21,25 @@ public class RecordsFetcherFactoryTest {
@Mock
private GetRecordsRetrievalStrategy getRecordsRetrievalStrategy;
@Mock
private IMetricsFactory metricsFactory;
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
recordsFetcherFactory = new SimpleRecordsFetcherFactory(1);
recordsFetcherFactory.setMetricsFactory(new NullMetricsFactory());
}
@Test
public void createDefaultRecordsFetcherTest() {
GetRecordsCache recordsCache = recordsFetcherFactory.createRecordsFetcher(getRecordsRetrievalStrategy);
GetRecordsCache recordsCache = recordsFetcherFactory.createRecordsFetcher(getRecordsRetrievalStrategy, metricsFactory);
assertThat(recordsCache, instanceOf(BlockingGetRecordsCache.class));
}
@Test
public void createPrefetchRecordsFetcherTest() {
recordsFetcherFactory.setDataFetchingStrategy(DataFetchingStrategy.PREFETCH_CACHED);
GetRecordsCache recordsCache = recordsFetcherFactory.createRecordsFetcher(getRecordsRetrievalStrategy);
GetRecordsCache recordsCache = recordsFetcherFactory.createRecordsFetcher(getRecordsRetrievalStrategy, metricsFactory);
assertThat(recordsCache, instanceOf(PrefetchGetRecordsCache.class));
}

View file

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

View file

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