@Override
    public ClusterConfigState snapshot() {
        synchronized (lock) {
            // Only a shallow copy is performed here; in order to avoid accidentally corrupting the worker's view
            // of the config topic, any nested structures should be copied before making modifications
            return new ClusterConfigState(
                    offset,
                    sessionKey,
                    new HashMap<>(connectorTaskCounts),
                    new HashMap<>(connectorConfigs),
                    new HashMap<>(connectorTargetStates),
                    new HashMap<>(taskConfigs),
                    new HashMap<>(connectorTaskCountRecords),
                    new HashMap<>(connectorTaskConfigGenerations),
                    new HashMap<>(appliedConnectorConfigs),
                    new HashSet<>(connectorsPendingFencing),
                    new HashSet<>(inconsistent),
                    configTransformer
            );
        }
    }
Get a snapshot of the current state of the cluster.
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java