@Test
    public void testCatchingUpState() throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("testLastAppliedOffset");
        List<MockPublisher> publishers = asList(new MockPublisher("a"),
                new MockPublisher("b"));
        AtomicReference<OptionalLong> highWaterMark = new AtomicReference<>(OptionalLong.empty());
        try (MetadataLoader loader = new MetadataLoader.Builder().
                setFaultHandler(faultHandler).
                setHighWaterMarkAccessor(highWaterMark::get).
                build()) {
            loader.installPublishers(publishers).get();
            loadTestSnapshot(loader, 200);

            // We don't update lastAppliedOffset because we're still in catchingUp state due to
            // highWaterMark being OptionalLong.empty (aka unknown).
            assertEquals(-1L, loader.lastAppliedOffset());
            assertFalse(publishers.get(0).firstPublish.isDone());

            // This still doesn't advance lastAppliedOffset since the high watermark at 221
            // is greater than our snapshot at 210.
            highWaterMark.set(OptionalLong.of(221));
            loadTestSnapshot(loader, 210);
            assertEquals(-1L, loader.lastAppliedOffset());

            // Loading a test snapshot at 220 allows us to leave catchUp state.
            loadTestSnapshot(loader, 220);
            assertEquals(220L, loader.lastAppliedOffset());
            publishers.get(0).firstPublish.get(1, TimeUnit.MINUTES);
        }
        faultHandler.maybeRethrowFirstException();
    }
Test that we do not leave the catchingUp state until we have loaded up to the high
watermark.
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java