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

Merge 7daae980c0 into 6fc148740d

Browse source
This commit is contained in:
Tom Coupland 2018-02-24 15:55:17 +00:00 committed by GitHub
commit a73bcffc9d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -103,6 +103,8 @@ public class Worker implements Runnable {
private volatile long shutdownStartTimeMillis; private volatile long shutdownStartTimeMillis;
private volatile boolean shutdownComplete = false; private volatile boolean shutdownComplete = false;
private volatile boolean initialized = false;
// Holds consumers for shards the worker is currently tracking. Key is shard // Holds consumers for shards the worker is currently tracking. Key is shard
// info, value is ShardConsumer. // info, value is ShardConsumer.
private ConcurrentMap<ShardInfo, ShardConsumer> shardInfoShardConsumerMap = new ConcurrentHashMap<ShardInfo, ShardConsumer>(); private ConcurrentMap<ShardInfo, ShardConsumer> shardInfoShardConsumerMap = new ConcurrentHashMap<ShardInfo, ShardConsumer>();
@ -543,6 +545,7 @@ public class Worker implements Runnable {
if (!isDone) { if (!isDone) {
throw new RuntimeException(lastException); throw new RuntimeException(lastException);
} }
initialized = true;
} }
/** /**
@ -751,6 +754,10 @@ public class Worker implements Runnable {
return shutdownComplete; return shutdownComplete;
} }
public boolean isInitialized() {
return initialized;
}
ConcurrentMap<ShardInfo, ShardConsumer> getShardInfoShardConsumerMap() { ConcurrentMap<ShardInfo, ShardConsumer> getShardInfoShardConsumerMap() {
return shardInfoShardConsumerMap; return shardInfoShardConsumerMap;
} }
@ -801,6 +808,7 @@ public class Worker implements Runnable {
if (metricsFactory instanceof WorkerCWMetricsFactory) { if (metricsFactory instanceof WorkerCWMetricsFactory) {
((CWMetricsFactory) metricsFactory).shutdown(); ((CWMetricsFactory) metricsFactory).shutdown();
} }
initialized = false;
shutdownComplete = true; shutdownComplete = true;
} }