Merge pull request #16083 from AndriiLandiak/merge-with-lts4.3

Merge with lts4.3

application/src/main/resources/thingsboard.yml

@@ -1676,6 +1676,11 @@ edges:
   state:
     # Persist state of edge (active, last connect, last disconnect) into timeseries or attributes tables. 'false' means to store edge state into attributes table
     persistToTelemetry: "${EDGES_PERSIST_STATE_TO_TELEMETRY:false}"
+  connectivity:
+    # 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: "${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

@@ -157,16 +157,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

@@ -0,0 +1,133 @@
+/**
+ * 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.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentMatcher;
+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.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@DaoSqlTest
+public class EdgeConnectionNotificationTest extends AbstractEdgeTest {
+
+    private static final long DELAY_MS = 1500L;
+
+    @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();
+
+        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...
+        edgeImitator.disconnect();
+        // Wait until the server has processed the disconnect and scheduled the pending notification
+        // (the edge id appears in the pendingDisconnectNotifications map) before reconnecting.
+        await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).until(() -> {
+            Map<?, ?> pending = (Map<?, ?>) ReflectionTestUtils.getField(edgeGrpcService, "pendingDisconnectNotifications");
+            return pending != null && pending.containsKey(edge.getId());
+        });
+
+        // ...and reconnects within the delay window, which must cancel the pending "disconnected" notification.
+        EdgeImitator reconnected = createEdgeImitator();
+        reconnected.connect();
+        edgeImitator = reconnected; // let teardown clean up the live session
+
+        // The "connected" notification still fires immediately on reconnect (we suppress the disconnect only).
+        await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS).untilAsserted(() ->
+                verify(notificationRuleProcessor, atLeastOnce()).process(argThat(edgeConnectionTrigger(true))));
+
+        // 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<NotificationRuleTrigger> edgeConnectionTrigger(boolean connected) {
+        return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger
+                && edge.getId().equals(edgeTrigger.getEdgeId())
+                && edgeTrigger.isConnected() == connected;
+    }
+
+}

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) {