private long parseResponseAndCalculateAverageDataVolume(CloseableHttpResponse response) throws IOException {
        HttpEntity entity = response.getEntity();
        if (entity == null) {
            log.warn("Empty response entity from audit API, returning default count.");
            return DEFAULT_ERROR_DATA_VOLUME;
        }

        String responseString = EntityUtils.toString(entity);
        log.debug("Flink dynamic parallelism optimizer got response from audit API: {}", responseString);

        JsonObject jsonObject = JsonParser.parseString(responseString).getAsJsonObject();
        AuditInfo[] auditInfoArray = new Gson().fromJson(jsonObject.getAsJsonArray("data"), AuditInfo[].class);

        ZonedDateTime minLogTs = null;
        ZonedDateTime maxLogTs = null;
        DateTimeFormatter logTsFormatter =
                DateTimeFormatter.ofPattern(LOGTS_DATE_TIME_FORMAT).withZone(ZoneId.systemDefault());
        long totalCount = 0L;
        for (AuditInfo auditData : auditInfoArray) {
            if (auditData != null) {
                ZonedDateTime logTs = ZonedDateTime.parse(auditData.getLogTs(), logTsFormatter);
                if (minLogTs == null || logTs.isBefore(minLogTs)) {
                    minLogTs = logTs;
                }
                if (maxLogTs == null || logTs.isAfter(maxLogTs)) {
                    maxLogTs = logTs;
                }
                log.debug("parsed AuditInfo, Count: {}, Size: {}", auditData.getCount(), auditData.getSize());
                totalCount += auditData.getCount();
            } else {
                log.error("Null AuditInfo found in response data.");
            }
        }

        if (minLogTs != null && maxLogTs != null) {
            long timeDifferenceInSeconds = maxLogTs.toEpochSecond() - minLogTs.toEpochSecond();
            log.info("Time difference in seconds: {}", timeDifferenceInSeconds);
            if (timeDifferenceInSeconds > 0) {
                return totalCount / timeDifferenceInSeconds;
            }
        }
        return DEFAULT_ERROR_DATA_VOLUME;
    }
Parse the HTTP response and calculate the total count from the audit data.

@param response The HTTP response
@return The total count of the audit data
@throws IOException If an I/O error occurs
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkParallelismOptimizer.java