@Test
    public void testTopicsAreCreatedWhenAutoCreateTopicsIsEnabledAtTheBroker() throws InterruptedException {
        brokerProps.put("auto.create.topics.enable", String.valueOf(true));
        workerProps.put(TOPIC_CREATION_ENABLE_CONFIG, String.valueOf(false));
        connect = connectBuilder.brokerProps(brokerProps).workerProps(workerProps).build();
        // start the clusters
        connect.start();

        Map<String, String> fooProps = sourceConnectorPropsWithGroups(FOO_TOPIC);

        // start a source connector
        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().assertConnectorAndAtLeastNumTasksAreRunning(FOO_CONNECTOR, NUM_TASKS,
                "Connector tasks did not start in time.");

        connect.assertions().assertTopicsExist(FOO_TOPIC);
        connect.assertions().assertTopicSettings(FOO_TOPIC, DEFAULT_REPLICATION_FACTOR,
                DEFAULT_PARTITIONS, "Topic " + FOO_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