dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Edge Misordering Compensation Config

Browse Source 
- Moved misordering compensation to configuration for better control of edge event handling and data consistency
This commit is contained in:
Yevhenii 2025-07-29 17:07:58 +03:00
parent 320e0b674e
commit b191c2a2ec
4 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeEventStorageSettings.java
View File

@ -29,4 +29,6 @@ public class EdgeEventStorageSettings {
private long noRecordsSleepInterval;
@Value("${edges.storage.sleep_between_batches}")
private long sleepIntervalBetweenBatches;
@Value("${edges.storage.misordering_compensation_millis:3600000}")
private long misorderingCompensationMillis;
}

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -589,7 +589,7 @@ public abstract class EdgeGrpcSession implements Closeable {
previousStartSeqId,
false,
Integer.toUnsignedLong(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount()),
ctx.getEdgeEventService());
ctx.getEdgeEventService(), ctx.getEdgeEventStorageSettings().getMisorderingCompensationMillis());
log.trace("[{}][{}] starting processing edge events, previousStartTs = {}, previousStartSeqId = {}",
tenantId, edge.getId(), previousStartTs, previousStartSeqId);
Futures.addCallback(startProcessingEdgeEvents(fetcher), new FutureCallback<>() {

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/GeneralEdgeEventFetcher.java
View File

@ -26,13 +26,9 @@ import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.page.TimePageLink;
import org.thingsboard.server.dao.edge.EdgeEventService;
import java.util.concurrent.TimeUnit;
@AllArgsConstructor
@Slf4j
public class GeneralEdgeEventFetcher implements EdgeEventFetcher {
// Subtract from queueStartTs to ensure no data is lost due to potential misordering of edge events by created_time.
private static final long MISORDERING_COMPENSATION_MILLIS = TimeUnit.SECONDS.toMillis(60);
private final Long queueStartTs;
private Long seqIdStart;
@ -40,6 +36,8 @@ public class GeneralEdgeEventFetcher implements EdgeEventFetcher {
private boolean seqIdNewCycleStarted;
private Long maxReadRecordsCount;
private final EdgeEventService edgeEventService;
// Subtract from queueStartTs to ensure no data is lost due to potential misordering of edge events by created_time.
private final long misorderingCompensationMillis;
@Override
public PageLink getPageLink(int pageSize) {
@ -48,7 +46,7 @@ public class GeneralEdgeEventFetcher implements EdgeEventFetcher {
0,
null,
null,
queueStartTs > 0 ? queueStartTs - MISORDERING_COMPENSATION_MILLIS : 0,
queueStartTs > 0 ? queueStartTs - misorderingCompensationMillis : 0,
System.currentTimeMillis());
}

3
application/src/main/resources/thingsboard.yml
View File

@ -1480,6 +1480,9 @@ edges:
no_read_records_sleep: "${EDGES_NO_READ_RECORDS_SLEEP:1000}"
# Number of milliseconds to wait before resending failed batch of edge events to edge
sleep_between_batches: "${EDGES_SLEEP_BETWEEN_BATCHES:60000}"
# Number of milliseconds to subtract from queue start timestamp to compensate for potential
# misordering of edge events by created_time. Prevents skipping early events.
misordering_compensation_millis: "${EDGES_MISORDERING_COMPENSATION_MILLIS:3600000}"
# Max number of high priority edge events per edge session. No persistence - stored in memory
max_high_priority_queue_size_per_session: "${EDGES_MAX_HIGH_PRIORITY_QUEUE_SIZE_PER_SESSION:10000}"
# Number of threads that are used to check DB for edge events