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

Making sure to call shutdown on strategy once the cache is interrupted.

Browse source
This commit is contained in:
Sahil Palvia 2017-09-27 21:01:29 -07:00
parent 7b76d1d56e
commit cd1f944add
1 changed files with 9 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -115,6 +115,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
while (true) { while (true) {
if (Thread.currentThread().isInterrupted()) { if (Thread.currentThread().isInterrupted()) {
log.warn("Prefetch thread was interrupted."); log.warn("Prefetch thread was interrupted.");
callShutdownOnStrategy();
break; break;
} }
if (prefetchCounters.shouldGetNewRecords()) { if (prefetchCounters.shouldGetNewRecords()) {
@ -129,8 +130,8 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
getRecordsResultQueue.put(processRecordsInput); getRecordsResultQueue.put(processRecordsInput);
prefetchCounters.added(processRecordsInput); prefetchCounters.added(processRecordsInput);
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.info("Thread was interrupted, indicating shutdown was called on the cache. Calling shutdown on the GetRecordsRetrievalStrategy."); log.info("Thread was interrupted, indicating shutdown was called on the cache.");
getRecordsRetrievalStrategy.shutdown(); callShutdownOnStrategy();
} catch (Error e) { } catch (Error e) {
log.error("Error was thrown while getting records, please check for the error", e); log.error("Error was thrown while getting records, please check for the error", e);
} }
@ -138,6 +139,12 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
} }
} }
private void callShutdownOnStrategy() {
if (!getRecordsRetrievalStrategy.isShutdown()) {
getRecordsRetrievalStrategy.shutdown();
}
}
private void sleepBeforeNextCall() throws InterruptedException { private void sleepBeforeNextCall() throws InterruptedException {
if (lastSuccessfulCall == null) { if (lastSuccessfulCall == null) {
return; return;