@Test public void sweepHandlesSequencesOfDeletesAndReadditionsInOneShot() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); enqueueTombstone(TABLE_CONS, LOW_TS + 2); enqueueWriteCommitted(TABLE_CONS, LOW_TS + 4); enqueueTombstone(TABLE_CONS, LOW_TS + 6); enqueueWriteCommitted(TABLE_CONS, LOW_TS + 8); for (int partition = 0; partition < readBatchSize; partition++) { enqueueWriteCommitted(TABLE_CONS, minTsForFinePartition(partition + 1)); } sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); for (int i = 0; i < 10; i = i + 2) { assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + i); } if (readBatchSize == 1) { assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS + 8); } else { assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, minTsForFinePartition(readBatchSize)); } verify(spiedKvs, times(1)).deleteAllTimestamps(any(), any()); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(1); assertThat(metricsManager).hasLastSweptTimestampConservativeEqualTo( maxTsForFinePartition(readBatchSize - 1)); setTimelockTime(10_000L); punchTimeAtTimestamp(5_000L, LOW_TS + 8); assertThat(metricsManager).hasMillisSinceLastSweptConservativeEqualTo(10_000L - 5000L); }
@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();  }