@Test public void checkpointEveryTenPartitions() { RangeComparator comparator = comparator(context(0L, 100L)); comparator.compare(keys(LongStream.range(0, 25).toArray()), keys(LongStream.range(0, 25).toArray()), this::alwaysMatch); assertReported(9, RangeStats.withValues(10, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L), journal); assertReported(19, RangeStats.withValues(20, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L), journal); assertEquals(2, journal.keySet().size()); }
public RangeStats compare(Iterator<PartitionKey> sourceKeys, Iterator<PartitionKey> targetKeys, Function<PartitionKey, PartitionComparator> partitionTaskProvider) { final RangeStats rangeStats = RangeStats.newStats(); if (context.startToken.equals(context.endToken)) return rangeStats; Phaser phaser = new Phaser(1); AtomicLong partitionCount = new AtomicLong(0); AtomicReference<BigInteger> highestTokenSeen = new AtomicReference<>(context.startToken); logger.info("Comparing range [{},{}]", context.startToken, context.endToken); try { PartitionKey sourceKey = nextKey(sourceKeys); PartitionKey targetKey = nextKey(targetKeys); if ((sourceKey == null) != (targetKey == null)) { if (sourceKey == null) { logger.info("First in range, source iter is empty {}", context); onlyInTarget(rangeStats, targetKey); targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key)); } else { logger.info("First in range, target iter is empty {}", context); onlyInSource(rangeStats, sourceKey); sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key)); } return rangeStats; } while (sourceKey != null && targetKey != null) { int ret = sourceKey.compareTo(targetKey); if (ret > 0) { onlyInTarget(rangeStats, targetKey); targetKey = nextKey(targetKeys); } else if (ret < 0) { onlyInSource(rangeStats, sourceKey); sourceKey = nextKey(sourceKeys); } else { Verify.verify(sourceKey.equals(targetKey), "Can only compare partitions with identical keys: (%s, %s)", sourceKey, targetKey); if (!context.isTokenAllowed(sourceKey.getTokenAsBigInteger())) { logger.debug("Skipping disallowed token {}", sourceKey.getTokenAsBigInteger()); rangeStats.skipPartition(); sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); continue; } BigInteger token = sourceKey.getTokenAsBigInteger(); try { PartitionComparator comparisonTask = partitionTaskProvider.apply(sourceKey); comparisonExecutor.submit(comparisonTask, onSuccess(rangeStats, partitionCount, token, highestTokenSeen, mismatchReporter, journal), onError(rangeStats, token, errorReporter), phaser); } catch (Throwable t) { recordError(rangeStats, token, errorReporter, t); } finally { sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); } } } if (sourceKey != null) onlyInSource(rangeStats, sourceKey); else if (targetKey != null) onlyInTarget(rangeStats, targetKey); drain(sourceKeys, targetKeys, rangeStats); } catch (Exception e) { logger.debug("Waiting for {} in flight tasks before propagating error", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); throw new RuntimeException(String.format("Error encountered during range comparison for [%s:%s]", context.startToken, context.endToken), e); } logger.debug("Waiting for {} in flight tasks before returning", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); if (!rangeStats.allMatches()) logger.info("Segment [{}:{}] stats - ({})", context.startToken, context.endToken, rangeStats); return rangeStats; }
RangeComparator { public RangeStats compare(Iterator<PartitionKey> sourceKeys, Iterator<PartitionKey> targetKeys, Function<PartitionKey, PartitionComparator> partitionTaskProvider) { final RangeStats rangeStats = RangeStats.newStats(); if (context.startToken.equals(context.endToken)) return rangeStats; Phaser phaser = new Phaser(1); AtomicLong partitionCount = new AtomicLong(0); AtomicReference<BigInteger> highestTokenSeen = new AtomicReference<>(context.startToken); logger.info("Comparing range [{},{}]", context.startToken, context.endToken); try { PartitionKey sourceKey = nextKey(sourceKeys); PartitionKey targetKey = nextKey(targetKeys); if ((sourceKey == null) != (targetKey == null)) { if (sourceKey == null) { logger.info("First in range, source iter is empty {}", context); onlyInTarget(rangeStats, targetKey); targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key)); } else { logger.info("First in range, target iter is empty {}", context); onlyInSource(rangeStats, sourceKey); sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key)); } return rangeStats; } while (sourceKey != null && targetKey != null) { int ret = sourceKey.compareTo(targetKey); if (ret > 0) { onlyInTarget(rangeStats, targetKey); targetKey = nextKey(targetKeys); } else if (ret < 0) { onlyInSource(rangeStats, sourceKey); sourceKey = nextKey(sourceKeys); } else { Verify.verify(sourceKey.equals(targetKey), "Can only compare partitions with identical keys: (%s, %s)", sourceKey, targetKey); if (!context.isTokenAllowed(sourceKey.getTokenAsBigInteger())) { logger.debug("Skipping disallowed token {}", sourceKey.getTokenAsBigInteger()); rangeStats.skipPartition(); sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); continue; } BigInteger token = sourceKey.getTokenAsBigInteger(); try { PartitionComparator comparisonTask = partitionTaskProvider.apply(sourceKey); comparisonExecutor.submit(comparisonTask, onSuccess(rangeStats, partitionCount, token, highestTokenSeen, mismatchReporter, journal), onError(rangeStats, token, errorReporter), phaser); } catch (Throwable t) { recordError(rangeStats, token, errorReporter, t); } finally { sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); } } } if (sourceKey != null) onlyInSource(rangeStats, sourceKey); else if (targetKey != null) onlyInTarget(rangeStats, targetKey); drain(sourceKeys, targetKeys, rangeStats); } catch (Exception e) { logger.debug("Waiting for {} in flight tasks before propagating error", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); throw new RuntimeException(String.format("Error encountered during range comparison for [%s:%s]", context.startToken, context.endToken), e); } logger.debug("Waiting for {} in flight tasks before returning", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); if (!rangeStats.allMatches()) logger.info("Segment [{}:{}] stats - ({})", context.startToken, context.endToken, rangeStats); return rangeStats; } }
RangeComparator { public RangeStats compare(Iterator<PartitionKey> sourceKeys, Iterator<PartitionKey> targetKeys, Function<PartitionKey, PartitionComparator> partitionTaskProvider) { final RangeStats rangeStats = RangeStats.newStats(); if (context.startToken.equals(context.endToken)) return rangeStats; Phaser phaser = new Phaser(1); AtomicLong partitionCount = new AtomicLong(0); AtomicReference<BigInteger> highestTokenSeen = new AtomicReference<>(context.startToken); logger.info("Comparing range [{},{}]", context.startToken, context.endToken); try { PartitionKey sourceKey = nextKey(sourceKeys); PartitionKey targetKey = nextKey(targetKeys); if ((sourceKey == null) != (targetKey == null)) { if (sourceKey == null) { logger.info("First in range, source iter is empty {}", context); onlyInTarget(rangeStats, targetKey); targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key)); } else { logger.info("First in range, target iter is empty {}", context); onlyInSource(rangeStats, sourceKey); sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key)); } return rangeStats; } while (sourceKey != null && targetKey != null) { int ret = sourceKey.compareTo(targetKey); if (ret > 0) { onlyInTarget(rangeStats, targetKey); targetKey = nextKey(targetKeys); } else if (ret < 0) { onlyInSource(rangeStats, sourceKey); sourceKey = nextKey(sourceKeys); } else { Verify.verify(sourceKey.equals(targetKey), "Can only compare partitions with identical keys: (%s, %s)", sourceKey, targetKey); if (!context.isTokenAllowed(sourceKey.getTokenAsBigInteger())) { logger.debug("Skipping disallowed token {}", sourceKey.getTokenAsBigInteger()); rangeStats.skipPartition(); sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); continue; } BigInteger token = sourceKey.getTokenAsBigInteger(); try { PartitionComparator comparisonTask = partitionTaskProvider.apply(sourceKey); comparisonExecutor.submit(comparisonTask, onSuccess(rangeStats, partitionCount, token, highestTokenSeen, mismatchReporter, journal), onError(rangeStats, token, errorReporter), phaser); } catch (Throwable t) { recordError(rangeStats, token, errorReporter, t); } finally { sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); } } } if (sourceKey != null) onlyInSource(rangeStats, sourceKey); else if (targetKey != null) onlyInTarget(rangeStats, targetKey); drain(sourceKeys, targetKeys, rangeStats); } catch (Exception e) { logger.debug("Waiting for {} in flight tasks before propagating error", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); throw new RuntimeException(String.format("Error encountered during range comparison for [%s:%s]", context.startToken, context.endToken), e); } logger.debug("Waiting for {} in flight tasks before returning", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); if (!rangeStats.allMatches()) logger.info("Segment [{}:{}] stats - ({})", context.startToken, context.endToken, rangeStats); return rangeStats; } RangeComparator(DiffContext context,
                           BiConsumer<Throwable, BigInteger> errorReporter,
                           BiConsumer<MismatchType,BigInteger> mismatchReporter,
                           BiConsumer<RangeStats, BigInteger> journal,
                           ComparisonExecutor comparisonExecutor); }
RangeComparator { public RangeStats compare(Iterator<PartitionKey> sourceKeys, Iterator<PartitionKey> targetKeys, Function<PartitionKey, PartitionComparator> partitionTaskProvider) { final RangeStats rangeStats = RangeStats.newStats(); if (context.startToken.equals(context.endToken)) return rangeStats; Phaser phaser = new Phaser(1); AtomicLong partitionCount = new AtomicLong(0); AtomicReference<BigInteger> highestTokenSeen = new AtomicReference<>(context.startToken); logger.info("Comparing range [{},{}]", context.startToken, context.endToken); try { PartitionKey sourceKey = nextKey(sourceKeys); PartitionKey targetKey = nextKey(targetKeys); if ((sourceKey == null) != (targetKey == null)) { if (sourceKey == null) { logger.info("First in range, source iter is empty {}", context); onlyInTarget(rangeStats, targetKey); targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key)); } else { logger.info("First in range, target iter is empty {}", context); onlyInSource(rangeStats, sourceKey); sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key)); } return rangeStats; } while (sourceKey != null && targetKey != null) { int ret = sourceKey.compareTo(targetKey); if (ret > 0) { onlyInTarget(rangeStats, targetKey); targetKey = nextKey(targetKeys); } else if (ret < 0) { onlyInSource(rangeStats, sourceKey); sourceKey = nextKey(sourceKeys); } else { Verify.verify(sourceKey.equals(targetKey), "Can only compare partitions with identical keys: (%s, %s)", sourceKey, targetKey); if (!context.isTokenAllowed(sourceKey.getTokenAsBigInteger())) { logger.debug("Skipping disallowed token {}", sourceKey.getTokenAsBigInteger()); rangeStats.skipPartition(); sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); continue; } BigInteger token = sourceKey.getTokenAsBigInteger(); try { PartitionComparator comparisonTask = partitionTaskProvider.apply(sourceKey); comparisonExecutor.submit(comparisonTask, onSuccess(rangeStats, partitionCount, token, highestTokenSeen, mismatchReporter, journal), onError(rangeStats, token, errorReporter), phaser); } catch (Throwable t) { recordError(rangeStats, token, errorReporter, t); } finally { sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); } } } if (sourceKey != null) onlyInSource(rangeStats, sourceKey); else if (targetKey != null) onlyInTarget(rangeStats, targetKey); drain(sourceKeys, targetKeys, rangeStats); } catch (Exception e) { logger.debug("Waiting for {} in flight tasks before propagating error", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); throw new RuntimeException(String.format("Error encountered during range comparison for [%s:%s]", context.startToken, context.endToken), e); } logger.debug("Waiting for {} in flight tasks before returning", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); if (!rangeStats.allMatches()) logger.info("Segment [{}:{}] stats - ({})", context.startToken, context.endToken, rangeStats); return rangeStats; } RangeComparator(DiffContext context,
                           BiConsumer<Throwable, BigInteger> errorReporter,
                           BiConsumer<MismatchType,BigInteger> mismatchReporter,
                           BiConsumer<RangeStats, BigInteger> journal,
                           ComparisonExecutor comparisonExecutor); RangeStats compare(Iterator<PartitionKey> sourceKeys,
                              Iterator<PartitionKey> targetKeys,
                              Function<PartitionKey, PartitionComparator> partitionTaskProvider); }
RangeComparator { public RangeStats compare(Iterator<PartitionKey> sourceKeys, Iterator<PartitionKey> targetKeys, Function<PartitionKey, PartitionComparator> partitionTaskProvider) { final RangeStats rangeStats = RangeStats.newStats(); if (context.startToken.equals(context.endToken)) return rangeStats; Phaser phaser = new Phaser(1); AtomicLong partitionCount = new AtomicLong(0); AtomicReference<BigInteger> highestTokenSeen = new AtomicReference<>(context.startToken); logger.info("Comparing range [{},{}]", context.startToken, context.endToken); try { PartitionKey sourceKey = nextKey(sourceKeys); PartitionKey targetKey = nextKey(targetKeys); if ((sourceKey == null) != (targetKey == null)) { if (sourceKey == null) { logger.info("First in range, source iter is empty {}", context); onlyInTarget(rangeStats, targetKey); targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key)); } else { logger.info("First in range, target iter is empty {}", context); onlyInSource(rangeStats, sourceKey); sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key)); } return rangeStats; } while (sourceKey != null && targetKey != null) { int ret = sourceKey.compareTo(targetKey); if (ret > 0) { onlyInTarget(rangeStats, targetKey); targetKey = nextKey(targetKeys); } else if (ret < 0) { onlyInSource(rangeStats, sourceKey); sourceKey = nextKey(sourceKeys); } else { Verify.verify(sourceKey.equals(targetKey), "Can only compare partitions with identical keys: (%s, %s)", sourceKey, targetKey); if (!context.isTokenAllowed(sourceKey.getTokenAsBigInteger())) { logger.debug("Skipping disallowed token {}", sourceKey.getTokenAsBigInteger()); rangeStats.skipPartition(); sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); continue; } BigInteger token = sourceKey.getTokenAsBigInteger(); try { PartitionComparator comparisonTask = partitionTaskProvider.apply(sourceKey); comparisonExecutor.submit(comparisonTask, onSuccess(rangeStats, partitionCount, token, highestTokenSeen, mismatchReporter, journal), onError(rangeStats, token, errorReporter), phaser); } catch (Throwable t) { recordError(rangeStats, token, errorReporter, t); } finally { sourceKey = nextKey(sourceKeys); targetKey = nextKey(targetKeys); } } } if (sourceKey != null) onlyInSource(rangeStats, sourceKey); else if (targetKey != null) onlyInTarget(rangeStats, targetKey); drain(sourceKeys, targetKeys, rangeStats); } catch (Exception e) { logger.debug("Waiting for {} in flight tasks before propagating error", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); throw new RuntimeException(String.format("Error encountered during range comparison for [%s:%s]", context.startToken, context.endToken), e); } logger.debug("Waiting for {} in flight tasks before returning", phaser.getUnarrivedParties()); phaser.arriveAndAwaitAdvance(); if (!rangeStats.allMatches()) logger.info("Segment [{}:{}] stats - ({})", context.startToken, context.endToken, rangeStats); return rangeStats; } RangeComparator(DiffContext context,
                           BiConsumer<Throwable, BigInteger> errorReporter,
                           BiConsumer<MismatchType,BigInteger> mismatchReporter,
                           BiConsumer<RangeStats, BigInteger> journal,
                           ComparisonExecutor comparisonExecutor); RangeStats compare(Iterator<PartitionKey> sourceKeys,
                              Iterator<PartitionKey> targetKeys,
                              Function<PartitionKey, PartitionComparator> partitionTaskProvider);  }