@Test
    public void testPublishTransactionWithinBatch() throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("testPublishTransactionWithinBatch");
        MockPublisher publisher = new MockPublisher("testPublishTransactionWithinBatch");
        List<MockPublisher> publishers = singletonList(publisher);
        try (MetadataLoader loader = new MetadataLoader.Builder().
                setFaultHandler(faultHandler).
                setHighWaterMarkAccessor(() -> OptionalLong.of(0)).
                build()) {
            loader.installPublishers(publishers).get();
            loader.waitForAllEventsToBeHandled();

            loader.handleCommit(
                MockBatchReader.newSingleBatchReader(500, 100, Arrays.asList(
                    new ApiMessageAndVersion(new BeginTransactionRecord()
                        .setName("txn-1"), (short) 0),
                    new ApiMessageAndVersion(new TopicRecord()
                        .setName("foo")
                        .setTopicId(Uuid.fromString("HQSM3ccPQISrHqYK_C8GpA")), (short) 0),
                    new ApiMessageAndVersion(new EndTransactionRecord(), (short) 0)
                )));
            loader.waitForAllEventsToBeHandled();

            // After MetadataLoader is fixed to handle arbitrary transactions, we would expect "foo"
            // to be visible at this point.
            publisher.firstPublish.get(30, TimeUnit.SECONDS);
            assertNotNull(publisher.latestImage.topics().getTopic("foo"));
        }
        faultHandler.maybeRethrowFirstException();
    }
Test that loading a snapshot clears the previous state.
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java