@ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testPublishTransaction(boolean abortTxn) throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("testTransactions");
        MockPublisher publisher = new MockPublisher("testTransactions");
        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("testTransactions"), (short) 0),
                    new ApiMessageAndVersion(new TopicRecord()
                        .setName("foo")
                        .setTopicId(Uuid.fromString("dMCqhcK4T5miGH5wEX7NsQ")), (short) 0)
            )));
            loader.waitForAllEventsToBeHandled();
            publisher.firstPublish.get(30, TimeUnit.SECONDS);
            assertNull(publisher.latestImage.topics().getTopic("foo"),
                "Topic should not be visible since we started transaction");

            loader.handleCommit(
                MockBatchReader.newSingleBatchReader(500, 100, Arrays.asList(
                    new ApiMessageAndVersion(new PartitionRecord()
                        .setTopicId(Uuid.fromString("dMCqhcK4T5miGH5wEX7NsQ"))
                        .setPartitionId(0), (short) 0),
                    new ApiMessageAndVersion(new PartitionRecord()
                        .setTopicId(Uuid.fromString("dMCqhcK4T5miGH5wEX7NsQ"))
                        .setPartitionId(1), (short) 0)
                )));
            loader.waitForAllEventsToBeHandled();
            assertNull(publisher.latestImage.topics().getTopic("foo"),
                "Topic should not be visible after subsequent batch");

            if (abortTxn) {
                loader.handleCommit(
                    MockBatchReader.newSingleBatchReader(500, 100, singletonList(
                        new ApiMessageAndVersion(new AbortTransactionRecord(), (short) 0)
                    )));
                loader.waitForAllEventsToBeHandled();

                assertNull(publisher.latestImage.topics().getTopic("foo"),
                    "Topic should not be visible since the transaction was aborted");
            } else {
                loader.handleCommit(
                    MockBatchReader.newSingleBatchReader(500, 100, singletonList(
                        new ApiMessageAndVersion(new EndTransactionRecord(), (short) 0)
                    )));
                loader.waitForAllEventsToBeHandled();

                assertNotNull(publisher.latestImage.topics().getTopic("foo"),
                    "Topic should be visible now that transaction has ended");
            }
        }
        faultHandler.maybeRethrowFirstException();
    }
Test that loading a snapshot clears the previous state.
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java