@Test public void testForSuccessWhenHashRangesAreComplete() { List<Lease> hashRanges = new ArrayList<HashKeyRangeForLease>() {{ add(deserialize("0", "1")); add(deserialize("2", "3")); add(deserialize("4", "23")); add(deserialize("6", "23")); add(deserialize("24", MAX_HASH_KEY.toString())); }}.stream().map(hashKeyRangeForLease -> { Lease lease = new MultiStreamLease(); lease.hashKeyRange(hashKeyRangeForLease); lease.checkpoint(ExtendedSequenceNumber.TRIM_HORIZON); return lease; }).collect(Collectors.toList()); Assert.assertFalse(PeriodicShardSyncManager .checkForHoleInHashKeyRanges(streamIdentifier, hashRanges).isPresent()); }
@VisibleForTesting static Optional<HashRangeHole> checkForHoleInHashKeyRanges(StreamIdentifier streamIdentifier, List<Lease> leasesWithHashKeyRanges) { List<Lease> sortedLeasesWithHashKeyRanges = sortLeasesByHashRange(leasesWithHashKeyRanges); if(sortedLeasesWithHashKeyRanges.isEmpty()) { log.error("No leases with valid hashranges found for stream {}", streamIdentifier); return Optional.of(new HashRangeHole()); } if (!sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease().startingHashKey().equals(MIN_HASH_KEY) || !sortedLeasesWithHashKeyRanges .get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease().endingHashKey().equals(MAX_HASH_KEY)) { log.error("Incomplete hash range found for stream {} between {} and {}.", streamIdentifier, sortedLeasesWithHashKeyRanges.get(0), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1)); return Optional.of(new HashRangeHole(sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease())); } if (sortedLeasesWithHashKeyRanges.size() > 1) { Lease leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(0); HashKeyRangeForLease leftLeaseHashRange = leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(); for (int i = 1; i < sortedLeasesWithHashKeyRanges.size(); i++) { final HashKeyRangeForLease rightLeaseHashRange = sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease(); final BigInteger rangeDiff = rightLeaseHashRange.startingHashKey().subtract(leftLeaseHashRange.endingHashKey()); if (rangeDiff.signum() <= 0) { leftLeaseHashRange = new HashKeyRangeForLease(leftLeaseHashRange.startingHashKey(), leftLeaseHashRange.endingHashKey().max(rightLeaseHashRange.endingHashKey())); } else { if (!rangeDiff.equals(BigInteger.ONE)) { log.error("Incomplete hash range found for {} between {} and {}.", streamIdentifier, leftMostLeaseToReportInCaseOfHole, sortedLeasesWithHashKeyRanges.get(i)); return Optional.of(new HashRangeHole(leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease())); } leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(i); leftLeaseHashRange = rightLeaseHashRange; } } } return Optional.empty(); }
PeriodicShardSyncManager { @VisibleForTesting static Optional<HashRangeHole> checkForHoleInHashKeyRanges(StreamIdentifier streamIdentifier, List<Lease> leasesWithHashKeyRanges) { List<Lease> sortedLeasesWithHashKeyRanges = sortLeasesByHashRange(leasesWithHashKeyRanges); if(sortedLeasesWithHashKeyRanges.isEmpty()) { log.error("No leases with valid hashranges found for stream {}", streamIdentifier); return Optional.of(new HashRangeHole()); } if (!sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease().startingHashKey().equals(MIN_HASH_KEY) || !sortedLeasesWithHashKeyRanges .get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease().endingHashKey().equals(MAX_HASH_KEY)) { log.error("Incomplete hash range found for stream {} between {} and {}.", streamIdentifier, sortedLeasesWithHashKeyRanges.get(0), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1)); return Optional.of(new HashRangeHole(sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease())); } if (sortedLeasesWithHashKeyRanges.size() > 1) { Lease leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(0); HashKeyRangeForLease leftLeaseHashRange = leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(); for (int i = 1; i < sortedLeasesWithHashKeyRanges.size(); i++) { final HashKeyRangeForLease rightLeaseHashRange = sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease(); final BigInteger rangeDiff = rightLeaseHashRange.startingHashKey().subtract(leftLeaseHashRange.endingHashKey()); if (rangeDiff.signum() <= 0) { leftLeaseHashRange = new HashKeyRangeForLease(leftLeaseHashRange.startingHashKey(), leftLeaseHashRange.endingHashKey().max(rightLeaseHashRange.endingHashKey())); } else { if (!rangeDiff.equals(BigInteger.ONE)) { log.error("Incomplete hash range found for {} between {} and {}.", streamIdentifier, leftMostLeaseToReportInCaseOfHole, sortedLeasesWithHashKeyRanges.get(i)); return Optional.of(new HashRangeHole(leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease())); } leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(i); leftLeaseHashRange = rightLeaseHashRange; } } } return Optional.empty(); } }
PeriodicShardSyncManager { @VisibleForTesting static Optional<HashRangeHole> checkForHoleInHashKeyRanges(StreamIdentifier streamIdentifier, List<Lease> leasesWithHashKeyRanges) { List<Lease> sortedLeasesWithHashKeyRanges = sortLeasesByHashRange(leasesWithHashKeyRanges); if(sortedLeasesWithHashKeyRanges.isEmpty()) { log.error("No leases with valid hashranges found for stream {}", streamIdentifier); return Optional.of(new HashRangeHole()); } if (!sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease().startingHashKey().equals(MIN_HASH_KEY) || !sortedLeasesWithHashKeyRanges .get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease().endingHashKey().equals(MAX_HASH_KEY)) { log.error("Incomplete hash range found for stream {} between {} and {}.", streamIdentifier, sortedLeasesWithHashKeyRanges.get(0), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1)); return Optional.of(new HashRangeHole(sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease())); } if (sortedLeasesWithHashKeyRanges.size() > 1) { Lease leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(0); HashKeyRangeForLease leftLeaseHashRange = leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(); for (int i = 1; i < sortedLeasesWithHashKeyRanges.size(); i++) { final HashKeyRangeForLease rightLeaseHashRange = sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease(); final BigInteger rangeDiff = rightLeaseHashRange.startingHashKey().subtract(leftLeaseHashRange.endingHashKey()); if (rangeDiff.signum() <= 0) { leftLeaseHashRange = new HashKeyRangeForLease(leftLeaseHashRange.startingHashKey(), leftLeaseHashRange.endingHashKey().max(rightLeaseHashRange.endingHashKey())); } else { if (!rangeDiff.equals(BigInteger.ONE)) { log.error("Incomplete hash range found for {} between {} and {}.", streamIdentifier, leftMostLeaseToReportInCaseOfHole, sortedLeasesWithHashKeyRanges.get(i)); return Optional.of(new HashRangeHole(leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease())); } leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(i); leftLeaseHashRange = rightLeaseHashRange; } } } return Optional.empty(); } 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 static Optional<HashRangeHole> checkForHoleInHashKeyRanges(StreamIdentifier streamIdentifier, List<Lease> leasesWithHashKeyRanges) { List<Lease> sortedLeasesWithHashKeyRanges = sortLeasesByHashRange(leasesWithHashKeyRanges); if(sortedLeasesWithHashKeyRanges.isEmpty()) { log.error("No leases with valid hashranges found for stream {}", streamIdentifier); return Optional.of(new HashRangeHole()); } if (!sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease().startingHashKey().equals(MIN_HASH_KEY) || !sortedLeasesWithHashKeyRanges .get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease().endingHashKey().equals(MAX_HASH_KEY)) { log.error("Incomplete hash range found for stream {} between {} and {}.", streamIdentifier, sortedLeasesWithHashKeyRanges.get(0), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1)); return Optional.of(new HashRangeHole(sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease())); } if (sortedLeasesWithHashKeyRanges.size() > 1) { Lease leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(0); HashKeyRangeForLease leftLeaseHashRange = leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(); for (int i = 1; i < sortedLeasesWithHashKeyRanges.size(); i++) { final HashKeyRangeForLease rightLeaseHashRange = sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease(); final BigInteger rangeDiff = rightLeaseHashRange.startingHashKey().subtract(leftLeaseHashRange.endingHashKey()); if (rangeDiff.signum() <= 0) { leftLeaseHashRange = new HashKeyRangeForLease(leftLeaseHashRange.startingHashKey(), leftLeaseHashRange.endingHashKey().max(rightLeaseHashRange.endingHashKey())); } else { if (!rangeDiff.equals(BigInteger.ONE)) { log.error("Incomplete hash range found for {} between {} and {}.", streamIdentifier, leftMostLeaseToReportInCaseOfHole, sortedLeasesWithHashKeyRanges.get(i)); return Optional.of(new HashRangeHole(leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease())); } leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(i); leftLeaseHashRange = rightLeaseHashRange; } } } return Optional.empty(); } 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 static Optional<HashRangeHole> checkForHoleInHashKeyRanges(StreamIdentifier streamIdentifier, List<Lease> leasesWithHashKeyRanges) { List<Lease> sortedLeasesWithHashKeyRanges = sortLeasesByHashRange(leasesWithHashKeyRanges); if(sortedLeasesWithHashKeyRanges.isEmpty()) { log.error("No leases with valid hashranges found for stream {}", streamIdentifier); return Optional.of(new HashRangeHole()); } if (!sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease().startingHashKey().equals(MIN_HASH_KEY) || !sortedLeasesWithHashKeyRanges .get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease().endingHashKey().equals(MAX_HASH_KEY)) { log.error("Incomplete hash range found for stream {} between {} and {}.", streamIdentifier, sortedLeasesWithHashKeyRanges.get(0), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1)); return Optional.of(new HashRangeHole(sortedLeasesWithHashKeyRanges.get(0).hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(sortedLeasesWithHashKeyRanges.size() - 1).hashKeyRangeForLease())); } if (sortedLeasesWithHashKeyRanges.size() > 1) { Lease leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(0); HashKeyRangeForLease leftLeaseHashRange = leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(); for (int i = 1; i < sortedLeasesWithHashKeyRanges.size(); i++) { final HashKeyRangeForLease rightLeaseHashRange = sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease(); final BigInteger rangeDiff = rightLeaseHashRange.startingHashKey().subtract(leftLeaseHashRange.endingHashKey()); if (rangeDiff.signum() <= 0) { leftLeaseHashRange = new HashKeyRangeForLease(leftLeaseHashRange.startingHashKey(), leftLeaseHashRange.endingHashKey().max(rightLeaseHashRange.endingHashKey())); } else { if (!rangeDiff.equals(BigInteger.ONE)) { log.error("Incomplete hash range found for {} between {} and {}.", streamIdentifier, leftMostLeaseToReportInCaseOfHole, sortedLeasesWithHashKeyRanges.get(i)); return Optional.of(new HashRangeHole(leftMostLeaseToReportInCaseOfHole.hashKeyRangeForLease(), sortedLeasesWithHashKeyRanges.get(i).hashKeyRangeForLease())); } leftMostLeaseToReportInCaseOfHole = sortedLeasesWithHashKeyRanges.get(i); leftLeaseHashRange = rightLeaseHashRange; } } } return Optional.empty(); } 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();  }