SweepBatchAccumulator { void accumulateBatch(SweepBatch sweepBatch) { Preconditions.checkState(sweepBatch.lastSweptTimestamp() < sweepTimestamp, "Tried to accumulate a batch %s at timestamp %s that went beyond the sweep timestamp %s!" + " This is unexpected, and suggests a bug in the way we read in targeted sweep." + " This by itself does not mean that AtlasDB service is compromised, but targeted sweep" + " may not be working.", sweepBatch, sweepBatch.lastSweptTimestamp(), sweepTimestamp); accumulatedWrites.addAll(sweepBatch.writes()); accumulatedDedicatedRows.addAll(sweepBatch.dedicatedRows().getDedicatedRows()); addRelevantFinePartitions(sweepBatch); progressTimestamp = Math.max(progressTimestamp, sweepBatch.lastSweptTimestamp()); anyBatchesPresent = true; nextBatchAvailable = sweepBatch.hasNext(); entriesRead += sweepBatch.entriesRead(); } SweepBatchAccumulator(long sweepTimestamp, long progressTimestamp);   }
@Test public void throwsWhenAttemptingToMergeWriteInfoAtTheSweepTimestamp() { assertThatThrownBy(() -> accumulator.accumulateBatch(SweepBatch.of( ImmutableList.of(WriteInfo.write(TABLE_REFERENCE_1, CELL_1, PROGRESS_TIMESTAMP + 5)), NO_DEDICATED_ROWS, SWEEP_TIMESTAMP))) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Tried to accumulate a batch") .hasMessageContaining("went beyond the sweep timestamp"); }
@Test public void throwsWhenAttemptingToMergeWriteInfoAfterTheSweepTimestamp() { assertThatThrownBy(() -> accumulator.accumulateBatch(SweepBatch.of( ImmutableList.of(WriteInfo.write(TABLE_REFERENCE_1, CELL_1, PROGRESS_TIMESTAMP + 5)), NO_DEDICATED_ROWS, SWEEP_TIMESTAMP + 1))) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Tried to accumulate a batch") .hasMessageContaining("went beyond the sweep timestamp"); }