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

Added a test case for BlockingGetRecordsCache. Updated the comments on BlockingGetRecords cache. Synchronized the added and removed methods instead of using volatile variables.

Browse source
This commit is contained in:
Sahil Palvia 2017-09-21 11:50:53 -07:00
parent 3404ddfcf4
commit ce8dd88846
3 changed files with 26 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -36,7 +36,9 @@ public class BlockingGetRecordsCache implements GetRecordsCache {
@Override @Override
public void start() { public void start() {
// Do nothing, this behavior is not supported by this cache. //
// Nothing to do here
//
} }
@Override @Override
@ -50,6 +52,8 @@ public class BlockingGetRecordsCache implements GetRecordsCache {
@Override @Override
public void shutdown() { public void shutdown() {
//
// Nothing to do here. // Nothing to do here.
//
} }
} }

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

@ -116,15 +116,15 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
} }
private class PrefetchCounters { private class PrefetchCounters {
private volatile long size = 0; private long size = 0;
private volatile long byteSize = 0; private long byteSize = 0;
public void added(final ProcessRecordsInput result) { public synchronized void added(final ProcessRecordsInput result) {
size += getSize(result); size += getSize(result);
byteSize += getByteSize(result); byteSize += getByteSize(result);
} }
public void removed(final ProcessRecordsInput result) { public synchronized void removed(final ProcessRecordsInput result) {
size -= getSize(result); size -= getSize(result);
byteSize -= getByteSize(result); byteSize -= getByteSize(result);
} }

19
src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/BlockingGetRecordsCacheTest.java
View file

@ -21,6 +21,7 @@ import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
import java.time.Duration; import java.time.Duration;
import java.util.ArrayList;
import java.util.List; import java.util.List;
import org.junit.Before; import org.junit.Before;
@ -44,9 +45,8 @@ public class BlockingGetRecordsCacheTest {
private GetRecordsRetrievalStrategy getRecordsRetrievalStrategy; private GetRecordsRetrievalStrategy getRecordsRetrievalStrategy;
@Mock @Mock
private GetRecordsResult getRecordsResult; private GetRecordsResult getRecordsResult;
@Mock
private List<Record> records;
private List<Record> records = new ArrayList<>();
private BlockingGetRecordsCache blockingGetRecordsCache; private BlockingGetRecordsCache blockingGetRecordsCache;
@Before @Before
@ -58,7 +58,7 @@ public class BlockingGetRecordsCacheTest {
} }
@Test @Test
public void testGetNextRecords() { public void testGetNextRecordsWithNoRecords() {
ProcessRecordsInput result = blockingGetRecordsCache.getNextResult(); ProcessRecordsInput result = blockingGetRecordsCache.getNextResult();
assertEquals(result.getRecords(), records); assertEquals(result.getRecords(), records);
@ -66,4 +66,17 @@ public class BlockingGetRecordsCacheTest {
assertNull(result.getCacheExitTime()); assertNull(result.getCacheExitTime());
assertEquals(result.getTimeSpentInCache(), Duration.ZERO); assertEquals(result.getTimeSpentInCache(), Duration.ZERO);
} }
@Test
public void testGetNextRecordsWithRecords() {
Record record = new Record();
records.add(record);
records.add(record);
records.add(record);
records.add(record);
ProcessRecordsInput result = blockingGetRecordsCache.getNextResult();
assertEquals(result.getRecords(), records);
}
} }