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

Only making leader to do the stream sync

Browse source
This commit is contained in:
Ashwin Giridharan 2020-06-22 14:30:28 -07:00
parent db2c22e046
commit 0246e1e852
2 changed files with 9 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java
View file

@ -423,7 +423,9 @@ public class Scheduler implements Runnable {
cleanupShardConsumers(assignedShards);
// check for new streams and sync with the scheduler state
checkAndSyncStreamShardsAndLeases();
if (isLeader()) {
checkAndSyncStreamShardsAndLeases();
}
logExecutorState();
slog.info("Sleeping ...");
@ -440,6 +442,10 @@ public class Scheduler implements Runnable {
slog.resetInfoLogging();
}
private boolean isLeader() {
return leaderDecider.isLeader(leaseManagementConfig.workerIdentifier());
}
/**
* Note: This method has package level access solely for testing purposes.

4
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownTask.java
View file

@ -113,8 +113,8 @@ public class ShutdownTask implements ConsumerTask {
try {
try {
log.debug("Invoking shutdown() for shard {}, concurrencyToken {}. Shutdown reason: {}",
leaseKeyProvider.apply(shardInfo), shardInfo.concurrencyToken(), reason);
log.debug("Invoking shutdown() for shard {} with child shards {} , concurrencyToken {}. Shutdown reason: {}",
leaseKeyProvider.apply(shardInfo), childShards, shardInfo.concurrencyToken(), reason);
final long startTime = System.currentTimeMillis();
if (reason == ShutdownReason.SHARD_END) {