ApiMessageAndVersion registerNodeRecord(int nodeId, NodeMetadata newNodeMetadata) {
        KVRecord kvRecord = new KVRecord().setKeyValues(List.of(
            new KVRecord.KeyValue()
                .setKey(KEY_PREFIX + nodeId)
                .setValue(NodeMetadataCodec.encode(newNodeMetadata))
        ));
        return new ApiMessageAndVersion(kvRecord, (short) 0);
    }
Note: It is costly to check if a node has opening streams, so it is recommended to use this method only when necessary.
metadata/src/main/java/org/apache/kafka/controller/stream/NodeControlManager.java