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

Copy streamConfigMap to avoid data race with Scheduler

Browse source
This commit is contained in:
Chenyuan Lee 2023-04-09 23:32:37 -07:00
parent 88246e717e
commit 4cc1754e16
1 changed files with 10 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/PeriodicShardSyncManager.java
View file

@ -178,12 +178,20 @@ class PeriodicShardSyncManager {
final long runStartMillis = System.currentTimeMillis(); final long runStartMillis = System.currentTimeMillis();
try { try {
// Create a shallow copy of the map to avoid data race with Scheduler.
final Map<StreamIdentifier, StreamConfig> streamConfigMap = currentStreamConfigMap.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
// Construct the stream to leases map to be used in the lease sync // Construct the stream to leases map to be used in the lease sync
final Map<StreamIdentifier, List<Lease>> streamToLeasesMap = getStreamToLeasesMap( final Map<StreamIdentifier, List<Lease>> streamToLeasesMap = getStreamToLeasesMap(
currentStreamConfigMap.keySet()); streamConfigMap.keySet());
// For each of the stream, check if shard sync needs to be done based on the leases state. // For each of the stream, check if shard sync needs to be done based on the leases state.
for (Map.Entry<StreamIdentifier, StreamConfig> streamConfigEntry : currentStreamConfigMap.entrySet()) { for (Map.Entry<StreamIdentifier, StreamConfig> streamConfigEntry : streamConfigMap.entrySet()) {
// Skip shardSync if CSCM has purged the stream.
if (!currentStreamConfigMap.containsKey(streamConfigEntry.getKey())) {
continue;
}
final ShardSyncResponse shardSyncResponse = checkForShardSync(streamConfigEntry.getKey(), final ShardSyncResponse shardSyncResponse = checkForShardSync(streamConfigEntry.getKey(),
streamToLeasesMap.get(streamConfigEntry.getKey())); streamToLeasesMap.get(streamConfigEntry.getKey()));