private void fillMetricItemsByEvent(Event event, boolean isSource,
            boolean isReadOp, boolean isSuccess,
            long size, long sendTime) {
        Map<String, String> dimensions = new HashMap<>();
        dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID, clusterId);
        dimensions.put(DataProxyMetricItem.KEY_INLONG_GROUP_ID,
                event.getHeaders().get(AttributeConstants.GROUP_ID));
        dimensions.put(DataProxyMetricItem.KEY_INLONG_STREAM_ID,
                event.getHeaders().get(AttributeConstants.STREAM_ID));
        long dataTime = NumberUtils.toLong(
                event.getHeaders().get(AttributeConstants.DATA_TIME));
        long msgCount = NumberUtils.toLong(
                event.getHeaders().get(ConfigConstants.MSG_COUNTER_KEY));
        long auditFormatTime =
                dataTime - dataTime % CommonConfigHolder.getInstance().getAuditFormatInvlMs();
        dimensions.put(DataProxyMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
        if (isSource) {
            dimensions.put(DataProxyMetricItem.KEY_SOURCE_ID, name);
            dimensions.put(DataProxyMetricItem.KEY_SOURCE_DATA_ID, sourceDataId);
        } else {
            dimensions.put(DataProxyMetricItem.KEY_SINK_ID, name);
            dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID,
                    event.getHeaders().get(ConfigConstants.TOPIC_KEY));
        }
        DataProxyMetricItem metricItem = findMetricItem(dimensions);
        if (isReadOp) {
            if (isSuccess) {
                metricItem.readSuccessCount.addAndGet(msgCount);
                metricItem.readSuccessSize.addAndGet(size);
            } else {
                metricItem.readFailCount.addAndGet(msgCount);
                metricItem.readFailSize.addAndGet(size);
            }
        } else {
            if (isSuccess) {
                metricItem.sendSuccessCount.addAndGet(msgCount);
                metricItem.sendSuccessSize.addAndGet(size);
                if (sendTime > 0) {
                    long currentTime = System.currentTimeMillis();
                    long msgDataTimeL = Long.parseLong(
                            event.getHeaders().get(AttributeConstants.DATA_TIME));
                    long msgRcvTimeL = Long.parseLong(
                            event.getHeaders().get(AttributeConstants.RCV_TIME));
                    metricItem.sinkDuration.addAndGet(currentTime - sendTime);
                    metricItem.nodeDuration.addAndGet(currentTime - msgRcvTimeL);
                    metricItem.wholeDuration.addAndGet(currentTime - msgDataTimeL);
                }
            } else {
                metricItem.sendFailCount.addAndGet(msgCount);
                metricItem.sendFailSize.addAndGet(size);
            }
            metricItem.sendCount.addAndGet(msgCount);
            metricItem.sendSize.addAndGet(size);
        }
    }
Fill metric items by event

@param event    the event object
@param isSource   whether source part
@param isReadOp   whether read operation
@param isSuccess  whether success read or send
@param size       the message size
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/DataProxyMetricItemSet.java