From 4993ffc66f4749d2465730b498b9f3ea5b5a4131 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Mon, 8 Jun 2026 17:07:28 +0300 Subject: [PATCH] refactor: address PR #15732 review - cancel helper, pooled executor, test coverage --- .../service/edge/rpc/EdgeGrpcService.java | 36 +++++++---- .../src/main/resources/thingsboard.yml | 2 +- .../server/edge/AbstractEdgeTest.java | 14 +++- .../edge/EdgeConnectionNotificationTest.java | 6 +- ...geImmediateDisconnectNotificationTest.java | 64 +++++++++++++++++++ 5 files changed, 99 insertions(+), 23 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java 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 7f57101205..a1910396dc 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 @@ -36,6 +36,7 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.rule.engine.api.AttributesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; +import org.thingsboard.server.cache.TbCacheValueWrapper; import org.thingsboard.server.cache.TbTransactionalCache; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.AttributeScope; @@ -242,11 +243,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i sessionEdgeEventChecks.remove(edgeId); } } - pendingDisconnectNotifications.values().forEach(task -> { - if (task != null && !task.isDone()) { - task.cancel(false); - } - }); + pendingDisconnectNotifications.values().forEach(EdgeGrpcService::cancelIfPending); pendingDisconnectNotifications.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); @@ -330,9 +327,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } - cancelPendingDisconnectNotification(edgeId); cancelScheduleEdgeEventsCheck(edgeId); } + cancelPendingDisconnectNotification(edgeId); } private void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) { @@ -576,7 +573,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId); } } - edgeIdServiceIdCache.evict(edgeId); + evictServiceIdCacheIfOwnedByThisNode(edgeId); + } + + // Only evict if the cache still points to this node. If the edge already reconnected to a different + // 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())) { + edgeIdServiceIdCache.evict(edgeId); + } } private void destroySession(EdgeGrpcSession session) { @@ -710,10 +717,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } EdgeId edgeId = edge.getId(); pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { - if (existing != null && !existing.isDone()) { - existing.cancel(false); - } - return executorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), + cancelIfPending(existing); + return edgeEventProcessingExecutorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); }); } @@ -731,9 +736,12 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void cancelPendingDisconnectNotification(EdgeId edgeId) { - ScheduledFuture pending = pendingDisconnectNotifications.remove(edgeId); - if (pending != null && !pending.isDone()) { - pending.cancel(false); + cancelIfPending(pendingDisconnectNotifications.remove(edgeId)); + } + + private static void cancelIfPending(ScheduledFuture future) { + if (future != null && !future.isDone()) { + future.cancel(false); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 0cb3e57ae8..07ea4dbf06 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1646,7 +1646,7 @@ edges: # Delay (ms) before sending an edge "disconnected" notification. Suppressed if the edge reconnects within # this window - debounces flapping edges from spamming the notification center. Only the notification is # delayed; rule-engine events and state attributes update immediately. Set to 0 to notify immediately. - disconnect_notification_delay_ms: "${TB_EDGE_DISCONNECT_NOTIFICATION_DELAY_MS:60000}" + disconnect_notification_delay_ms: "${EDGES_DISCONNECT_NOTIFICATION_DELAY_MS:60000}" stats: # Enable or disable reporting of edge communication stats (true or false) enabled: "${EDGES_STATS_ENABLED:true}" diff --git a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java index d75c7b8f59..71802468ab 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -125,6 +125,7 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { public static final String EDGE_HOST = "localhost"; public static final int EDGE_PORT = TestSocketUtils.findAvailableTcpPort(); + @DynamicPropertySource static void props(DynamicPropertyRegistry registry) { log.debug("edges.rpc.port = {}", EDGE_PORT); @@ -157,16 +158,23 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { //8 installation messages installation(); - edgeImitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); + edgeImitator = createEdgeImitator(); // 17 connect messages + 8 installation messages edgeImitator.expectMessageAmount(SYNC_MESSAGE_COUNT); - edgeImitator.ignoreType(OAuth2ClientUpdateMsg.class); - edgeImitator.ignoreType(OAuth2DomainUpdateMsg.class); edgeImitator.connect(); verifyEdgeConnectionAndInitialData(); } + // Creates an EdgeImitator wired with the standard ignored message types, but not yet connected. + // Callers add any expectations (e.g. expectMessageAmount) before invoking connect() themselves. + protected EdgeImitator createEdgeImitator() throws Exception { + EdgeImitator imitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); + imitator.ignoreType(OAuth2ClientUpdateMsg.class); + imitator.ignoreType(OAuth2DomainUpdateMsg.class); + return imitator; + } + @After public void teardownEdgeTest() { try { 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 f72ad51f4a..8179e5894a 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -25,8 +25,6 @@ import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.controller.AbstractWebTest; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.edge.imitator.EdgeImitator; -import org.thingsboard.server.gen.edge.v1.OAuth2ClientUpdateMsg; -import org.thingsboard.server.gen.edge.v1.OAuth2DomainUpdateMsg; import java.util.concurrent.TimeUnit; @@ -70,9 +68,7 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { TimeUnit.SECONDS.sleep(1); // ...and reconnects within the delay window, which must cancel the pending "disconnected" notification. - EdgeImitator reconnected = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); - reconnected.ignoreType(OAuth2ClientUpdateMsg.class); - reconnected.ignoreType(OAuth2DomainUpdateMsg.class); + EdgeImitator reconnected = createEdgeImitator(); reconnected.connect(); edgeImitator = reconnected; // let teardown clean up the live session diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java new file mode 100644 index 0000000000..a9d6dbe455 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java @@ -0,0 +1,64 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.edge; + +import org.junit.Test; +import org.mockito.ArgumentMatcher; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTrigger; +import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; +import org.thingsboard.server.controller.AbstractWebTest; +import org.thingsboard.server.dao.service.DaoSqlTest; + +import java.util.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +// Covers the disconnect_notification_delay_ms <= 0 branch in EdgeGrpcService.scheduleDisconnectNotification, +// where the "disconnected" notification must be sent immediately instead of being scheduled with a delay. +@DaoSqlTest +@TestPropertySource(properties = { + "edges.connectivity.disconnect_notification_delay_ms=0" +}) +public class EdgeImmediateDisconnectNotificationTest extends AbstractEdgeTest { + + @MockitoSpyBean + private NotificationRuleProcessor notificationRuleProcessor; + + @Test + public void givenZeroDelay_whenEdgeDisconnects_thenDisconnectNotificationSentImmediately() throws Exception { + clearInvocations(notificationRuleProcessor); + + edgeImitator.disconnect(); + + // With a zero delay there is no debounce window - the "disconnected" notification fires right away. + await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> + verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger()))); + } + + private ArgumentMatcher edgeConnectionTrigger() { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger + && edge.getId().equals(edgeTrigger.getEdgeId()) + && !edgeTrigger.isConnected(); + } + +}