public synchronized void replay(KVRecord kvRecord) {
        for (KVRecord.KeyValue kv : kvRecord.keyValues()) {
            if (!(kv.key() != null && kv.key().startsWith(KEY_PREFIX))) {
                continue;
            }
            try {
                int nodeId = Integer.parseInt(kv.key().substring(KEY_PREFIX.length()));
                NodeMetadata nodeMetadata = NodeMetadataCodec.decode(kv.value());
                nodeMetadataMap.put(nodeId, nodeMetadata);
                if ("CLOSED".equals(nodeMetadata.getTags().getOrDefault("CIRCUIT_BREAKER", "CLOSED"))) {
                    nodeRuntimeInfoManager.unlock(nodeId);
                    lockedNodes.remove(nodeId);
                } else {
                    nodeRuntimeInfoManager.lock(nodeId);
                    lockedNodes.add(nodeId);
                }
            } catch (Throwable e) {
                LOGGER.error("[FATAL] replay NodeMetadata from KV fail", e);
            }
        }
    }
Note: It is costly to check if a node has opening streams, so it is recommended to use this method only when necessary.
metadata/src/main/java/org/apache/kafka/controller/stream/NodeControlManager.java