@Test public void testValidDataSizeAfterExpiration() throws StoreException, IOException { assumeTrue(!bucketingEnabled); BlobStoreStats blobStoreStats = setupBlobStoreStats(0, 0); advanceTimeToNextSecond(); long timeInMsBeforePuts = state.time.milliseconds(); long totalLogSegmentValidSizeBeforePuts = verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(timeInMsBeforePuts, 0L)); long totalContainerValidSizeBeforePuts = verifyAndGetContainerValidSize(blobStoreStats, state.time.milliseconds()); state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time); long expiresAtInMs = state.time.milliseconds() + TimeUnit.SECONDS.toMillis(20); state.addPutEntries(3, PUT_RECORD_SIZE, expiresAtInMs); state.advanceTime(expiresAtInMs - state.time.milliseconds()); long expectedDeltaAfterPut = 4 * PUT_RECORD_SIZE; long timeInMsAfterPuts = state.time.milliseconds(); long totalLogSegmentValidSizeAfterPuts = verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(timeInMsAfterPuts, 0L)); long totalContainerValidSizeAfterPuts = verifyAndGetContainerValidSize(blobStoreStats, timeInMsAfterPuts); assertEquals("Put entries with expiry are not properly counted for log segment valid size", totalLogSegmentValidSizeAfterPuts, totalLogSegmentValidSizeBeforePuts + expectedDeltaAfterPut); assertEquals("Put entries with expiry are not properly counted for container valid size", totalContainerValidSizeAfterPuts, totalContainerValidSizeBeforePuts + expectedDeltaAfterPut); advanceTimeToNextSecond(); long expectedDeltaAfterExpiration = PUT_RECORD_SIZE; long timeInMsAfterExpiration = state.time.milliseconds(); long totalLogSegmentValidSizeAfterExpiration = verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(timeInMsAfterExpiration, 0L)); long totalContainerValidSizeAfterExpiration = verifyAndGetContainerValidSize(blobStoreStats, timeInMsAfterExpiration); assertEquals("Expired put entries are not properly counted for log segment valid size", totalLogSegmentValidSizeAfterExpiration, totalLogSegmentValidSizeBeforePuts + expectedDeltaAfterExpiration); assertEquals("Expired put entries are not properly counted for container valid size", totalContainerValidSizeAfterExpiration, totalContainerValidSizeBeforePuts + expectedDeltaAfterExpiration); 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();  }