public void replay(S3StreamSetObjectRecord record) {
        long objectId = record.objectId();
        int nodeId = record.nodeId();
        long orderId = record.orderId();
        long dataTs = record.dataTimeInMs();
        NodeRuntimeMetadata nodeRuntimeMetadata = this.nodesMetadata.get(nodeId);
        if (nodeRuntimeMetadata == null) {
            // should not happen
            log.error("nodeId={} not exist when replay stream set object record {}", nodeId, record);
            return;
        }
        S3StreamSetObject s3StreamSetObject = new S3StreamSetObject(objectId, nodeId, record.ranges(), orderId, dataTs);
        nodeRuntimeMetadata.streamSetObjects().put(objectId, s3StreamSetObject);

        // update range
        s3StreamSetObject.offsetRangeList().forEach(index -> {
            long streamId = index.streamId();
            StreamRuntimeMetadata metadata = this.streamsMetadata.get(streamId);
            if (metadata == null) {
                // ignore it, the stream may be deleted
                return;
            }
            // the offset continuous is ensured by the process layer
            // when replay from checkpoint, the record may be out of order, so we need to update the end offset to the largest end offset.
            metadata.endOffset(index.endOffset());
        });
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java