private void injectInlongMetric(StreamInfo streamInfo) {
        streamInfo.getNodes().stream().filter(node -> node instanceof InlongMetric).forEach(node -> {
            Map<String, String> properties = node.getProperties();
            if (properties == null) {
                properties = new LinkedHashMap<>();
                if (node instanceof LoadNode) {
                    ((LoadNode) node).setProperties(properties);
                } else if (node instanceof ExtractNode) {
                    ((ExtractNode) node).setProperties(properties);
                } else {
                    throw new UnsupportedOperationException(String.format(
                            "Unsupported inlong group stream node for: %s", node.getClass().getSimpleName()));
                }
            }
            properties.put(Constants.METRICS_LABELS.key(),
                    Stream.of(Constants.GROUP_ID + "=" + groupInfo.getGroupId(),
                            Constants.STREAM_ID + "=" + streamInfo.getStreamId(),
                            Constants.NODE_ID + "=" + node.getId())
                            .collect(Collectors.joining("&")));
            // METRICS_AUDIT_PROXY_HOSTS depends on INLONG_GROUP_STREAM_NODE
            if (StringUtils.isNotEmpty(groupInfo.getProperties().get(Constants.METRICS_AUDIT_PROXY_HOSTS.key()))) {
                properties.put(Constants.METRICS_AUDIT_PROXY_HOSTS.key(),
                        groupInfo.getProperties().get(Constants.METRICS_AUDIT_PROXY_HOSTS.key()));
            }
        });
    }
Inject the metric option for ExtractNode or LoadNode

@param streamInfo The encapsulation of nodes and node relations
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java