@Test
    public void testFailToCreateInternalTopicsWithMoreReplicasThanBrokers() throws InterruptedException {
        workerProps.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG, "3");
        workerProps.put(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG, "2");
        workerProps.put(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG, "1");
        int numWorkers = 0;
        int numBrokers = 1;
        connect = new EmbeddedConnectCluster.Builder().name("connect-cluster-1")
                                                      .workerProps(workerProps)
                                                      .numWorkers(numWorkers)
                                                      .numBrokers(numBrokers)
                                                      .brokerProps(brokerProps)
                                                      .build();

        // Start the brokers and Connect, but Connect should fail to create config and offset topic
        connect.start();
        log.info("Completed startup of {} Kafka broker. Expected Connect worker to fail", numBrokers);

        // Try to start a worker
        WorkerHandle worker = connect.addWorker();

        connect.requestTimeout(1000);
        try (Response response = connect.healthCheck(worker)) {
            assertEquals(Response.Status.SERVICE_UNAVAILABLE.getStatusCode(), response.getStatus());
            assertNotNull(response.getEntity());
            String body = response.getEntity().toString();
            assertTrue(
                    body.contains("The worker is currently initializing and reading to the end of internal topics"),
                    "Body did not contain expected message detailing the worker's in-progress operation: " + body
            );
        }

        connect.resetRequestTimeout();

        //Synchronously await and verify that the worker fails during startup
        Future<?> herderTask = worker.herderTask();
        assertThrows(
                ExecutionException.class,
                () -> herderTask.get(1, TimeUnit.MINUTES)
        );

        // Verify that the offset and config topic don't exist;
        // the status topic may have been created if timing was right, but we don't care
        log.info("Verifying the internal topics for Connect");
        connect.assertions().assertTopicsDoNotExist(configTopic(), offsetTopic());
    }
Integration test for the creation of internal topics.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java