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

Addressing PR comments.

Browse source
This commit is contained in:
Sahil Palvia 2017-09-19 14:48:43 -07:00
parent 5172f4f936
commit a8edb70552
1 changed files with 28 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -26,8 +26,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
private final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy;
private final ExecutorService executorService;
private volatile int currentSizeInBytes = 0;
private volatile int currentRecordsCount = 0;
private PrefetchCounters prefetchCounters;
private boolean started = false;
@ -41,6 +40,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
this.maxByteSize = maxByteSize;
this.maxRecordsCount = maxRecordsCount;
this.getRecordsResultQueue = new LinkedBlockingQueue<>(this.maxSize);
prefetchCounters = new PrefetchCounters();
this.executorService = executorService;
}
@ -60,8 +60,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
GetRecordsResult result = null;
try {
result = getRecordsResultQueue.take();
updateBytes(result, false);
updateRecordsCount(result, false);
prefetchCounters.removed(result);
} catch (InterruptedException e) {
log.error("Interrupted while getting records from the cache", e);
}
@ -73,38 +72,15 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
executorService.shutdown();
}
private void updateBytes(final GetRecordsResult getRecordsResult, final boolean add) {
getRecordsResult.getRecords().forEach(record -> {
int newLength = record.getData().array().length;
if (add) {
currentSizeInBytes += newLength;
} else {
currentSizeInBytes -= newLength;
}
});
}
private void updateRecordsCount(final GetRecordsResult getRecordsResult, final boolean add) {
int newSize = getRecordsResult.getRecords().size();
if (add) {
currentRecordsCount += newSize;
} else {
currentRecordsCount -= newSize;
}
}
private class DefaultGetRecordsCacheDaemon implements Runnable {
@Override
public void run() {
while (true) {
if (currentSizeInBytes < maxByteSize && currentRecordsCount < maxRecordsCount) {
if (prefetchCounters.byteSize < maxByteSize && prefetchCounters.size < maxRecordsCount) {
try {
GetRecordsResult getRecordsResult = getRecordsRetrievalStrategy.getRecords(maxRecordsPerCall);
getRecordsResultQueue.put(getRecordsResult);
if (getRecordsResultQueue.contains(getRecordsResult)) {
updateBytes(getRecordsResult, true);
updateRecordsCount(getRecordsResult, true);
}
prefetchCounters.added(getRecordsResult);
} catch (InterruptedException e) {
log.error("Interrupted while adding records to the cache", e);
}
@ -113,4 +89,27 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
}
}
private class PrefetchCounters {
private volatile long size = 0;
private volatile long byteSize = 0;
public void added(final GetRecordsResult result) {
size += getSize(result);
byteSize += getByteSize(result);
}
public void removed(final GetRecordsResult result) {
size -= getSize(result);
byteSize -= getByteSize(result);
}
private long getSize(final GetRecordsResult result) {
return result.getRecords().size();
}
private long getByteSize(final GetRecordsResult result) {
return result.getRecords().stream().mapToLong(record -> record.getData().array().length).sum();
}
}
}