public void replay(S3StreamRecord record) {
        long streamId = record.streamId();
        // already exist, update the stream's metadata
        if (this.streamsMetadata.containsKey(streamId)) {
            StreamRuntimeMetadata streamMetadata = this.streamsMetadata.get(streamId);
            StreamState newState = StreamState.fromByte(record.streamState());
            streamMetadata.startOffset(record.startOffset());
            streamMetadata.currentEpoch(record.epoch());
            streamMetadata.currentRangeIndex(record.rangeIndex());
            streamMetadata.currentState(newState);
            if (streamMetadata.tags().isEmpty() && record.tags().size() > 0) {
                Map<String, String> tags = new HashMap<>();
                record.tags().forEach(tag -> tags.put(tag.key(), tag.value()));
                streamMetadata.setTags(tags);
            }
            return;
        }
        Map<String, String> tags = new HashMap<>();
        record.tags().forEach(tag -> tags.put(tag.key(), tag.value()));
        // not exist, create a new stream
        StreamRuntimeMetadata streamMetadata = new StreamRuntimeMetadata(record.streamId(), record.epoch(), record.rangeIndex(),
            record.startOffset(), StreamState.fromByte(record.streamState()), tags, this.snapshotRegistry);
        this.streamsMetadata.put(streamId, streamMetadata);

    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java