private void resetActiveTopics(Collection<String> connectors, Collection<ConnectorTaskId> tasks) {
            String stageDescription = "resetting the list of active topics for " + connectors.size() + " and " + tasks.size() + " tasks";
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                connectors.stream()
                        .filter(connectorName -> !configState.contains(connectorName))
                        .forEach(DistributedHerder.this::resetConnectorActiveTopics);
                tasks.stream()
                        .map(ConnectorTaskId::connector)
                        .distinct()
                        .filter(connectorName -> !configState.contains(connectorName))
                        .forEach(DistributedHerder.this::resetConnectorActiveTopics);
            }
        }
Execute the given action and subsequent callback immediately if the current thread is the herder's tick thread,
or use them to create and store a {@link DistributedHerderRequest} on the request queue and return the resulting request
if not.
@param action the action that should be run on the herder's tick thread
@param callback the callback that should be invoked once the action is complete
@return a new {@link DistributedHerderRequest} if one has been created and added to the request queue, and {@code null} otherwise
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java