@Test public void sweepProgressesAndSkipsEmptyFinePartitions() { setSweepTimestamp(minTsForFinePartition(2 * (2 * readBatchSize) + 2)); List<Integer> permittedPartitions = Lists.newArrayList(); for (int index = 0; index <= 2 * readBatchSize; index++) { int partitionToUse = 2 * index; long timestampToUse = minTsForFinePartition(partitionToUse) + LOW_TS; enqueueWriteCommitted(TABLE_CONS, timestampToUse); punchTimeAtTimestamp(100 * (index + 1), timestampToUse); permittedPartitions.add(partitionToUse); } int finalPartition = 2 * (2 * readBatchSize); long additionalTimestampForFinalPartition = minTsForFinePartition(finalPartition) + LOW_TS + 1; enqueueWriteCommitted(TABLE_CONS, additionalTimestampForFinalPartition); int finalValueWallClockTime = 100 * (2 * readBatchSize + 1) + 1; punchTimeAtTimestamp(finalValueWallClockTime, additionalTimestampForFinalPartition); enqueueWriteCommitted(TABLE_CONS, minTsForFinePartition(9_999_999)); punchTimeAtTimestamp(Long.MAX_VALUE, minTsForFinePartition(9_999_999)); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); long timestampEndOfBatchOne = minTsForFinePartition(permittedPartitions.get(readBatchSize - 1)) + LOW_TS; assertReadAtTimestampReturnsSentinel(TABLE_CONS, timestampEndOfBatchOne); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, timestampEndOfBatchOne); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(1); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); long timestampEndOfBatchTwo = minTsForFinePartition(permittedPartitions.get(2 * readBatchSize - 1)) + LOW_TS; assertReadAtTimestampReturnsSentinel(TABLE_CONS, timestampEndOfBatchTwo); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, timestampEndOfBatchTwo); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(2); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); long lastSweepableTimestamp = minTsForFinePartition(permittedPartitions.get(2 * readBatchSize)) + LOW_TS + 1; assertReadAtTimestampReturnsSentinel(TABLE_CONS, lastSweepableTimestamp); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, lastSweepableTimestamp); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(3); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(2 * readBatchSize + 2); assertThat(metricsManager).hasLastSweptTimestampConservativeEqualTo( maxTsForFinePartition(permittedPartitions.get(2 * readBatchSize))); assertThat(metricsManager).containsEntriesReadInBatchConservative(readBatchSize, readBatchSize, 2L); assertThat(metricsManager).hasEntriesReadInBatchMeanConservativeEqualTo((2 * readBatchSize + 2) / 3.0); setTimelockTime(5000L); assertThat(metricsManager).hasMillisSinceLastSweptConservativeEqualTo(5000L - finalValueWallClockTime); }
@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();  }