@Test public void batchIncludesAllWritesWithTheSameTimestampAndDoesNotSkipOrRepeatAnyWritesInNextIteration() { TargetedSweeper sweeperConservative = getSingleShardSweeper(); int relativePrime = MAX_CELLS_GENERIC - 1; assertThat(SWEEP_BATCH_SIZE % relativePrime).isNotEqualTo(0); int minTsToReachBatchSize = (SWEEP_BATCH_SIZE - 1) / relativePrime + 1; commitTransactionsWithWritesIntoUniqueCells(minTsToReachBatchSize + 1, relativePrime, sweeperConservative); sweepNextBatch(sweeperConservative, ShardAndStrategy.conservative(0)); ArgumentCaptor<Map> map = ArgumentCaptor.forClass(Map.class); verify(spiedKvs, atLeast(1)).deleteAllTimestamps(eq(TABLE_CONS), map.capture()); assertThat(map.getAllValues().stream().map(Map::size).mapToInt(x -> x).sum()) .isEqualTo(minTsToReachBatchSize * relativePrime); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(0))) .isEqualTo(1000 + minTsToReachBatchSize - 1); sweepNextBatch(sweeperConservative, ShardAndStrategy.conservative(0)); verify(spiedKvs, atLeast(2)).deleteAllTimestamps(eq(TABLE_CONS), map.capture()); assertThat(map.getValue().size()).isEqualTo(relativePrime); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(0))) .isEqualTo(maxTsForFinePartition(0)); }
@SuppressWarnings("checkstyle:RegexpMultiline") @VisibleForTesting public long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive) { assertInitialized(); return queue.sweepNextBatch(shardStrategy, maxTsExclusive); }
TargetedSweeper implements MultiTableSweepQueueWriter, BackgroundSweeper { @SuppressWarnings("checkstyle:RegexpMultiline") @VisibleForTesting public long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive) { assertInitialized(); return queue.sweepNextBatch(shardStrategy, maxTsExclusive); } }
TargetedSweeper implements MultiTableSweepQueueWriter, BackgroundSweeper { @SuppressWarnings("checkstyle:RegexpMultiline") @VisibleForTesting public long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive) { assertInitialized(); return queue.sweepNextBatch(shardStrategy, maxTsExclusive); } private  TargetedSweeper(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); }
TargetedSweeper implements MultiTableSweepQueueWriter, BackgroundSweeper { @SuppressWarnings("checkstyle:RegexpMultiline") @VisibleForTesting public long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive) { assertInitialized(); return queue.sweepNextBatch(shardStrategy, maxTsExclusive); } private  TargetedSweeper(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); static TargetedSweeper createUninitialized(MetricsManager metrics,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); static TargetedSweeper createUninitializedForTest(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime); static TargetedSweeper createUninitializedForTest(Supplier<Integer> shards); @Override void initialize(TransactionManager txManager); void initializeWithoutRunning(TransactionManager txManager); void initializeWithoutRunning(
            SpecialTimestampsSupplier timestamps,
            TimelockService timelockService,
            KeyValueService kvs,
            TransactionService transaction,
            TargetedSweepFollower follower); @Override void runInBackground(); @Override void enqueue(List<WriteInfo> writes); @SuppressWarnings("checkstyle:RegexpMultiline") // Suppress VisibleForTesting warning @VisibleForTesting long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive); @Override void close(); @Override void shutdown(); }
TargetedSweeper implements MultiTableSweepQueueWriter, BackgroundSweeper { @SuppressWarnings("checkstyle:RegexpMultiline") @VisibleForTesting public long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive) { assertInitialized(); return queue.sweepNextBatch(shardStrategy, maxTsExclusive); } private  TargetedSweeper(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); static TargetedSweeper createUninitialized(MetricsManager metrics,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); static TargetedSweeper createUninitializedForTest(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime); static TargetedSweeper createUninitializedForTest(Supplier<Integer> shards); @Override void initialize(TransactionManager txManager); void initializeWithoutRunning(TransactionManager txManager); void initializeWithoutRunning(
            SpecialTimestampsSupplier timestamps,
            TimelockService timelockService,
            KeyValueService kvs,
            TransactionService transaction,
            TargetedSweepFollower follower); @Override void runInBackground(); @Override void enqueue(List<WriteInfo> writes); @SuppressWarnings("checkstyle:RegexpMultiline") // Suppress VisibleForTesting warning @VisibleForTesting long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive); @Override void close(); @Override void shutdown();  }