@Test
    public void testFailToStartWhenInternalTopicsAreNotCompacted() throws InterruptedException {
        // Change the broker default cleanup policy to something Connect doesn't like
        brokerProps.put("log." + TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
        // Start out using the improperly configured topics
        workerProps.put(DistributedConfig.CONFIG_TOPIC_CONFIG, "bad-config");
        workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "bad-offset");
        workerProps.put(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, "bad-status");
        workerProps.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG, "1");
        workerProps.put(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG, "1");
        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 but not Connect
        log.info("Starting {} Kafka brokers, but no Connect workers yet", numBrokers);
        connect.start();
        log.info("Completed startup of {} Kafka broker. Expected Connect worker to fail", numBrokers);

        // Create the good topics
        connect.kafka().createTopic("good-config", 1, 1, compactCleanupPolicy());
        connect.kafka().createTopic("good-offset", 1, 1, compactCleanupPolicy());
        connect.kafka().createTopic("good-status", 1, 1, compactCleanupPolicy());

        // Create the poorly-configured topics
        connect.kafka().createTopic("bad-config", 1, 1, deleteCleanupPolicy());
        connect.kafka().createTopic("bad-offset", 1, 1, compactAndDeleteCleanupPolicy());
        connect.kafka().createTopic("bad-status", 1, 1, noTopicSettings());

        // Check the topics
        log.info("Verifying the internal topics for Connect were manually created");
        connect.assertions().assertTopicsExist("good-config", "good-offset", "good-status", "bad-config", "bad-offset", "bad-status");

        // Try to start one worker, with three bad topics
        WorkerHandle worker = connect.addWorker(); // should have failed to start before returning
        assertFalse(connect.isHealthy(worker));
        assertFalse(connect.allWorkersHealthy());
        assertFalse(connect.anyWorkersHealthy());
        connect.removeWorker(worker);

        // We rely upon the fact that we can change the worker properties before the workers are started
        workerProps.put(DistributedConfig.CONFIG_TOPIC_CONFIG, "good-config");

        // Try to start one worker, with two bad topics remaining
        worker = connect.addWorker(); // should have failed to start before returning
        assertFalse(connect.isHealthy(worker));
        assertFalse(connect.allWorkersHealthy());
        assertFalse(connect.anyWorkersHealthy());
        connect.removeWorker(worker);

        // We rely upon the fact that we can change the worker properties before the workers are started
        workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "good-offset");

        // Try to start one worker, with one bad topic remaining
        worker = connect.addWorker(); // should have failed to start before returning
        assertFalse(connect.isHealthy(worker));
        assertFalse(connect.allWorkersHealthy());
        assertFalse(connect.anyWorkersHealthy());
        connect.removeWorker(worker);
        // We rely upon the fact that we can change the worker properties before the workers are started
        workerProps.put(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, "good-status");

        // Try to start one worker, now using all good internal topics
        connect.addWorker();
        connect.assertions().assertAtLeastNumWorkersAreUp(1, "Worker did not start in time.");
    }
Integration test for the creation of internal topics.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java