|
|
|
@ -54,23 +54,25 @@ public class GeneralEdgeEventFetcher implements EdgeEventFetcher { |
|
|
|
log.trace("[{}] Finding general edge events [{}], seqIdStart = {}, pageLink = {}", |
|
|
|
tenantId, edge.getId(), seqIdStart, pageLink); |
|
|
|
PageData<EdgeEvent> 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<>(); |
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|