@Test
    public void testReloadSnapshot() throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("testLastAppliedOffset");
        List<MockPublisher> publishers = singletonList(new MockPublisher("a"));
        try (MetadataLoader loader = new MetadataLoader.Builder().
                setFaultHandler(faultHandler).
                setHighWaterMarkAccessor(() -> OptionalLong.of(0)).
                build()) {
            loadTestSnapshot(loader, 100);
            loader.installPublishers(publishers).get();
            loader.waitForAllEventsToBeHandled();
            assertTrue(publishers.get(0).firstPublish.isDone());
            assertTrue(publishers.get(0).latestDelta.image().isEmpty());
            assertEquals(100L, publishers.get(0).latestImage.provenance().lastContainedOffset());

            loadTestSnapshot(loader, 200);
            assertEquals(200L, loader.lastAppliedOffset());
            assertEquals(IBP_3_3_IV1.featureLevel(),
                loader.metrics().currentMetadataVersion().featureLevel());
            assertFalse(publishers.get(0).latestDelta.image().isEmpty());

            loadTestSnapshot2(loader, 400);
            assertEquals(400L, loader.lastAppliedOffset());
            assertEquals(IBP_3_3_IV2.featureLevel(),
                loader.metrics().currentMetadataVersion().featureLevel());

            // Make sure the topic in the initial snapshot was overwritten by loading the new snapshot.
            assertFalse(publishers.get(0).latestImage.topics().topicsByName().containsKey("foo"));
            assertTrue(publishers.get(0).latestImage.topics().topicsByName().containsKey("bar"));

            loader.handleCommit(new MockBatchReader(500, singletonList(
                MockBatchReader.newBatch(500, 100, singletonList(
                    new ApiMessageAndVersion(new FeatureLevelRecord().
                        setName(MetadataVersion.FEATURE_NAME).
                        setFeatureLevel(IBP_3_5_IV0.featureLevel()), (short) 0))))));
            loader.waitForAllEventsToBeHandled();
            assertEquals(IBP_3_5_IV0.featureLevel(),
                loader.metrics().currentMetadataVersion().featureLevel());
        }
        faultHandler.maybeRethrowFirstException();
    }
Test that loading a snapshot clears the previous state.
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java