@Test
    public void testTopicTrackingIsDisabled() throws InterruptedException {
        workerProps.put(TOPIC_TRACKING_ENABLE_CONFIG, "false");
        connect = connectBuilder.build();
        // start the clusters
        connect.start();

        // create test topic
        connect.kafka().createTopic(FOO_TOPIC, NUM_TOPIC_PARTITIONS);
        connect.kafka().createTopic(BAR_TOPIC, NUM_TOPIC_PARTITIONS);

        // start a source connector
        connect.configureConnector(FOO_CONNECTOR, defaultSourceConnectorProps(FOO_TOPIC));
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(FOO_CONNECTOR, NUM_TASKS,
                "Connector tasks did not start in time.");

        // resetting active topics for the sink connector won't work when the config is disabled
        Exception e = assertThrows(ConnectRestException.class, () -> connect.resetConnectorTopics(SINK_CONNECTOR));
        assertTrue(e.getMessage().contains("Topic tracking is disabled."));

        e = assertThrows(ConnectRestException.class, () -> connect.connectorTopics(SINK_CONNECTOR));
        assertTrue(e.getMessage().contains("Topic tracking is disabled."));

        // Wait for tasks to produce a few records
        Thread.sleep(5000);

        assertNoTopicStatusInStatusTopic();
    }
Integration test for the endpoints that offer topic tracking of a connector's active
topics.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorTopicsIntegrationTest.java