protected Optional<Boolean> checkConnectorActiveTopics(String connectorName, Collection<String> topics) {
        try {
            ActiveTopicsInfo info = connect.connectorTopics(connectorName);
            boolean result = info != null
                    && topics.size() == info.topics().size()
                    && topics.containsAll(info.topics());
            log.debug("Found connector {} using topics: {}", connectorName, info.topics());
            return Optional.of(result);
        } catch (Exception e) {
            log.error("Could not check connector {} state info.", connectorName, e);
            return Optional.empty();
        }
    }
Check whether a connector's set of active topics matches the given collection of topic names.

@param connectorName the connector name
@param topics a collection of topics to compare against
@return true if the connector's active topics matches the given collection; false otherwise
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/ConnectAssertions.java