public void replay(S3StreamEndOffsetsRecord record) {
        for (StreamEndOffset streamEndOffset : S3StreamEndOffsetsCodec.decode(record.endOffsets())) {
            StreamRuntimeMetadata streamMetadata = this.streamsMetadata.get(streamEndOffset.streamId());
            if (streamMetadata == null) {
                // should not happen
                log.error("streamId={} not exist when replay S3StreamEndOffsetsRecord", streamEndOffset.streamId());
                continue;
            }
            streamMetadata.endOffset(streamEndOffset.endOffset());
        }
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java