public void replay(NodeWALMetadataRecord record) {
        int nodeId = record.nodeId();
        long nodeEpoch = record.nodeEpoch();
        // already exist, update the node's metadata
        if (this.nodesMetadata.containsKey(nodeId)) {
            NodeRuntimeMetadata nodeRuntimeMetadata = this.nodesMetadata.get(nodeId);
            nodeRuntimeMetadata.setNodeEpoch(nodeEpoch);
            nodeRuntimeMetadata.setFailoverMode(record.failoverMode());
            return;
        }
        // not exist, create a new node
        this.nodesMetadata.put(nodeId, new NodeRuntimeMetadata(nodeId, nodeEpoch, record.failoverMode(), this.snapshotRegistry));
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java