@Test
    public void testCreateInternalTopicsWithDefaultSettings() throws InterruptedException {
        int numWorkers = 1;
        int numBrokers = 3;
        connect = new EmbeddedConnectCluster.Builder().name("connect-cluster-1")
                                                      .workerProps(workerProps)
                                                      .numWorkers(numWorkers)
                                                      .numBrokers(numBrokers)
                                                      .brokerProps(brokerProps)
                                                      .build();

        // Start the Connect cluster
        connect.start();
        log.info("Completed startup of {} Kafka brokers and {} Connect workers", numBrokers, numWorkers);

        // Check the topics
        log.info("Verifying the internal topics for Connect");
        connect.assertions().assertTopicsExist(configTopic(), offsetTopic(), statusTopic());
        assertInternalTopicSettings();

        // Remove the Connect worker
        log.info("Stopping the Connect worker");
        connect.removeWorker();

        // And restart
        log.info("Starting the Connect worker");
        connect.startConnect();

        // Check the topics
        log.info("Verifying the internal topics for Connect");
        connect.assertions().assertTopicsExist(configTopic(), offsetTopic(), statusTopic());
        assertInternalTopicSettings();
    }
Integration test for the creation of internal topics.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java