private long getAverageDataVolume(InlongStreamInfo streamInfo) {
        // Since the audit module uses local time, we need to use ZonedDateTime to get the current time
        String dataTimeZone = streamInfo.getSourceList().get(0).getDataTimeZone();

        // This regex pattern matches a time zone offset in the format of "GMT+/-X:00"
        // where X is a single digit (e.g., "GMT+8:00"). The pattern captures the "+" or "-" sign
        // and the single digit, then it replaces the single digit with two digits by adding a "0" in front of it.
        // For example, "GMT+8:00" becomes "GMT+08:00" in order to match standard offset-based ZoneId.
        dataTimeZone = dataTimeZone.replaceAll(TIMEZONE_REGEX, "$10$2:");
        ZoneId dataZone = ZoneId.of(dataTimeZone);

        ZonedDateTime endTime = ZonedDateTime.now(dataZone);
        ZonedDateTime startTime = endTime.minusHours(DATA_SCALE_COUNTING_RANGE_IN_HOURS);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern(AUDIT_QUERY_DATE_TIME_FORMAT);

        // counting data volume on with DATA_PROXY_OUTPUT auditId
        int auditId = AuditIdEnum.getAuditId(DEFAULT_AUDIT_TYPE, DEFAULT_FLOWTYPE).getValue();
        StringJoiner urlParameters = new StringJoiner(AMPERSAND)
                .add(PARAMS_START_TIME + EQUAL + startTime.format(formatter))
                .add(PARAMS_END_TIME + EQUAL + endTime.format(formatter))
                .add(PARAMS_INLONG_GROUP_ID + EQUAL + streamInfo.getInlongGroupId())
                .add(PARAMS_INLONG_STREAM_ID + EQUAL + streamInfo.getInlongStreamId())
                .add(PARAMS_AUDIT_ID + EQUAL + auditId)
                .add(PARAMS_AUDIT_CYCLE + EQUAL + AUDIT_CYCLE_REALTIME);

        String url = auditQueryUrl + DEFAULT_API_MINUTES_PATH + QUESTION_MARK + urlParameters;

        return getAverageDataVolumeFromAuditInfo(url);
    }
Get average data volume on the scale specified by DATA_SCALE_COUNTING_RANGE_IN_HOURS

@param streamInfo inlong stream info
@return The average data count per hour
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkParallelismOptimizer.java