public void replay(RangeRecord record) {
        long streamId = record.streamId();
        StreamRuntimeMetadata streamMetadata = this.streamsMetadata.get(streamId);
        if (streamMetadata == null) {
            // should not happen
            log.error("streamId={} not exist when replay range record {}", streamId, record);
            return;
        }
        RangeMetadata rangeMetadata = RangeMetadata.of(record);

        streamMetadata.ranges().put(record.rangeIndex(), rangeMetadata);

        // When load from image the ranges are not orderly replayed.
        boolean islastRange = rangeMetadata.rangeIndex() == streamMetadata.currentRangeIndex();
        // The stream trim also generate RangeRecord
        if (islastRange) {
            Integer lastNodeId = stream2node.get(streamId);
            if (lastNodeId == null || lastNodeId != record.nodeId()) {
                node2streams.compute(record.nodeId(), (k, v) -> {
                    if (v == null) {
                        v = new DeltaList<>();
                    } else {
                        v = v.copy();
                    }
                    v.add(streamId);
                    return v;
                });
                if (lastNodeId != null) {
                    node2streams.compute(lastNodeId, (k, v) -> {
                        if (v == null) {
                            return null;
                        } else {
                            v = v.copy();
                        }
                        v.remove(m -> m == streamId);
                        return v;
                    });
                }
            }
            stream2node.put(streamId, rangeMetadata.nodeId());
        }
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java