From f10ca1338c99c22148bad5a026435e916bb69be3 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Wed, 10 Jun 2026 16:34:02 +0300 Subject: [PATCH] removeOnCancelPolicy, cluster-path unit test, cleanups --- .../service/edge/rpc/EdgeGrpcService.java | 90 +++++--- .../server/edge/AbstractEdgeTest.java | 1 - .../edge/EdgeConnectionNotificationTest.java | 56 ++++- ...geImmediateDisconnectNotificationTest.java | 64 ------ .../service/edge/rpc/EdgeGrpcServiceTest.java | 204 ++++++++++++++++++ .../common/util/ThingsBoardExecutors.java | 10 +- 6 files changed, 317 insertions(+), 108 deletions(-) delete mode 100644 application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java create mode 100644 application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.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 a1910396dc..b875226034 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 @@ -69,6 +69,7 @@ import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.EdgeContextComponent; +import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import java.io.IOException; import java.util.ArrayList; @@ -103,7 +104,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private final ConcurrentMap sessionNewEventsLocks = new ConcurrentHashMap<>(); private final Map sessionNewEvents = new HashMap<>(); private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>(); - private final ConcurrentMap> pendingDisconnectNotifications = new ConcurrentHashMap<>(); + private final ConcurrentMap pendingDisconnectNotifications = new ConcurrentHashMap<>(); private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>(); private final ConcurrentMap edgeEventsMigrationProcessed = new ConcurrentHashMap<>(); private final List zombieSessions = new ArrayList<>(); @@ -144,6 +145,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i @Lazy private EdgeContextComponent ctx; + @Autowired + private TelemetrySubscriptionService tsSubService; + @Autowired private TbClusterService clusterService; @@ -198,7 +202,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } this.edgeEventProcessingExecutorService = ThingsBoardExecutors.newScheduledThreadPool(schedulerPoolSize, "edge-event-check-scheduler"); this.sendDownlinkExecutorService = ThingsBoardExecutors.newScheduledThreadPool(sendSchedulerPoolSize, "edge-send-scheduler"); - this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service"); + // removeOnCancelPolicy: cancelled delayed disconnect notifications (common with flapping edges) are dropped + // from the queue right away on reconnect, instead of piling up until their original fire time. + this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service", true); this.executorService.scheduleAtFixedRate(this::cleanupZombieSessions, 60, 60, TimeUnit.SECONDS); log.info("Edge RPC service initialized!"); } @@ -232,19 +238,23 @@ 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); + }); if (server != null) { server.shutdownNow(); } - for (Map.Entry> entry : sessionEdgeEventChecks.entrySet()) { - EdgeId edgeId = entry.getKey(); - ScheduledFuture sessionEdgeEventCheck = entry.getValue(); - if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) { - sessionEdgeEventCheck.cancel(true); - sessionEdgeEventChecks.remove(edgeId); - } - } - pendingDisconnectNotifications.values().forEach(EdgeGrpcService::cancelIfPending); - pendingDisconnectNotifications.clear(); + sessionEdgeEventChecks.values().forEach(task -> cancelIfPending(task, true)); + sessionEdgeEventChecks.clear(); if (edgeEventProcessingExecutorService != null) { edgeEventProcessingExecutorService.shutdownNow(); } @@ -393,6 +403,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i // If the edge reconnected within the disconnect-notification delay window, suppress the pending // "disconnected" notification - the drop was transient (debounce for flapping edges). cancelPendingDisconnectNotification(edgeId); + // Connect notifies immediately; only the disconnect notification is debounced (see scheduleDisconnectNotification). pushStateEventToRuleEngine(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); notifyEdgeConnectivity(tenantId, edge, true); cancelScheduleEdgeEventsCheck(edgeId); @@ -528,13 +539,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void cancelScheduleEdgeEventsCheck(EdgeId edgeId) { log.trace("[{}] cancelling edge event check for edge", edgeId); - if (sessionEdgeEventChecks.containsKey(edgeId)) { - ScheduledFuture sessionEdgeEventCheck = sessionEdgeEventChecks.get(edgeId); - if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) { - sessionEdgeEventCheck.cancel(true); - sessionEdgeEventChecks.remove(edgeId); - } - } + cancelIfPending(sessionEdgeEventChecks.remove(edgeId), true); } private void onEdgeDisconnect(Edge edge, UUID sessionId) { @@ -602,14 +607,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void save(TenantId tenantId, EdgeId edgeId, String key, long value) { log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { - ctx.getTsSubService().saveTimeseries(TimeseriesSaveRequest.builder() + tsSubService.saveTimeseries(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .entry(new LongDataEntry(key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .build()); } else { - ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder() + tsSubService.saveAttributes(AttributesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .scope(AttributeScope.SERVER_SCOPE) @@ -622,14 +627,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private void save(TenantId tenantId, EdgeId edgeId, String key, boolean value) { log.debug("[{}][{}] Updating boolean edge telemetry [{}] [{}]", tenantId, edgeId, key, value); if (persistToTelemetry) { - ctx.getTsSubService().saveTimeseries(TimeseriesSaveRequest.builder() + tsSubService.saveTimeseries(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .entry(new BooleanDataEntry(key, value)) .callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) .build()); } else { - ctx.getTsSubService().saveAttributes(AttributesSaveRequest.builder() + tsSubService.saveAttributes(AttributesSaveRequest.builder() .tenantId(tenantId) .entityId(edgeId) .scope(AttributeScope.SERVER_SCOPE) @@ -717,31 +722,51 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } EdgeId edgeId = edge.getId(); pendingDisconnectNotifications.compute(edgeId, (id, existing) -> { - cancelIfPending(existing); - return edgeEventProcessingExecutorService.schedule(() -> fireDelayedDisconnectNotification(tenantId, edge), + if (existing != null) { + cancelIfPending(existing.future()); + } + // Single-element holder so the scheduled task can reference its own pending entry, which doesn't exist + // until schedule(...) returns. The task needs it for the identity-keyed remove in fireDelayedDisconnectNotification. + PendingDisconnect[] holder = {null}; + ScheduledFuture future = executorService.schedule( + () -> fireDelayedDisconnectNotification(holder[0]), disconnectNotificationDelayMs, TimeUnit.MILLISECONDS); + holder[0] = new PendingDisconnect(tenantId, edge, future); + return holder[0]; }); } - private void fireDelayedDisconnectNotification(TenantId tenantId, Edge edge) { - EdgeId edgeId = edge.getId(); - pendingDisconnectNotifications.remove(edgeId); + private void fireDelayedDisconnectNotification(PendingDisconnect pending) { + EdgeId edgeId = pending.edge().getId(); + // Identity-keyed remove: don't clobber a newer entry if a second disconnect races with this task firing. + 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) { - log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", tenantId, edgeId); + log.debug("[{}][{}] Edge reconnected within the disconnect notification delay - skipping disconnect notification", pending.tenantId(), edgeId); return; } - notifyEdgeConnectivity(tenantId, edge, false); + notifyEdgeConnectivity(pending.tenantId(), pending.edge(), false); } private void cancelPendingDisconnectNotification(EdgeId edgeId) { - cancelIfPending(pendingDisconnectNotifications.remove(edgeId)); + PendingDisconnect pending = pendingDisconnectNotifications.remove(edgeId); + if (pending != null) { + cancelIfPending(pending.future()); + } } + // Carries the context the delayed task needs, so a pending notification can still be fired on shutdown + // (see destroy()), not just cancelled. Package-private for unit testing of fireDelayedDisconnectNotification. + record PendingDisconnect(TenantId tenantId, Edge edge, ScheduledFuture future) {} + private static void cancelIfPending(ScheduledFuture future) { + cancelIfPending(future, false); + } + + private static void cancelIfPending(ScheduledFuture future, boolean mayInterruptIfRunning) { if (future != null && !future.isDone()) { - future.cancel(false); + future.cancel(mayInterruptIfRunning); } } @@ -797,6 +822,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i !kafkaSession.getConsumer().getConsumer().isStopped(); } return false; + } } 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 71802468ab..8a3c3c1ae7 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -125,7 +125,6 @@ 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); 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 8179e5894a..c46d82437a 100644 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java @@ -15,16 +15,20 @@ */ package org.thingsboard.server.edge; +import org.junit.After; +import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentMatcher; -import org.springframework.test.context.TestPropertySource; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.springframework.test.util.ReflectionTestUtils; 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 org.thingsboard.server.edge.imitator.EdgeImitator; +import org.thingsboard.server.service.edge.rpc.EdgeGrpcService; import java.util.concurrent.TimeUnit; @@ -37,9 +41,6 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @DaoSqlTest -@TestPropertySource(properties = { - "edges.connectivity.disconnect_notification_delay_ms=5000" -}) public class EdgeConnectionNotificationTest extends AbstractEdgeTest { private static final long DELAY_MS = 5000L; @@ -47,19 +48,55 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { @MockitoSpyBean private NotificationRuleProcessor notificationRuleProcessor; + @Autowired + private EdgeGrpcService edgeGrpcService; + + private long originalDisconnectNotificationDelayMs; + + // Capture the bean's configured delay before each test and restore it after, so a test method that forgets + // to call setDisconnectNotificationDelayMs can't silently inherit the previous method's mutated value + // (the shared context means the mutation would otherwise persist across methods). + @Before + public void captureDisconnectNotificationDelay() { + originalDisconnectNotificationDelayMs = (long) ReflectionTestUtils.getField(edgeGrpcService, "disconnectNotificationDelayMs"); + } + + @After + public void restoreDisconnectNotificationDelay() { + setDisconnectNotificationDelayMs(originalDisconnectNotificationDelayMs); + } + + // The delay is overridden per test (rather than via a per-class @TestPropertySource) so all cases share a + // single Spring application context instead of booting a separate heavy context per delay value. + private void setDisconnectNotificationDelayMs(long delayMs) { + ReflectionTestUtils.setField(edgeGrpcService, "disconnectNotificationDelayMs", delayMs); + } + @Test public void givenEdgeStaysDisconnected_whenDelayElapses_thenDisconnectNotificationSent() throws Exception { + // After the configured delay, the "disconnected" notification is sent exactly once. + assertDisconnectNotificationSentOnce(DELAY_MS); + } + + @Test + public void givenZeroDelay_whenEdgeDisconnects_thenDisconnectNotificationSentImmediately() throws Exception { + // With a zero delay there is no debounce window - the "disconnected" notification fires right away. + assertDisconnectNotificationSentOnce(0); + } + + private void assertDisconnectNotificationSentOnce(long delayMs) throws Exception { + setDisconnectNotificationDelayMs(delayMs); clearInvocations(notificationRuleProcessor); edgeImitator.disconnect(); - // After the configured delay, the "disconnected" notification is sent exactly once. await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> verify(notificationRuleProcessor, times(1)).process(argThat(edgeConnectionTrigger(false)))); } @Test public void givenEdgeReconnectsWithinDelay_whenEdgeFlaps_thenDisconnectNotificationSuppressed() throws Exception { + setDisconnectNotificationDelayMs(DELAY_MS); clearInvocations(notificationRuleProcessor); // Edge drops... @@ -76,11 +113,10 @@ public class EdgeConnectionNotificationTest extends AbstractEdgeTest { await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> verify(notificationRuleProcessor, atLeastOnce()).process(argThat(edgeConnectionTrigger(true)))); - // Wait until the original disconnect-notification window has fully elapsed... - TimeUnit.MILLISECONDS.sleep(DELAY_MS + 1000); - - // ...the "disconnected" notification must have never been sent. - verify(notificationRuleProcessor, never()).process(argThat(edgeConnectionTrigger(false))); + // 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)))); } private ArgumentMatcher edgeConnectionTrigger(boolean connected) { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java deleted file mode 100644 index a9d6dbe455..0000000000 --- a/application/src/test/java/org/thingsboard/server/edge/EdgeImmediateDisconnectNotificationTest.java +++ /dev/null @@ -1,64 +0,0 @@ -/** - * 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(); - } - -} 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 new file mode 100644 index 0000000000..f538bd1baa --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcServiceTest.java @@ -0,0 +1,204 @@ +/** + * 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.service.edge.rpc; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentMatcher; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.cache.SimpleTbCacheValueWrapper; +import org.thingsboard.server.cache.TbTransactionalCache; +import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.TenantId; +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.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.service.edge.EdgeContextComponent; + +import java.util.UUID; +import java.util.concurrent.ConcurrentMap; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +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 { + + private static final String THIS_NODE = "tb-core-1"; + private static final String OTHER_NODE = "tb-core-2"; + + @Mock + private TbTransactionalCache edgeIdServiceIdCache; + + @Mock + private TbServiceInfoProvider serviceInfoProvider; + + @Mock + private EdgeContextComponent ctx; + + @Mock + private NotificationRuleProcessor ruleProcessor; + + @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.setName("test-edge"); + } + + // --- evictServiceIdCacheIfOwnedByThisNode --- + + @Test + public void givenCacheOwnedByThisNode_whenEvict_thenEntryIsEvicted() { + when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(THIS_NODE)); + + evictServiceIdCacheIfOwnedByThisNode(); + + verify(edgeIdServiceIdCache, times(1)).evict(edgeId); + } + + @Test + public void givenCacheOwnedByAnotherNode_whenEvict_thenEntryIsKept() { + // The edge already reconnected to another node within the keep-alive window: must NOT wipe the live owner. + when(serviceInfoProvider.getServiceId()).thenReturn(THIS_NODE); + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); + + evictServiceIdCacheIfOwnedByThisNode(); + + verify(edgeIdServiceIdCache, never()).evict(edgeId); + } + + @Test + public void givenEmptyCache_whenEvict_thenNothingEvicted() { + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); + + evictServiceIdCacheIfOwnedByThisNode(); + + verify(edgeIdServiceIdCache, never()).evict(edgeId); + } + + // --- fireDelayedDisconnectNotification re-verify guard --- + + @Test + public void givenEdgeReconnectedToThisNode_whenDelayFires_thenNotificationSuppressed() { + // A live session exists again on this node - suppress the stale disconnect notification. + sessions().put(edgeId, mock(EdgeGrpcSession.class)); + + fireDelayedDisconnectNotification(); + + verify(ruleProcessor, never()).process(any()); + } + + @Test + public void givenEdgeReconnectedToAnotherNode_whenDelayFires_thenNotificationSuppressed() { + // No local session, but the cluster cache still points at some node: the edge is connected elsewhere. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); + + fireDelayedDisconnectNotification(); + + verify(ruleProcessor, never()).process(any()); + } + + @Test + public void givenEdgeStaysDisconnectedClusterWide_whenDelayFires_thenNotificationSent() { + // No local session and no cache entry on any node: the edge is genuinely down - fire the notification. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); + when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); + + fireDelayedDisconnectNotification(); + + verify(ruleProcessor, times(1)).process(argThat(disconnectTrigger())); + } + + @Test + public void givenPendingDisconnect_whenDestroy_thenNotificationFlushed() { + // The edge is genuinely down (no session, no cache). A graceful shutdown must flush the pending + // notification rather than drop it, otherwise a restart within the delay window swallows the alert. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(null); + when(ctx.getRuleProcessor()).thenReturn(ruleProcessor); + pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(tenantId, edge, null)); + + destroy(); + + verify(ruleProcessor, times(1)).process(argThat(disconnectTrigger())); + } + + @Test + public void givenPendingDisconnectButReconnectedElsewhere_whenDestroy_thenNotificationSuppressed() { + // The flush still honors the re-verify guard: an edge that reconnected to another node must not alert. + when(edgeIdServiceIdCache.get(edgeId)).thenReturn(SimpleTbCacheValueWrapper.wrap(OTHER_NODE)); + pendingDisconnects().put(edgeId, new EdgeGrpcService.PendingDisconnect(tenantId, edge, null)); + + destroy(); + + verify(ruleProcessor, never()).process(any()); + } + + private void destroy() { + ReflectionTestUtils.invokeMethod(edgeGrpcService, "destroy"); + } + + private void evictServiceIdCacheIfOwnedByThisNode() { + ReflectionTestUtils.invokeMethod(edgeGrpcService, "evictServiceIdCacheIfOwnedByThisNode", edgeId); + } + + private void fireDelayedDisconnectNotification() { + EdgeGrpcService.PendingDisconnect pending = new EdgeGrpcService.PendingDisconnect(tenantId, edge, null); + ReflectionTestUtils.invokeMethod(edgeGrpcService, "fireDelayedDisconnectNotification", pending); + } + + @SuppressWarnings("unchecked") + private ConcurrentMap sessions() { + return (ConcurrentMap) ReflectionTestUtils.getField(edgeGrpcService, "sessions"); + } + + @SuppressWarnings("unchecked") + private ConcurrentMap pendingDisconnects() { + return (ConcurrentMap) ReflectionTestUtils.getField(edgeGrpcService, "pendingDisconnectNotifications"); + } + + private static ArgumentMatcher disconnectTrigger() { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger && !edgeTrigger.isConnected(); + } + +} diff --git a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java index 71b4925901..47934b4df0 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java +++ b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardExecutors.java @@ -69,7 +69,15 @@ public class ThingsBoardExecutors { } public static ScheduledExecutorService newSingleThreadScheduledExecutor(String name) { - return Executors.unconfigurableScheduledExecutorService(new ThingsBoardScheduledThreadPoolExecutor(1, ThingsBoardThreadFactory.forName(name))); + return newSingleThreadScheduledExecutor(name, false); + } + + public static ScheduledExecutorService newSingleThreadScheduledExecutor(String name, boolean removeOnCancelPolicy) { + ThingsBoardScheduledThreadPoolExecutor executor = new ThingsBoardScheduledThreadPoolExecutor(1, ThingsBoardThreadFactory.forName(name)); + // Must be set before wrapping: unconfigurableScheduledExecutorService hides the setter. With it enabled, + // cancelled tasks are removed from the delay queue immediately instead of lingering until their fire time. + executor.setRemoveOnCancelPolicy(removeOnCancelPolicy); + return Executors.unconfigurableScheduledExecutorService(executor); } public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, String name) {