private Errors nodeEpochCheck(int nodeId, long nodeEpoch, boolean checkFailover) {
        NodeRuntimeMetadata nodeRuntimeMetadata = this.nodesMetadata.get(nodeId);
        if (nodeRuntimeMetadata == null) {
            // should not happen
            log.error("[NodeEpochCheck]: nodeId={} not exist when check node epoch", nodeId);
            return Errors.NODE_EPOCH_NOT_EXIST;
        }
        if (nodeRuntimeMetadata.getNodeEpoch() > nodeEpoch) {
            log.warn("[NodeEpochCheck]: nodeId={}'s epoch={} is larger than request epoch={}", nodeId,
                this.nodesMetadata.get(nodeId).getNodeEpoch(), nodeEpoch);
            return Errors.NODE_EPOCH_EXPIRED;
        }
        if (checkFailover && nodeRuntimeMetadata.getFailoverMode()) {
            log.warn("[NodeEpochCheck]: nodeId={} epoch={} is fenced", nodeId, nodeEpoch);
            return Errors.NODE_FENCED;
        }
        return Errors.NONE;
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java