@Test public void differentDeleteAndExpiryTimesTest() throws Exception { Pair<Long, List<String>> expiryTimeAndSegmentsUnderCompaction = setupStateWithExpiredBlobsAtSpecificTime(); state.advanceTime(expiryTimeAndSegmentsUnderCompaction.getFirst() + Time.MsPerSec - state.time.milliseconds()); long deleteReferenceTimeMs = state.time.milliseconds(); state.advanceTime(Time.MsPerSec); int deleteCount = 10; Set<MockId> idsInSegments = getIdsWithPutInSegments(expiryTimeAndSegmentsUnderCompaction.getSecond()); List<MockId> idsToExamine = new ArrayList<>(); for (MockId id : idsInSegments) { if (state.liveKeys.contains(id)) { state.addDeleteEntry(id); idsToExamine.add(id); if (idsToExamine.size() == deleteCount) { break; } } } state.reloadIndex(true, false); compactAndVerify(expiryTimeAndSegmentsUnderCompaction.getSecond(), deleteReferenceTimeMs, true); state.reloadIndex(true, false); for (MockId id : idsToExamine) { state.index.getBlobReadInfo(id, EnumSet.of(StoreGetOptions.Store_Include_Deleted)).close(); } }
void close(long waitTimeSecs) throws InterruptedException { isActive = false; if (waitTimeSecs > 0 && !runningLatch.await(waitTimeSecs, TimeUnit.SECONDS)) { logger.error("Compactor did not shutdown within {} seconds for {}", waitTimeSecs, storeId); } logger.trace("Closed BlobStoreCompactor for {}", storeId); }
BlobStoreCompactor { void close(long waitTimeSecs) throws InterruptedException { isActive = false; if (waitTimeSecs > 0 && !runningLatch.await(waitTimeSecs, TimeUnit.SECONDS)) { logger.error("Compactor did not shutdown within {} seconds for {}", waitTimeSecs, storeId); } logger.trace("Closed BlobStoreCompactor for {}", storeId); } }
BlobStoreCompactor { void close(long waitTimeSecs) throws InterruptedException { isActive = false; if (waitTimeSecs > 0 && !runningLatch.await(waitTimeSecs, TimeUnit.SECONDS)) { logger.error("Compactor did not shutdown within {} seconds for {}", waitTimeSecs, storeId); } logger.trace("Closed BlobStoreCompactor for {}", storeId); } BlobStoreCompactor(String dataDir, String storeId, StoreKeyFactory storeKeyFactory, StoreConfig config,
      StoreMetrics srcMetrics, StoreMetrics tgtMetrics, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, Log srcLog, Time time, UUID sessionId, UUID incarnationId,
      AccountService accountService); }
BlobStoreCompactor { void close(long waitTimeSecs) throws InterruptedException { isActive = false; if (waitTimeSecs > 0 && !runningLatch.await(waitTimeSecs, TimeUnit.SECONDS)) { logger.error("Compactor did not shutdown within {} seconds for {}", waitTimeSecs, storeId); } logger.trace("Closed BlobStoreCompactor for {}", storeId); } BlobStoreCompactor(String dataDir, String storeId, StoreKeyFactory storeKeyFactory, StoreConfig config,
      StoreMetrics srcMetrics, StoreMetrics tgtMetrics, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, Log srcLog, Time time, UUID sessionId, UUID incarnationId,
      AccountService accountService);  }
BlobStoreCompactor { void close(long waitTimeSecs) throws InterruptedException { isActive = false; if (waitTimeSecs > 0 && !runningLatch.await(waitTimeSecs, TimeUnit.SECONDS)) { logger.error("Compactor did not shutdown within {} seconds for {}", waitTimeSecs, storeId); } logger.trace("Closed BlobStoreCompactor for {}", storeId); } BlobStoreCompactor(String dataDir, String storeId, StoreKeyFactory storeKeyFactory, StoreConfig config,
      StoreMetrics srcMetrics, StoreMetrics tgtMetrics, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, Log srcLog, Time time, UUID sessionId, UUID incarnationId,
      AccountService accountService);   }