@Test
    public void testGetActiveTopics() throws InterruptedException {
        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);

        connect.assertions().assertConnectorActiveTopics(FOO_CONNECTOR, Collections.emptyList(),
                "Active topic set is not empty for connector: " + FOO_CONNECTOR);

        // 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.");

        connect.assertions().assertConnectorActiveTopics(FOO_CONNECTOR, Collections.singletonList(FOO_TOPIC),
                "Active topic set is not: " + Collections.singletonList(FOO_TOPIC) + " for connector: " + FOO_CONNECTOR);

        // start another source connector
        connect.configureConnector(BAR_CONNECTOR, defaultSourceConnectorProps(BAR_TOPIC));

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

        connect.assertions().assertConnectorActiveTopics(BAR_CONNECTOR, Collections.singletonList(BAR_TOPIC),
                "Active topic set is not: " + Collections.singletonList(BAR_TOPIC) + " for connector: " + BAR_CONNECTOR);

        // start a sink connector
        connect.configureConnector(SINK_CONNECTOR, defaultSinkConnectorProps(FOO_TOPIC, BAR_TOPIC));

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

        connect.assertions().assertConnectorActiveTopics(SINK_CONNECTOR, Arrays.asList(FOO_TOPIC, BAR_TOPIC),
                "Active topic set is not: " + Arrays.asList(FOO_TOPIC, BAR_TOPIC) + " for connector: " + SINK_CONNECTOR);

        // deleting a connector resets its active topics
        connect.deleteConnector(BAR_CONNECTOR);

        connect.assertions().assertConnectorDoesNotExist(BAR_CONNECTOR,
                "Connector wasn't deleted in time.");

        connect.assertions().assertConnectorActiveTopics(BAR_CONNECTOR, Collections.emptyList(),
                "Active topic set is not empty for deleted connector: " + BAR_CONNECTOR);

        // Unfortunately there's currently no easy way to know when the consumer caught up with
        // the last records that the producer of the stopped connector managed to produce.
        // Repeated runs show that this amount of time is sufficient for the consumer to catch up.
        Thread.sleep(5000);

        // reset active topics for the sink connector after one of the topics has become idle
        connect.resetConnectorTopics(SINK_CONNECTOR);

        connect.assertions().assertConnectorActiveTopics(SINK_CONNECTOR, Collections.singletonList(FOO_TOPIC),
                "Active topic set is not: " + Collections.singletonList(FOO_TOPIC) + " for connector: " + SINK_CONNECTOR);
    }
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