@Test(timeout = 5000) public void testBucketingWaitTimeout() throws InterruptedException, StoreException { assumeTrue(bucketingEnabled); CountDownLatch scanStartedLatch = new CountDownLatch(1); CountDownLatch scanHoldLatch = new CountDownLatch(1); MockThrottler mockThrottler = new MockThrottler(scanStartedLatch, scanHoldLatch); throttlers.put(BlobStoreStats.IO_SCHEDULER_JOB_TYPE, mockThrottler); int expectedMinimumThrottleCount = 2 * state.referenceIndex.size(); BlobStoreStats blobStoreStats = new BlobStoreStats("", state.index, 10, BUCKET_SPAN_IN_MS, 0, QUEUE_PROCESSOR_PERIOD_IN_Ms, 1, true, state.time, indexScannerScheduler, queueProcessorScheduler, diskIOScheduler, METRICS); assertTrue("IndexScanner took too long to start", scanStartedLatch.await(5, TimeUnit.SECONDS)); advanceTimeToNextSecond(); verifyAndGetContainerValidSize(blobStoreStats, state.time.milliseconds()); verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(state.time.milliseconds(), 0)); scanHoldLatch.countDown(); assertTrue("Throttle count is lower than the expected minimum throttle count", mockThrottler.throttleCount.get() >= expectedMinimumThrottleCount); 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();  }