From 9fdb32da0568c57e2930edbf98fd1a591a4c50ea Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 10 Jun 2026 17:59:08 +0300 Subject: [PATCH] refactor: tidy edge disconnect-notification per review --- .../service/edge/rpc/EdgeGrpcService.java | 31 +++++++++---------- .../edge/EdgeConnectionNotificationTest.java | 6 ++-- .../service/edge/rpc/EdgeGrpcServiceTest.java | 12 +------ 3 files changed, 19 insertions(+), 30 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 259f63c206..411ccd5f05 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -238,18 +238,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @PreDestroy public void destroy() { - // Flush already-pending disconnect notifications BEFORE shutting the server down. These are edges that - // disconnected on their own within the delay window; without this a graceful restart would silently - // swallow their "disconnected" alert. Snapshotting first means we only flush those - not edges that this - // very shutdown is about to disconnect (they rebalance to another node and shouldn't alert) - and we run - // while the scheduler and rule engine are still alive. The guard inside fireDelayedDisconnectNotification - // still suppresses any edge that reconnected elsewhere. List pendingToFlush = new ArrayList<>(pendingDisconnectNotifications.values()); pendingDisconnectNotifications.clear(); - pendingToFlush.forEach(pending -> { - cancelIfPending(pending.future()); - fireDelayedDisconnectNotification(pending); - }); + pendingToFlush.forEach(this::fireDelayedDisconnectNotification); if (server != null) { server.shutdownNow(); } @@ -590,12 +581,22 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i // TB-Core node within the keep-alive window, that node has overwritten the entry - evicting it here // would wipe the live owner and make fireDelayedDisconnectNotification raise a false 'disconnected'. private void evictServiceIdCacheIfOwnedByThisNode(EdgeId edgeId) { - TbCacheValueWrapper wrapper = edgeIdServiceIdCache.get(edgeId); - if (wrapper != null && serviceInfoProvider.getServiceId().equals(wrapper.get())) { + if (isOwnedByThisNode(edgeId)) { edgeIdServiceIdCache.evict(edgeId); } } + // The edge's service-id cache entry still points at this node, i.e. this node is the recorded owner. + private boolean isOwnedByThisNode(EdgeId edgeId) { + TbCacheValueWrapper wrapper = edgeIdServiceIdCache.get(edgeId); + return wrapper != null && serviceInfoProvider.getServiceId().equals(wrapper.get()); + } + + // The edge has a live owner somewhere in the cluster (the cache is cluster-wide), regardless of which node. + private boolean isConnectedClusterWide(EdgeId edgeId) { + return edgeIdServiceIdCache.get(edgeId) != null; + } + private void destroySession(EdgeGrpcSession session) { try (session) { if (!session.destroy()) { @@ -749,7 +750,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i pendingDisconnectNotifications.remove(edgeId, pending); // Re-verify the edge is still disconnected. The cache is cluster-wide, so this also covers the case // where the edge dropped on this node and reconnected to a different TB-Core node within the window. - if (sessions.containsKey(edgeId) || edgeIdServiceIdCache.get(edgeId) != null) { + if (sessions.containsKey(edgeId) || isConnectedClusterWide(edgeId)) { log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", edge.getTenantId(), edgeId); return; } @@ -763,9 +764,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - // Carries the context the delayed task needs, so a pending notification can still be fired on shutdown - // (see destroy()), not just cancelled. The future is back-filled right after scheduling (see - // scheduleDisconnectNotification). Package-private for unit testing of fireDelayedDisconnectNotification. static final class PendingDisconnect { private final Edge edge; private ScheduledFuture future; @@ -785,6 +783,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i void setFuture(ScheduledFuture future) { this.future = future; } + } private static void cancelIfPending(ScheduledFuture future) { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java index c46d82437a..66023c7372 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -43,7 +43,7 @@ import static org.mockito.Mockito.verify; @DaoSqlTest public class EdgeConnectionNotificationTest extends AbstractEdgeTest { - private static final long DELAY_MS = 5000L; + private static final long DELAY_MS = 1500L; @MockitoSpyBean private NotificationRuleProcessor notificationRuleProcessor; @@ -115,8 +115,8 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { // The "disconnected" notification must never be sent throughout the full delay window. await().during(DELAY_MS + 500, TimeUnit.MILLISECONDS) - .atMost(DELAY_MS + 2000, TimeUnit.MILLISECONDS) - .untilAsserted(() -> verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false)))); + .atMost(DELAY_MS + 2000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false)))); } private ArgumentMatcher edgeConnectionTrigger(boolean connected) { diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java index 86758cb68f..61bb04a803 100644 --- a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java @@ -45,12 +45,6 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -/** - * Unit coverage for the cluster-aware predicates that guard the delayed disconnect notification: - * {@code evictServiceIdCacheIfOwnedByThisNode} and the re-verify check in - * {@code fireDelayedDisconnectNotification}. These exercise the cross-node ownership logic that the - * single-node integration tests ({@code EdgeConnectionNotificationTest}) cannot reach. - */ @ExtendWith(MockitoExtension.class) public class EdgeGrpcServiceTest { @@ -72,21 +66,17 @@ public class EdgeGrpcServiceTest { @InjectMocks private EdgeGrpcService edgeGrpcService; - private TenantId tenantId; private EdgeId edgeId; private Edge edge; @BeforeEach public void setUp() { - tenantId = new TenantId(UUID.randomUUID()); edgeId = new EdgeId(UUID.randomUUID()); edge = new Edge(edgeId); - edge.setTenantId(tenantId); + edge.setTenantId(TenantId.fromUUID(UUID.randomUUID())); edge.setName("test-edge"); } - // --- evictServiceIdCacheIfOwnedByThisNode --- - @Test public void givenCacheOwnedByThisNode_whenEvict_thenEntryIsEvicted() { when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE);