test_class.declarator: errors = HashMultimap.create()
test_class.modifier: private
test_class.original_string: private Multimap<BigInteger, Throwable> errors = HashMultimap.create();
test_class.type: Multimap<BigInteger, Throwable>
test_class.var_name: errors
test_class.declarator: errorReporter = (e, t) -> errors.put(t, e)
test_class.modifier: private
test_class.original_string: private BiConsumer<Throwable, BigInteger> errorReporter = (e, t) -> errors.put(t, e);
test_class.type: BiConsumer<Throwable, BigInteger>
test_class.var_name: errorReporter
test_class.declarator: mismatches = HashMultimap.create()
test_class.modifier: private
test_class.original_string: private Multimap<BigInteger, MismatchType> mismatches = HashMultimap.create();
test_class.type: Multimap<BigInteger, MismatchType>
test_class.var_name: mismatches
test_class.declarator: mismatchReporter = (m, t) -> mismatches.put(t, m)
test_class.modifier: private
test_class.original_string: private BiConsumer<MismatchType, BigInteger> mismatchReporter = (m, t) -> mismatches.put(t, m);
test_class.type: BiConsumer<MismatchType, BigInteger>
test_class.var_name: mismatchReporter
test_class.declarator: journal= HashMultimap.create()
test_class.modifier: private
test_class.original_string: private Multimap<BigInteger, RangeStats> journal= HashMultimap.create();
test_class.type: Multimap<BigInteger, RangeStats>
test_class.var_name: journal
test_class.declarator: progressReporter = (r, t) -> journal.put(t, copyOf(r))
test_class.modifier: private
test_class.original_string: private BiConsumer<RangeStats, BigInteger> progressReporter = (r, t) -> journal.put(t, copyOf(r));
test_class.type: BiConsumer<RangeStats, BigInteger>
test_class.var_name: progressReporter
test_class.declarator: comparedPartitions = new HashSet<>()
test_class.modifier: private
test_class.original_string: private Set<BigInteger> comparedPartitions = new HashSet<>();
test_class.type: Set<BigInteger>
test_class.var_name: comparedPartitions
test_class.declarator: executor = ComparisonExecutor.newExecutor(1, new MetricRegistry())
test_class.modifier: private
test_class.original_string: private ComparisonExecutor executor = ComparisonExecutor.newExecutor(1, new MetricRegistry());
test_class.type: ComparisonExecutor
test_class.var_name: executor
test_class.declarator: mockRetryStrategyFactory = RetryStrategyProvider.create(null)
test_class.modifier: private
test_class.original_string: private RetryStrategyProvider mockRetryStrategyFactory = RetryStrategyProvider.create(null);
test_class.type: RetryStrategyProvider
test_class.var_name: mockRetryStrategyFactory
test_class.file: spark-job/src/test/java/org/apache/cassandra/diff/RangeComparatorTest.java
test_class.identifier: RangeComparatorTest
test_case.body: @Test
      public void targetKeysIsEmpty() {
          RangeComparator comparator = comparator(context(0L, 100L));
          RangeStats stats = comparator.compare(keys(4, 5), keys(), this::alwaysMatch);
          assertFalse(stats.isEmpty());
          assertEquals(2, stats.getOnlyInSource());
          assertEquals(0, stats.getOnlyInTarget());
          assertEquals(0, stats.getMatchedPartitions());
          assertReported(4, MismatchType.ONLY_IN_SOURCE, mismatches);
          assertReported(5, MismatchType.ONLY_IN_SOURCE, mismatches);
          assertNothingReported(errors, journal);
          assertCompared();
      }
test_case.class_method_signature: RangeComparatorTest.targetKeysIsEmpty()
test_case.full_signature: @Test public void targetKeysIsEmpty()
test_case.identifier: targetKeysIsEmpty
test_case.invocations: comparator context compare keys keys assertFalse isEmpty assertEquals getOnlyInSource assertEquals getOnlyInTarget assertEquals getMatchedPartitions assertReported assertReported assertNothingReported assertCompared
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void targetKeysIsEmpty()
focal_class.declarator: logger = LoggerFactory.getLogger(RangeComparator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger logger = LoggerFactory.getLogger(RangeComparator.class);
focal_class.type: Logger
focal_class.var_name: logger
focal_class.declarator: context
focal_class.modifier: private final
focal_class.original_string: private final DiffContext context;
focal_class.type: DiffContext
focal_class.var_name: context
focal_class.declarator: errorReporter
focal_class.modifier: private final
focal_class.original_string: private final BiConsumer<Throwable, BigInteger> errorReporter;
focal_class.type: BiConsumer<Throwable, BigInteger>
focal_class.var_name: errorReporter
focal_class.declarator: mismatchReporter
focal_class.modifier: private final
focal_class.original_string: private final BiConsumer<MismatchType, BigInteger> mismatchReporter;
focal_class.type: BiConsumer<MismatchType, BigInteger>
focal_class.var_name: mismatchReporter
focal_class.declarator: journal
focal_class.modifier: private final
focal_class.original_string: private final BiConsumer<RangeStats, BigInteger> journal;
focal_class.type: BiConsumer<RangeStats, BigInteger>
focal_class.var_name: journal
focal_class.declarator: comparisonExecutor
focal_class.modifier: private final
focal_class.original_string: private final ComparisonExecutor comparisonExecutor;
focal_class.type: ComparisonExecutor
focal_class.var_name: comparisonExecutor
focal_class.file: spark-job/src/main/java/org/apache/cassandra/diff/RangeComparator.java
focal_class.identifier: RangeComparator
focal_class.class_method_signature: RangeComparator.RangeComparator(DiffContext context,
                             BiConsumer<Throwable, BigInteger> errorReporter,
                             BiConsumer<MismatchType,BigInteger> mismatchReporter,
                             BiConsumer<RangeStats, BigInteger> journal,
                             ComparisonExecutor comparisonExecutor)
focal_class.full_signature: public  RangeComparator(DiffContext context,
                             BiConsumer<Throwable, BigInteger> errorReporter,
                             BiConsumer<MismatchType,BigInteger> mismatchReporter,
                             BiConsumer<RangeStats, BigInteger> journal,
                             ComparisonExecutor comparisonExecutor)
focal_class.identifier: RangeComparator
focal_class.modifiers: public
focal_class.parameters: (DiffContext context,
                             BiConsumer<Throwable, BigInteger> errorReporter,
                             BiConsumer<MismatchType,BigInteger> mismatchReporter,
                             BiConsumer<RangeStats, BigInteger> journal,
                             ComparisonExecutor comparisonExecutor)
focal_class.signature:  RangeComparator(DiffContext context,
                             BiConsumer<Throwable, BigInteger> errorReporter,
                             BiConsumer<MismatchType,BigInteger> mismatchReporter,
                             BiConsumer<RangeStats, BigInteger> journal,
                             ComparisonExecutor comparisonExecutor)
focal_class.class_method_signature: RangeComparator.compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_class.full_signature: public RangeStats compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_class.identifier: compare
focal_class.modifiers: public
focal_class.parameters: (Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_class.return: RangeStats
focal_class.signature: RangeStats compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_class.class_method_signature: RangeComparator.drain(Iterator<PartitionKey> sourceKeys,
                               Iterator<PartitionKey> targetKeys,
                               RangeStats rangeStats)
focal_class.full_signature: private void drain(Iterator<PartitionKey> sourceKeys,
                               Iterator<PartitionKey> targetKeys,
                               RangeStats rangeStats)
focal_class.identifier: drain
focal_class.modifiers: private
focal_class.parameters: (Iterator<PartitionKey> sourceKeys,
                               Iterator<PartitionKey> targetKeys,
                               RangeStats rangeStats)
focal_class.return: void
focal_class.signature: void drain(Iterator<PartitionKey> sourceKeys,
                               Iterator<PartitionKey> targetKeys,
                               RangeStats rangeStats)
focal_class.class_method_signature: RangeComparator.onlyInTarget(RangeStats stats, PartitionKey key)
focal_class.full_signature: private void onlyInTarget(RangeStats stats, PartitionKey key)
focal_class.identifier: onlyInTarget
focal_class.modifiers: private
focal_class.parameters: (RangeStats stats, PartitionKey key)
focal_class.return: void
focal_class.signature: void onlyInTarget(RangeStats stats, PartitionKey key)
focal_class.class_method_signature: RangeComparator.onlyInSource(RangeStats stats, PartitionKey key)
focal_class.full_signature: private void onlyInSource(RangeStats stats, PartitionKey key)
focal_class.identifier: onlyInSource
focal_class.modifiers: private
focal_class.parameters: (RangeStats stats, PartitionKey key)
focal_class.return: void
focal_class.signature: void onlyInSource(RangeStats stats, PartitionKey key)
focal_class.class_method_signature: RangeComparator.nextKey(Iterator<PartitionKey> keys)
focal_class.full_signature: private PartitionKey nextKey(Iterator<PartitionKey> keys)
focal_class.identifier: nextKey
focal_class.modifiers: private
focal_class.parameters: (Iterator<PartitionKey> keys)
focal_class.return: PartitionKey
focal_class.signature: PartitionKey nextKey(Iterator<PartitionKey> keys)
focal_class.class_method_signature: RangeComparator.onSuccess(final RangeStats rangeStats,
                                                 final AtomicLong partitionCount,
                                                 final BigInteger currentToken,
                                                 final AtomicReference<BigInteger> highestSeenToken,
                                                 final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                                 final BiConsumer<RangeStats, BigInteger> journal)
focal_class.full_signature: private Consumer<PartitionStats> onSuccess(final RangeStats rangeStats,
                                                 final AtomicLong partitionCount,
                                                 final BigInteger currentToken,
                                                 final AtomicReference<BigInteger> highestSeenToken,
                                                 final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                                 final BiConsumer<RangeStats, BigInteger> journal)
focal_class.identifier: onSuccess
focal_class.modifiers: private
focal_class.parameters: (final RangeStats rangeStats,
                                                 final AtomicLong partitionCount,
                                                 final BigInteger currentToken,
                                                 final AtomicReference<BigInteger> highestSeenToken,
                                                 final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                                 final BiConsumer<RangeStats, BigInteger> journal)
focal_class.return: Consumer<PartitionStats>
focal_class.signature: Consumer<PartitionStats> onSuccess(final RangeStats rangeStats,
                                                 final AtomicLong partitionCount,
                                                 final BigInteger currentToken,
                                                 final AtomicReference<BigInteger> highestSeenToken,
                                                 final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                                 final BiConsumer<RangeStats, BigInteger> journal)
focal_class.class_method_signature: RangeComparator.onError(final RangeStats rangeStats,
                                          final BigInteger currentToken,
                                          final BiConsumer<Throwable, BigInteger> errorReporter)
focal_class.full_signature: private Consumer<Throwable> onError(final RangeStats rangeStats,
                                          final BigInteger currentToken,
                                          final BiConsumer<Throwable, BigInteger> errorReporter)
focal_class.identifier: onError
focal_class.modifiers: private
focal_class.parameters: (final RangeStats rangeStats,
                                          final BigInteger currentToken,
                                          final BiConsumer<Throwable, BigInteger> errorReporter)
focal_class.return: Consumer<Throwable>
focal_class.signature: Consumer<Throwable> onError(final RangeStats rangeStats,
                                          final BigInteger currentToken,
                                          final BiConsumer<Throwable, BigInteger> errorReporter)
focal_class.class_method_signature: RangeComparator.recordError(final RangeStats rangeStats,
                               final BigInteger currentToken,
                               final BiConsumer<Throwable, BigInteger> errorReporter,
                               final Throwable error)
focal_class.full_signature: private void recordError(final RangeStats rangeStats,
                               final BigInteger currentToken,
                               final BiConsumer<Throwable, BigInteger> errorReporter,
                               final Throwable error)
focal_class.identifier: recordError
focal_class.modifiers: private
focal_class.parameters: (final RangeStats rangeStats,
                               final BigInteger currentToken,
                               final BiConsumer<Throwable, BigInteger> errorReporter,
                               final Throwable error)
focal_class.return: void
focal_class.signature: void recordError(final RangeStats rangeStats,
                               final BigInteger currentToken,
                               final BiConsumer<Throwable, BigInteger> errorReporter,
                               final Throwable error)
focal_method.body: public RangeStats compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider) {
  
          final RangeStats rangeStats = RangeStats.newStats();
          // We can catch this condition earlier, but it doesn't hurt to also check here
          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);
  
              // special case for start of range - handles one cluster supplying an empty range
              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);
  
                      // For results where the key exists in both, we'll fire off an async task to walk the
                      // partition and compare all the rows. The result of that comparison is added to the
                      // totals for the range and the highest seen token updated in the onSuccess callback
  
                      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) {
                          // Handle errors thrown when creating the comparison task. This should trap timeouts and
                          // unavailables occurring when performing the initial query to read the full partition.
                          // Errors thrown when paging through the partition in comparisonTask will be handled by
                          // the onError callback.
                          recordError(rangeStats, token, errorReporter, t);
                      } finally {
                          // if the cluster has been shutdown because the task failed the underlying iterators
                          // of partition keys will return hasNext == false
                          sourceKey = nextKey(sourceKeys);
                          targetKey = nextKey(targetKeys);
                      }
                  }
              }
  
              // handle case where only one iterator is exhausted
              if (sourceKey != null)
                  onlyInSource(rangeStats, sourceKey);
              else if (targetKey != null)
                  onlyInTarget(rangeStats, targetKey);
  
              drain(sourceKeys, targetKeys, rangeStats);
  
          } catch (Exception e) {
              // Handles errors thrown by iteration of underlying resultsets of partition keys by
              // calls to nextKey(). Such errors should cause the overall range comparison to fail,
              // but we must ensure that any in-flight partition comparisons complete so that either
              // the onSuccess or onError callback is fired for each one. This is necessary to ensure
              // that we record the highest seen token and any failed partitions and can safely re-run.
              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;
      }
focal_method.class_method_signature: RangeComparator.compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_method.full_signature: public RangeStats compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_method.identifier: compare
focal_method.invocations: newStats equals info nextKey nextKey info onlyInTarget forEachRemaining onlyInTarget info onlyInSource forEachRemaining onlyInSource compareTo onlyInTarget nextKey onlyInSource nextKey verify equals isTokenAllowed getTokenAsBigInteger debug getTokenAsBigInteger skipPartition nextKey nextKey getTokenAsBigInteger apply submit onSuccess onError recordError nextKey nextKey onlyInSource onlyInTarget drain debug getUnarrivedParties arriveAndAwaitAdvance format debug getUnarrivedParties arriveAndAwaitAdvance allMatches info
focal_method.modifiers: public
focal_method.parameters: (Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)
focal_method.return: RangeStats
focal_method.signature: RangeStats compare(Iterator<PartitionKey> sourceKeys,
                                Iterator<PartitionKey> targetKeys,
                                Function<PartitionKey, PartitionComparator> partitionTaskProvider)