@Test public void testBucketingBasic() throws StoreException, InterruptedException, IOException { assumeTrue(bucketingEnabled); final CountDownLatch scanStartedLatch = new CountDownLatch(1); MockThrottler mockThrottler = new MockThrottler(scanStartedLatch, new CountDownLatch(0)); throttlers.put(BlobStoreStats.IO_SCHEDULER_JOB_TYPE, mockThrottler); long logSegmentForecastOffsetMs = state.time.milliseconds(); int bucketCount = 2 * (int) (logSegmentForecastOffsetMs / BUCKET_SPAN_IN_MS); long expiresAtInMs = ((long) bucketCount - 2) * BUCKET_SPAN_IN_MS; state.addPutEntries(3, PUT_RECORD_SIZE, expiresAtInMs); List<IndexEntry> puts = state.addPutEntries(2, PUT_RECORD_SIZE, Utils.Infinite_Time); MockId id1 = (MockId) puts.get(0).getKey(); MockId id2 = (MockId) puts.get(1).getKey(); state.addDeleteEntry(id1); state.addDeleteEntry(id1, null, (short) 1); state.addDeleteEntry(id2); state.addUndeleteEntry(id2); state.addUndeleteEntry(id2, (short) 2); int expectedThrottleCount = state.referenceIndex.size(); long logSegmentForecastStartTimeInMs = state.time.milliseconds() - logSegmentForecastOffsetMs; long logSegmentForecastEndTimeInMs = logSegmentForecastStartTimeInMs + bucketCount * BUCKET_SPAN_IN_MS; long containerForecastEndTimeInMs = state.time.milliseconds() + bucketCount * BUCKET_SPAN_IN_MS; BlobStoreStats blobStoreStats = setupBlobStoreStats(bucketCount, logSegmentForecastOffsetMs); assertTrue("IndexScanner took too long to start", scanStartedLatch.await(5, TimeUnit.SECONDS)); verifyAndGetContainerValidSize(blobStoreStats, state.time.milliseconds()); for (long i = logSegmentForecastStartTimeInMs; i <= state.time.milliseconds() + TEST_TIME_INTERVAL_IN_MS; i += TEST_TIME_INTERVAL_IN_MS) { verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(i, 0L)); } long timeToLiveInMs = expiresAtInMs - state.time.milliseconds() < 0 ? 0 : expiresAtInMs - state.time.milliseconds(); state.advanceTime(timeToLiveInMs + Time.MsPerSec); for (long i = logSegmentForecastStartTimeInMs; i <= state.time.milliseconds() + TEST_TIME_INTERVAL_IN_MS; i += TEST_TIME_INTERVAL_IN_MS) { verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(i, 0L)); } verifyAndGetContainerValidSize(blobStoreStats, state.time.milliseconds()); state.advanceTime(logSegmentForecastEndTimeInMs - state.time.milliseconds() - 1); verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(state.time.milliseconds(), Time.MsPerSec)); state.advanceTime(containerForecastEndTimeInMs - state.time.milliseconds() - 1); verifyAndGetContainerValidSize(blobStoreStats, state.time.milliseconds()); assertEquals("Throttle count mismatch from expected value", expectedThrottleCount, mockThrottler.throttleCount.get()); blobStoreStats.close(); }
@Override public void close() { if (enabled.compareAndSet(true, false)) { if (indexScanner != null) { indexScanner.cancel(); } if (queueProcessor != null) { queueProcessor.cancel(); } } }
BlobStoreStats implements StoreStats, Closeable { @Override public void close() { if (enabled.compareAndSet(true, false)) { if (indexScanner != null) { indexScanner.cancel(); } if (queueProcessor != null) { queueProcessor.cancel(); } } } }
BlobStoreStats implements StoreStats, Closeable { @Override public void close() { if (enabled.compareAndSet(true, false)) { if (indexScanner != null) { indexScanner.cancel(); } if (queueProcessor != null) { queueProcessor.cancel(); } } } BlobStoreStats(String storeId, PersistentIndex index, int bucketCount, long bucketSpanTimeInMs,
      long logSegmentForecastOffsetMs, long queueProcessingPeriodInMs, long waitTimeoutInSecs,
      boolean enableBucketForLogSegmentReports, Time time, ScheduledExecutorService longLiveTaskScheduler,
      ScheduledExecutorService shortLiveTaskScheduler, DiskIOScheduler diskIOScheduler, StoreMetrics metrics); }
BlobStoreStats implements StoreStats, Closeable { @Override public void close() { if (enabled.compareAndSet(true, false)) { if (indexScanner != null) { indexScanner.cancel(); } if (queueProcessor != null) { queueProcessor.cancel(); } } } BlobStoreStats(String storeId, PersistentIndex index, int bucketCount, long bucketSpanTimeInMs,
      long logSegmentForecastOffsetMs, long queueProcessingPeriodInMs, long waitTimeoutInSecs,
      boolean enableBucketForLogSegmentReports, Time time, ScheduledExecutorService longLiveTaskScheduler,
      ScheduledExecutorService shortLiveTaskScheduler, DiskIOScheduler diskIOScheduler, StoreMetrics metrics); @Override Pair<Long, Long> getValidSize(TimeRange timeRange); @Override Map<StatsReportType, StatsSnapshot> getStatsSnapshots(Set<StatsReportType> statsReportTypes,
      long referenceTimeInMs); @Override void close(); }
BlobStoreStats implements StoreStats, Closeable { @Override public void close() { if (enabled.compareAndSet(true, false)) { if (indexScanner != null) { indexScanner.cancel(); } if (queueProcessor != null) { queueProcessor.cancel(); } } } BlobStoreStats(String storeId, PersistentIndex index, int bucketCount, long bucketSpanTimeInMs,
      long logSegmentForecastOffsetMs, long queueProcessingPeriodInMs, long waitTimeoutInSecs,
      boolean enableBucketForLogSegmentReports, Time time, ScheduledExecutorService longLiveTaskScheduler,
      ScheduledExecutorService shortLiveTaskScheduler, DiskIOScheduler diskIOScheduler, StoreMetrics metrics); @Override Pair<Long, Long> getValidSize(TimeRange timeRange); @Override Map<StatsReportType, StatsSnapshot> getStatsSnapshots(Set<StatsReportType> statsReportTypes,
      long referenceTimeInMs); @Override void close();  }