Question regarding Checkpoint logic in SortedLedgerStorage

[Charan Reddy G]

Hey Sijie/IvanK,

With
https://github.com/apache/bookkeeper/commit/d175ada58dcaf78f0a70b0ebebf489255ae67b5f
you introduced Bookkeeper-564 : Better checkpoint mechanism -  Scheduling
checkpoint only when rotating an entry log file.

I'm trying to understand how it would work in the following scenario
- using SortedLedgerStorage
- since it is SortedLedgerStorage entries would be in EntryMemtable
- GarbageCollectorThread.EntryLogScanner.process method calls entryLogger
.addEntry(ledgerId, entry)
- in EntryLogger.addEntry method, lets say it comes to know it has reached
EntryLogLimit and creates NewLog
-  since current active entrylog is rotated,
EntryLogListener.onRotateEntryLog is called
- which sets the currentMark of journal to checkpointHolder. Point to note,
that all the entries added to the Bookie are not added to entryLog yet,
there are entries still in entrymemtable
- lets say SyncThread tries to checkpoint at this instant

now the concern is, in SortedLedgerStorage.checkpoint method, before
calling super.checkpoint(checkpoint), it does memTable.flush(this,
checkpoint); But memTable.flush would just add entries to the current
active entrylog (BufferedLogChannel) and it doesn't guarantee persistence.
super(InterLeavedLedgerStorage).checkpoint will only flushRotatedLogs
(persists) and finally mark the checkpointcomplete with 'lastcheckpoint',
but the 'lastCheckpoint' in the checkpointHolder would also include the
entries which were in Entrymemtable and are not actually persisted in the
whole process. Is there issue in SortedLedgerStorage checkpoint logic?

        @Override
    public Checkpoint checkpoint(final Checkpoint checkpoint) throws
IOException {
        Checkpoint lastCheckpoint = checkpointHolder.getLastCheckpoint();
        // if checkpoint is less than last checkpoint, we don't need to do
checkpoint again.
        if (lastCheckpoint.compareTo(checkpoint) > 0) {
            return lastCheckpoint;
        }
        memTable.flush(this, checkpoint);
        return super.checkpoint(checkpoint);
    }

Thanks,
Charan