@Test public void stopReadingEarlyWhenEncounteringEntryKnownToBeCommittedAfterSweepTs() { immutableTs = 100L; enqueueWriteCommitted(TABLE_CONS, 10); enqueueWriteCommitedAt(TABLE_CONS, 30, 150); putTimestampIntoTransactionTable(50, 200); Map<Integer, Integer> largeWriteDistribution = enqueueAtLeastThresholdWritesInDefaultShardWithStartTs(100, 50); int writesInDedicated = largeWriteDistribution.get(CONS_SHARD); enqueueWriteUncommitted(TABLE_CONS, 70); enqueueWriteCommitted(TABLE_CONS, 90); assertThat(sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(4 + writesInDedicated); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated + 1); immutableTs = 170; sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated + 3); immutableTs = 250; sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated + 3 + writesInDedicated + 2); }
@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();  }