From a315f18195188f7b592dce46d65e4dca90b5606b Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 11 Feb 2025 08:40:40 +0200 Subject: [PATCH] Improved GeneralEdgeEventFetcher for case when seqId less maxRecordsReadSize --- .../rpc/fetch/GeneralEdgeEventFetcher.java | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/GeneralEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/GeneralEdgeEventFetcher.java index b2a1044981..93461f8c39 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/GeneralEdgeEventFetcher.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/GeneralEdgeEventFetcher.java @@ -54,23 +54,25 @@ public class GeneralEdgeEventFetcher implements EdgeEventFetcher { log.trace("[{}] Finding general edge events [{}], seqIdStart = {}, pageLink = {}", tenantId, edge.getId(), seqIdStart, pageLink); PageData edgeEvents = edgeEventService.findEdgeEvents(tenantId, edge.getId(), seqIdStart, null, (TimePageLink) pageLink); - if (edgeEvents.getData().isEmpty()) { + if (!edgeEvents.getData().isEmpty()) { + return edgeEvents; + } + if (seqIdStart > this.maxReadRecordsCount) { edgeEvents = edgeEventService.findEdgeEvents(tenantId, edge.getId(), 0L, Math.max(this.maxReadRecordsCount, seqIdStart - this.maxReadRecordsCount), (TimePageLink) pageLink); if (edgeEvents.getData().stream().anyMatch(ee -> ee.getSeqId() < seqIdStart)) { log.info("[{}] seqId column of edge_event table started new cycle [{}]", tenantId, edge.getId()); this.seqIdNewCycleStarted = true; this.seqIdStart = 0L; - } else { - edgeEvents = new PageData<>(); - log.warn("[{}] unexpected edge notification message received. " + - "no new events found and seqId column of edge_event table doesn't started new cycle [{}]", tenantId, edge.getId()); + return edgeEvents; } } - return edgeEvents; + log.info("[{}] Unexpected edge notification message received. " + + "No new events found, and the seqId column of the edge_event table has not started a new cycle [{}].", tenantId, edge.getId()); + return new PageData<>(); } catch (Exception e) { - log.error("[{}] failed to find edge events [{}]", tenantId, edge.getId()); + log.error("[{}] Failed to find edge events [{}]", tenantId, edge.getId(), e); + return new PageData<>(); } - return new PageData<>(); } }