protected ConnectMetrics.LiteralSupplier<Long> taskStatusCounter(String connName, TaskStatus.State state) {
            return now -> tasks.values()
                .stream()
                .filter(task ->
                    task.id().connector().equals(connName) &&
                    herder.taskStatus(task.id()).state().equalsIgnoreCase(state.toString()))
                .count();
        }
Gives a best-effort guess for whether the given offsets topic is the same topic as the worker-global offsets topic.
Even if the name of the topic is the same as the name of the worker's offsets topic, the two may still be different topics
if the connector is configured to produce to a different Kafka cluster than the one that hosts the worker's offsets topic.
@param offsetsTopic the name of the offsets topic for the connector
@param producerProps the producer configuration for the connector
@return whether it appears that the connector's offsets topic is the same topic as the worker-global offsets topic.
If {@code true}, it is guaranteed that the two are the same;
if {@code false}, it is likely but not guaranteed that the two are not the same
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java