db-storage-cleanup would impact write request

[gaozhangmin]

I have a question about db-storage-cleanup executor, Why indexes and ledgers metadata are cleaned Only when writeCacge.isEmpty()

bookkeeper/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java

Lines 742 to 816 in 4ca4b4e

	public void checkpoint(Checkpoint checkpoint) throws IOException {
	Checkpoint thisCheckpoint = checkpointSource.newCheckpoint();
	if (lastCheckpoint.compareTo(checkpoint) > 0) {
	return;
	}
	// Only a single flush operation can happen at a time
	flushMutex.lock();
	long startTime = -1;
	try {
	startTime = MathUtils.nowInNano();
	} catch (Throwable e) {
	// Fix spotbugs warning. Should never happen
	flushMutex.unlock();
	throw new IOException(e);
	}
	try {
	if (writeCache.isEmpty()) {
	return;
	}
	// Swap the write cache so that writes can continue to happen while the flush is
	// ongoing
	swapWriteCache();
	long sizeToFlush = writeCacheBeingFlushed.size();
	if (log.isDebugEnabled()) {
	log.debug("Flushing entries. count: {} -- size {} Mb", writeCacheBeingFlushed.count(),
	sizeToFlush / 1024.0 / 1024);
	}
	// Write all the pending entries into the entry logger and collect the offset
	// position for each entry
	Batch batch = entryLocationIndex.newBatch();
	writeCacheBeingFlushed.forEach((ledgerId, entryId, entry) -> {
	try {
	long location = entryLogger.addEntry(ledgerId, entry);
	entryLocationIndex.addLocation(batch, ledgerId, entryId, location);
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	});
	long entryLoggerStart = MathUtils.nowInNano();
	entryLogger.flush();
	recordSuccessfulEvent(dbLedgerStorageStats.getFlushEntryLogStats(), entryLoggerStart);
	long batchFlushStartTime = MathUtils.nowInNano();
	batch.flush();
	batch.close();
	recordSuccessfulEvent(dbLedgerStorageStats.getFlushLocationIndexStats(), batchFlushStartTime);
	if (log.isDebugEnabled()) {
	log.debug("DB batch flushed time : {} s",
	MathUtils.elapsedNanos(batchFlushStartTime) / (double) TimeUnit.SECONDS.toNanos(1));
	}
	long ledgerIndexStartTime = MathUtils.nowInNano();
	ledgerIndex.flush();
	recordSuccessfulEvent(dbLedgerStorageStats.getFlushLedgerIndexStats(), ledgerIndexStartTime);
	cleanupExecutor.execute(() -> {
	// There can only be one single cleanup task running because the cleanupExecutor
	// is single-threaded
	try {
	if (log.isDebugEnabled()) {
	log.debug("Removing deleted ledgers from db indexes");
	}
	entryLocationIndex.removeOffsetFromDeletedLedgers();
	ledgerIndex.removeDeletedLedgers();
	} catch (Throwable t) {
	log.warn("Failed to cleanup db indexes", t);
	}
	});

There is a case, if bookie transits to readonly mode for a long time, during readonly, lots of ledgers would be deleted, but because writeCacge.isEmpty(), the index would remain.
After we make the bookie writable, a lot of indexes gathered that need be deleted. put heavy load on RocksDB database