public void replay(RemoveKVRecord kvRecord) {
        for (String key : kvRecord.keys()) {
            if (!key.startsWith(KEY_PREFIX)) {
                continue;
            }
            try {
                int nodeId = Integer.parseInt(key.substring(KEY_PREFIX.length()));
                nodeMetadataMap.remove(nodeId);
                nodeRuntimeInfoManager.unlock(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