private boolean stillNeedToCatchUp(String where, long offset) {
        if (!catchingUp) {
            log.trace("{}: we are not in the initial catching up state.", where);
            return false;
        }
        OptionalLong highWaterMark = highWaterMarkAccessor.get();
        if (!highWaterMark.isPresent()) {
            log.info("{}: the loader is still catching up because we still don't know the high " +
                    "water mark yet.", where);
            return true;
        }
        if (highWaterMark.getAsLong() - 1 > offset) {
            log.info("{}: The loader is still catching up because we have loaded up to offset " +
                    offset + ", but the high water mark is {}", where, highWaterMark.getAsLong());
            return true;
        }
        if (!batchLoader.hasSeenRecord()) {
            log.info("{}: The loader is still catching up because we have not loaded a controller record as of offset " +
                    offset + " and high water mark is {}", where, highWaterMark.getAsLong());
            return true;
        }
        log.info("{}: The loader finished catching up to the current high water mark of {}",
                where, highWaterMark.getAsLong());
        catchingUp = false;
        return false;
    }
The event queue which runs this loader.
metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java