public List<StatData> queryAuditData(String startTime, String endTime,
            String inlongGroupId, String inlongStreamId, String auditId,
            String auditTag, boolean distinct) {
        long currentTime = System.currentTimeMillis();
        List<StatData> statDataList = new CopyOnWriteArrayList<>();
        if (dataSourceList.isEmpty()) {
            return null;
        }
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (DataSource dataSource : dataSourceList) {
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                StatData statDataListTemp =
                        doQueryAuditData(dataSource, startTime, endTime, inlongGroupId, inlongStreamId, auditId,
                                auditTag, distinct);
                if (statDataListTemp != null) {
                    statDataList.add(statDataListTemp);
                }
            }, executor);
            futures.add(future);
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        LOGGER.info("Query audit data by params: {} {} {} {} {}, total cost {} ms", startTime, endTime, inlongGroupId,
                inlongStreamId, auditId, System.currentTimeMillis() - currentTime);
        return statDataList;

    }
Do query report ips.

@param dataSource
@param startTime
@param endTime
@param inlongGroupId
@param inlongStreamId
@param auditId
@return
inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/cache/RealTimeQuery.java