@Test
    public void testSwitchingToTopicCreationEnabled() throws InterruptedException {
        workerProps.put(TOPIC_CREATION_ENABLE_CONFIG, String.valueOf(false));
        connect = connectBuilder.build();
        // start the clusters
        connect.start();

        connect.kafka().createTopic(BAR_TOPIC, DEFAULT_PARTITIONS, DEFAULT_REPLICATION_FACTOR, Collections.emptyMap());

        connect.assertions().assertTopicsExist(BAR_TOPIC);
        connect.assertions().assertTopicSettings(BAR_TOPIC, DEFAULT_REPLICATION_FACTOR,
                DEFAULT_PARTITIONS, "Topic " + BAR_TOPIC + " does not have the expected settings");

        Map<String, String> barProps = defaultSourceConnectorProps(BAR_TOPIC);
        // start a source connector with topic creation properties
        connect.configureConnector(BAR_CONNECTOR, barProps);
        barProps.put(NAME_CONFIG, BAR_CONNECTOR);

        Map<String, String> fooProps = sourceConnectorPropsWithGroups(FOO_TOPIC);
        // start a source connector without topic creation properties
        connect.configureConnector(FOO_CONNECTOR, fooProps);
        fooProps.put(NAME_CONFIG, FOO_CONNECTOR);

        connect.assertions().assertExactlyNumErrorsOnConnectorConfigValidation(fooProps.get(CONNECTOR_CLASS_CONFIG), fooProps, 0,
                "Validating connector configuration produced an unexpected number or errors.");

        connect.assertions().assertExactlyNumErrorsOnConnectorConfigValidation(barProps.get(CONNECTOR_CLASS_CONFIG), barProps, 0,
                "Validating connector configuration produced an unexpected number or errors.");

        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(FOO_CONNECTOR, NUM_TASKS,
                "Connector tasks did not start in time.");

        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(BAR_CONNECTOR, NUM_TASKS,
                "Connector tasks did not start in time.");

        connect.assertions().assertTopicsExist(BAR_TOPIC);
        connect.assertions().assertTopicSettings(BAR_TOPIC, DEFAULT_REPLICATION_FACTOR,
                DEFAULT_PARTITIONS, "Topic " + BAR_TOPIC + " does not have the expected settings");

        connect.assertions().assertTopicsDoNotExist(FOO_TOPIC);

        connect.healthyWorkers().forEach(w -> connect.removeWorker(w));

        workerProps.put(TOPIC_CREATION_ENABLE_CONFIG, String.valueOf(true));

        IntStream.range(0, 3).forEach(i -> connect.addWorker());
        connect.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS, "Workers did not start in time after cluster was rolled.");

        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(FOO_CONNECTOR, NUM_TASKS,
                "Connector tasks did not start in time.");

        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(BAR_CONNECTOR, NUM_TASKS,
                "Connector tasks did not start in time.");

        connect.assertions().assertTopicsExist(FOO_TOPIC);
        connect.assertions().assertTopicSettings(FOO_TOPIC, FOO_GROUP_REPLICATION_FACTOR,
                FOO_GROUP_PARTITIONS, "Topic " + FOO_TOPIC + " does not have the expected settings");
        connect.assertions().assertTopicsExist(BAR_TOPIC);
        connect.assertions().assertTopicSettings(BAR_TOPIC, DEFAULT_REPLICATION_FACTOR,
                DEFAULT_PARTITIONS, "Topic " + BAR_TOPIC + " does not have the expected settings");
    }
Integration test for source connectors with a focus on topic creation with custom properties by
the connector tasks.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/SourceConnectorsIntegrationTest.java