@Test
    public void testNoPublishEmptyImage() throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("testNoPublishEmptyImage");
        List<MetadataImage> capturedImages = new ArrayList<>();
        CompletableFuture<Void> firstPublish = new CompletableFuture<>();
        MetadataPublisher capturingPublisher = new MetadataPublisher() {
            @Override
            public String name() {
                return "testNoPublishEmptyImage";
            }

            @Override
            public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage, LoaderManifest manifest) {
                capturedImages.add(newImage);
                if (!firstPublish.isDone()) {
                    firstPublish.complete(null);
                }
            }
        };

        try (MetadataLoader loader = new MetadataLoader.Builder().
                setFaultHandler(faultHandler).
                setHighWaterMarkAccessor(() -> OptionalLong.of(1)).
                build()) {
            loader.installPublishers(singletonList(capturingPublisher)).get();
            loader.handleCommit(
                MockBatchReader.newSingleBatchReader(0, 1, singletonList(
                    // Any record will work here
                    new ApiMessageAndVersion(new ConfigRecord()
                        .setResourceType(ConfigResource.Type.BROKER.id())
                        .setResourceName("3000")
                        .setName("foo")
                        .setValue("bar"), (short) 0)
                )));
            firstPublish.get(30, TimeUnit.SECONDS);

            assertFalse(capturedImages.isEmpty());
            capturedImages.forEach(metadataImage ->
                assertFalse(metadataImage.isEmpty())
            );

        }
        faultHandler.maybeRethrowFirstException();
    }
Test that loading a snapshot clears the previous state.
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java