Oplog implements CompactableOplog, Flushable { @Override public int compact(OplogCompactor compactor) { if (!needsCompaction()) { return 0; } isCompactorThread.set(Boolean.TRUE); assert calledByCompactorThread(); getParent().acquireCompactorReadLock(); try { if (!compactor.keepCompactorRunning()) { return 0; } lockCompactor(); try { if (hasNoLiveValues()) { handleNoLiveValues(); return 0; } long opStart = getStats().getStatTime(); BytesAndBitsForCompactor wrapper = new BytesAndBitsForCompactor(); DiskEntry de; DiskEntry lastDe = null; boolean compactFailed = !compactor.keepCompactorRunning(); int totalCount = 0; for (DiskRegionInfo dri : this.regionMap.values()) { final DiskRegionView dr = dri.getDiskRegion(); if (dr == null) continue; boolean didCompact = false; while ((de = dri.getNextLiveEntry()) != null) { if (!compactor.keepCompactorRunning()) { compactFailed = true; break; } if (lastDe != null) { if (lastDe == de) { throw new IllegalStateException("compactor would have gone into infinite loop"); } assert lastDe != de; } lastDe = de; didCompact = false; synchronized (de) { DiskId did = de.getDiskId(); assert did != null; synchronized (did) { long oplogId = did.getOplogId(); if (oplogId != getOplogId()) { continue; } boolean toCompact = getBytesAndBitsForCompaction(dr, de, wrapper); if (toCompact) { if (oplogId != did.getOplogId()) { if (!wrapper.isReusable()) { wrapper = new BytesAndBitsForCompactor(); } else if (wrapper.getOffHeapData() != null) { wrapper.setOffHeapData(null, (byte) 0); } continue; } getOplogSet().getChild().copyForwardModifyForCompact(dr, de, wrapper); didCompact = true; } } } if (didCompact) { totalCount++; getStats().endCompactionUpdate(opStart); opStart = getStats().getStatTime(); if (!wrapper.isReusable()) { wrapper = new BytesAndBitsForCompactor(); } } } } cleanupAfterCompaction(compactFailed); return totalCount; } finally { unlockCompactor(); } } finally { getParent().releaseCompactorReadLock(); assert calledByCompactorThread(); isCompactorThread.remove(); } } Oplog(long oplogId, PersistentOplogSet parent, DirectoryHolder dirHolder); private  Oplog(long oplogId, DirectoryHolder dirHolder, Oplog prevOplog);   Oplog(long oplogId, PersistentOplogSet parent); Object getLock(); void replaceIncompatibleEntry(DiskRegionView dr, DiskEntry old, DiskEntry repl); KnownVersion currentRecoveredGFVersion(); DiskStoreStats getStats(); File getCrfFile(); File getDrfFile(); Set<String> gatherMatchingOplogFiles(Set<String> oplogFileNames); boolean hasKrf(); long getOplogId(); @Override BytesAndBits getBytesAndBits(DiskRegionView dr, DiskId id, boolean faultIn,
      boolean bitOnly); @Override BytesAndBits getNoBuffer(DiskRegion dr, DiskId id); long calcModEntryId(long delta); long calcDelEntryId(long delta); DiskEntry.RecoveredEntry createRecoveredEntry(byte[] valueBytes, int valueLength,
      byte userBits, long oplogId, long offsetInOplog, long oplogKeyId, boolean recoverValue,
      KnownVersion version, ByteArrayDataInput in); void close(); void testClose(); void destroy(); void destroyCrfOnly(); void create(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async); void krfFileCreate(); void krfClose(); File getKrfFile(); List<KRFEntry> getSortedLiveEntries(Collection<DiskRegionInfo> targetRegions); void modify(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async); void offlineModify(DiskRegionView drv, DiskEntry entry, byte[] value,
      boolean isSerializedObject); void saveConflictVersionTag(InternalRegion region, VersionTag tag, boolean async); boolean needsKrf(); void remove(InternalRegion region, DiskEntry entry, boolean async, boolean isClear); void writeGCRVV(DiskRegion dr); void writeRVV(DiskRegion dr, RegionVersionVector sourceRVV, Boolean isRVVTrusted); @Override void flush(); @Override void flush(ByteBuffer bb, ByteBuffer chunkbb); void flushAll(); void flushAllNoSync(boolean skipDrf); void flushAll(boolean skipDrf); void flushAll(boolean skipDrf, boolean doSync); void deleteCRF(); void deleteCRFFileOnly(); void deleteDRF(); void deleteDRFFileOnly(); DirectoryHolder getDirectoryHolder(); boolean hadLiveEntries(); boolean hasNoLiveValues(); @Override void prepareForCompact(); @Override int compact(OplogCompactor compactor); static boolean isCRFFile(String filename); static boolean isDRFFile(String filename); static String getKRFFilenameFromCRFFilename(String crfFilename); void recoverValuesIfNeeded(Map<Long, DiskRecoveryStore> diskRecoveryStores); @Override String toString(); void finishKrf(); KnownVersion getProductVersionIfOld(); KnownVersion getDataVersionIfOld(); static final String CRF_FILE_EXT; static final String DRF_FILE_EXT; static final String KRF_FILE_EXT; static final int OPLOG_MAGIC_SEQ_REC_SIZE; }
@Test public void noCompactIfNotKeepCompactorRunning() { when(compactor.keepCompactorRunning()).thenReturn(false); assertThat(oplog.compact(compactor)).isEqualTo(0); }