@Test public void testIfShardSyncIsNotInitiatedWhenConfidenceFactorIsNotReached() { List<Lease> multiStreamLeases = new ArrayList<HashKeyRangeForLease>() {{ add(deserialize(MIN_HASH_KEY.toString(), "1")); add(deserialize("2", "3")); add(deserialize("4", "23")); add(deserialize("6", "23")); add(deserialize("25", MAX_HASH_KEY.toString())); }}.stream().map(hashKeyRangeForLease -> { MultiStreamLease lease = new MultiStreamLease(); lease.hashKeyRange(hashKeyRangeForLease); lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); return lease; }).collect(Collectors.toList()); IntStream.range(1, DEFAULT_CONSECUTIVE_HOLES_FOR_TRIGGERING_LEASE_RECOVERY).forEach(i -> Assert .assertFalse(periodicShardSyncManager.checkForShardSync(streamIdentifier, multiStreamLeases).shouldDoShardSync())); }
@VisibleForTesting ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Lease> leases) { if (CollectionUtils.isNullOrEmpty(leases)) { log.info("No leases found for {}. Will be triggering shard sync", streamIdentifier); return new ShardSyncResponse(true, false, "No leases found for " + streamIdentifier); } Optional<HashRangeHole> hashRangeHoleOpt = hasHoleInLeases(streamIdentifier, leases); if (hashRangeHoleOpt.isPresent()) { final HashRangeHoleTracker hashRangeHoleTracker = hashRangeHoleTrackerMap .computeIfAbsent(streamIdentifier, s -> new HashRangeHoleTracker()); final boolean hasHoleWithHighConfidence = hashRangeHoleTracker .hasHighConfidenceOfHoleWith(hashRangeHoleOpt.get()); return new ShardSyncResponse(hasHoleWithHighConfidence, true, "Detected same hole for " + hashRangeHoleTracker.getNumConsecutiveHoles() + " times. Shard sync will be initiated when threshold reaches " + leasesRecoveryAuditorInconsistencyConfidenceThreshold); } else { hashRangeHoleTrackerMap.remove(streamIdentifier); return new ShardSyncResponse(false, false, "Hash Ranges are complete for " + streamIdentifier); } }
PeriodicShardSyncManager { @VisibleForTesting ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Lease> leases) { if (CollectionUtils.isNullOrEmpty(leases)) { log.info("No leases found for {}. Will be triggering shard sync", streamIdentifier); return new ShardSyncResponse(true, false, "No leases found for " + streamIdentifier); } Optional<HashRangeHole> hashRangeHoleOpt = hasHoleInLeases(streamIdentifier, leases); if (hashRangeHoleOpt.isPresent()) { final HashRangeHoleTracker hashRangeHoleTracker = hashRangeHoleTrackerMap .computeIfAbsent(streamIdentifier, s -> new HashRangeHoleTracker()); final boolean hasHoleWithHighConfidence = hashRangeHoleTracker .hasHighConfidenceOfHoleWith(hashRangeHoleOpt.get()); return new ShardSyncResponse(hasHoleWithHighConfidence, true, "Detected same hole for " + hashRangeHoleTracker.getNumConsecutiveHoles() + " times. Shard sync will be initiated when threshold reaches " + leasesRecoveryAuditorInconsistencyConfidenceThreshold); } else { hashRangeHoleTrackerMap.remove(streamIdentifier); return new ShardSyncResponse(false, false, "Hash Ranges are complete for " + streamIdentifier); } } }
PeriodicShardSyncManager { @VisibleForTesting ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Lease> leases) { if (CollectionUtils.isNullOrEmpty(leases)) { log.info("No leases found for {}. Will be triggering shard sync", streamIdentifier); return new ShardSyncResponse(true, false, "No leases found for " + streamIdentifier); } Optional<HashRangeHole> hashRangeHoleOpt = hasHoleInLeases(streamIdentifier, leases); if (hashRangeHoleOpt.isPresent()) { final HashRangeHoleTracker hashRangeHoleTracker = hashRangeHoleTrackerMap .computeIfAbsent(streamIdentifier, s -> new HashRangeHoleTracker()); final boolean hasHoleWithHighConfidence = hashRangeHoleTracker .hasHighConfidenceOfHoleWith(hashRangeHoleOpt.get()); return new ShardSyncResponse(hasHoleWithHighConfidence, true, "Detected same hole for " + hashRangeHoleTracker.getNumConsecutiveHoles() + " times. Shard sync will be initiated when threshold reaches " + leasesRecoveryAuditorInconsistencyConfidenceThreshold); } else { hashRangeHoleTrackerMap.remove(streamIdentifier); return new ShardSyncResponse(false, false, "Hash Ranges are complete for " + streamIdentifier); } } PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, LeaseRefresher leaseRefresher,
            Map<StreamIdentifier, StreamConfig> currentStreamConfigMap,
            Function<StreamConfig, ShardSyncTaskManager> shardSyncTaskManagerProvider, boolean isMultiStreamingMode,
            MetricsFactory metricsFactory,
            long leasesRecoveryAuditorExecutionFrequencyMillis,
            int leasesRecoveryAuditorInconsistencyConfidenceThreshold);   PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, LeaseRefresher leaseRefresher,
            Map<StreamIdentifier, StreamConfig> currentStreamConfigMap,
            Function<StreamConfig, ShardSyncTaskManager> shardSyncTaskManagerProvider,
            ScheduledExecutorService shardSyncThreadPool, boolean isMultiStreamingMode,
            MetricsFactory metricsFactory,
            long leasesRecoveryAuditorExecutionFrequencyMillis,
            int leasesRecoveryAuditorInconsistencyConfidenceThreshold); }
PeriodicShardSyncManager { @VisibleForTesting ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Lease> leases) { if (CollectionUtils.isNullOrEmpty(leases)) { log.info("No leases found for {}. Will be triggering shard sync", streamIdentifier); return new ShardSyncResponse(true, false, "No leases found for " + streamIdentifier); } Optional<HashRangeHole> hashRangeHoleOpt = hasHoleInLeases(streamIdentifier, leases); if (hashRangeHoleOpt.isPresent()) { final HashRangeHoleTracker hashRangeHoleTracker = hashRangeHoleTrackerMap .computeIfAbsent(streamIdentifier, s -> new HashRangeHoleTracker()); final boolean hasHoleWithHighConfidence = hashRangeHoleTracker .hasHighConfidenceOfHoleWith(hashRangeHoleOpt.get()); return new ShardSyncResponse(hasHoleWithHighConfidence, true, "Detected same hole for " + hashRangeHoleTracker.getNumConsecutiveHoles() + " times. Shard sync will be initiated when threshold reaches " + leasesRecoveryAuditorInconsistencyConfidenceThreshold); } else { hashRangeHoleTrackerMap.remove(streamIdentifier); return new ShardSyncResponse(false, false, "Hash Ranges are complete for " + streamIdentifier); } } PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, LeaseRefresher leaseRefresher,
            Map<StreamIdentifier, StreamConfig> currentStreamConfigMap,
            Function<StreamConfig, ShardSyncTaskManager> shardSyncTaskManagerProvider, boolean isMultiStreamingMode,
            MetricsFactory metricsFactory,
            long leasesRecoveryAuditorExecutionFrequencyMillis,
            int leasesRecoveryAuditorInconsistencyConfidenceThreshold);   PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, LeaseRefresher leaseRefresher,
            Map<StreamIdentifier, StreamConfig> currentStreamConfigMap,
            Function<StreamConfig, ShardSyncTaskManager> shardSyncTaskManagerProvider,
            ScheduledExecutorService shardSyncThreadPool, boolean isMultiStreamingMode,
            MetricsFactory metricsFactory,
            long leasesRecoveryAuditorExecutionFrequencyMillis,
            int leasesRecoveryAuditorInconsistencyConfidenceThreshold); synchronized TaskResult start(); synchronized void syncShardsOnce(); void stop(); }
PeriodicShardSyncManager { @VisibleForTesting ShardSyncResponse checkForShardSync(StreamIdentifier streamIdentifier, List<Lease> leases) { if (CollectionUtils.isNullOrEmpty(leases)) { log.info("No leases found for {}. Will be triggering shard sync", streamIdentifier); return new ShardSyncResponse(true, false, "No leases found for " + streamIdentifier); } Optional<HashRangeHole> hashRangeHoleOpt = hasHoleInLeases(streamIdentifier, leases); if (hashRangeHoleOpt.isPresent()) { final HashRangeHoleTracker hashRangeHoleTracker = hashRangeHoleTrackerMap .computeIfAbsent(streamIdentifier, s -> new HashRangeHoleTracker()); final boolean hasHoleWithHighConfidence = hashRangeHoleTracker .hasHighConfidenceOfHoleWith(hashRangeHoleOpt.get()); return new ShardSyncResponse(hasHoleWithHighConfidence, true, "Detected same hole for " + hashRangeHoleTracker.getNumConsecutiveHoles() + " times. Shard sync will be initiated when threshold reaches " + leasesRecoveryAuditorInconsistencyConfidenceThreshold); } else { hashRangeHoleTrackerMap.remove(streamIdentifier); return new ShardSyncResponse(false, false, "Hash Ranges are complete for " + streamIdentifier); } } PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, LeaseRefresher leaseRefresher,
            Map<StreamIdentifier, StreamConfig> currentStreamConfigMap,
            Function<StreamConfig, ShardSyncTaskManager> shardSyncTaskManagerProvider, boolean isMultiStreamingMode,
            MetricsFactory metricsFactory,
            long leasesRecoveryAuditorExecutionFrequencyMillis,
            int leasesRecoveryAuditorInconsistencyConfidenceThreshold);   PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, LeaseRefresher leaseRefresher,
            Map<StreamIdentifier, StreamConfig> currentStreamConfigMap,
            Function<StreamConfig, ShardSyncTaskManager> shardSyncTaskManagerProvider,
            ScheduledExecutorService shardSyncThreadPool, boolean isMultiStreamingMode,
            MetricsFactory metricsFactory,
            long leasesRecoveryAuditorExecutionFrequencyMillis,
            int leasesRecoveryAuditorInconsistencyConfidenceThreshold); synchronized TaskResult start(); synchronized void syncShardsOnce(); void stop();  }