refactor: address PR #15732 review - cancel helper, pooled executor, test coverage

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<String> 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);
         }
     }
 

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}"

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 {

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
 

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<NotificationRuleTrigger> edgeConnectionTrigger() {
+        return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger
+                && edge.getId().equals(edgeTrigger.getEdgeId())
+                && !edgeTrigger.isConnected();
+    }
+
+}