private static void waitForLastCommittedOffset(long targetOffset,
                LocalLogManager logManager) throws InterruptedException {
        TestUtils.retryOnExceptionWithTimeout(20000, 3, () -> {
            MockMetaLogManagerListener listener =
                (MockMetaLogManagerListener) logManager.listeners().get(0);
            long highestOffset = -1;
            for (String event : listener.serializedEvents()) {
                if (event.startsWith(LAST_COMMITTED_OFFSET)) {
                    long offset = Long.parseLong(
                        event.substring(LAST_COMMITTED_OFFSET.length() + 1));
                    if (offset < highestOffset) {
                        throw new RuntimeException("Invalid offset: " + offset +
                            " is less than the previous offset of " + highestOffset);
                    }
                    highestOffset = offset;
                }
            }
            if (highestOffset < targetOffset) {
                throw new RuntimeException("Offset for log manager " +
                    logManager.nodeId() + " only reached " + highestOffset);
            }
        });
    }
Test that we can pass leadership back and forth between log managers.
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java