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

        List<StatData> maxAuditVersion = filterMaxAuditVersion(statDataList);
        // If querying for wildcard stream ID, aggregate the data
        if (WILDCARD_STREAM_ID.equals(inlongStreamId)) {
            return AuditUtils.aggregateStatData(maxAuditVersion, WILDCARD_STREAM_ID);
        }
        // Otherwise return the filtered data directly
        return maxAuditVersion;
    }
Query the audit data of log time.

@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