@ParameterizedTest
    @CsvSource(value = {"false,false", "false,true", "true,false", "true,true"})
    public void testPublisherCannotBeInstalledMoreThanOnce(
        boolean loadSnapshot,
        boolean sameObject
    ) throws Exception {
        MockFaultHandler faultHandler =
                new MockFaultHandler("testPublisherCannotBeInstalledMoreThanOnce");
        MockPublisher publisher = new MockPublisher();
        try (MetadataLoader loader = new MetadataLoader.Builder().
                setFaultHandler(faultHandler).
                setHighWaterMarkAccessor(() -> OptionalLong.of(0L)).
                build()) {
            loader.installPublishers(singletonList(publisher)).get();
            if (loadSnapshot) {
                MockSnapshotReader snapshotReader = new MockSnapshotReader(
                    new MetadataProvenance(200, 100, 4000),
                    singletonList(
                        Batch.control(
                            200,
                            100,
                            4000,
                            10,
                            singletonList(new ControlRecord(ControlRecordType.SNAPSHOT_HEADER, new SnapshotHeaderRecord()))
                        )
                    )
                );
                loader.handleLoadSnapshot(snapshotReader);
                TestUtils.retryOnExceptionWithTimeout(30_000, () ->
                    assertEquals(1L, loader.metrics().handleLoadSnapshotCount())
                );
            } else {
                TestUtils.retryOnExceptionWithTimeout(30_000, () ->
                    assertEquals(0L, loader.metrics().handleLoadSnapshotCount())
                );
            }
            loader.waitForAllEventsToBeHandled();
            if (sameObject) {
                assertEquals("testPublisherCannotBeInstalledMoreThanOnce: Attempted to install " +
                    "publisher MockPublisher, which is already installed.",
                        assertThrows(ExecutionException.class,
                                () -> loader.installPublishers(singletonList(publisher)).get()).
                                getCause().getMessage());
            } else {
                assertEquals("testPublisherCannotBeInstalledMoreThanOnce: Attempted to install " +
                    "a new publisher named MockPublisher, but there is already a publisher with that name.",
                        assertThrows(ExecutionException.class,
                                () -> loader.installPublishers(singletonList(new MockPublisher())).get()).
                                getCause().getMessage());
            }
        }
    }
Test that a publisher cannot be installed more than once.
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java