diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java index 5df3e638f5..7b5a2a0fad 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java @@ -25,6 +25,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.exception.TenantNotFoundException; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; @@ -86,7 +87,15 @@ public abstract class AbstractSubscriptionService extends TbApplicationEventList protected void forwardToSubscriptionManagerService(TenantId tenantId, EntityId entityId, Consumer toSubscriptionManagerService, Supplier toCore) { - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + TopicPartitionInfo tpi; + try { + tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + } catch (TenantNotFoundException e) { + // The tenant was deleted (e.g. concurrently with an in-flight asynchronous save callback), + // so there is no partition to route to and no subscribers to notify. Nothing to forward. + log.debug("[{}][{}] Skipping subscription update: tenant no longer exists.", tenantId, entityId); + return; + } if (currentPartitions.contains(tpi)) { if (subscriptionManagerService.isPresent()) { toSubscriptionManagerService.accept(subscriptionManagerService.get()); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index bde1873627..e3a69be624 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/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}" 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..8a3c3c1ae7 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/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 { diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeConnectionNotificationTest.java new file mode 100644 index 0000000000..4482177750 --- /dev/null +++ b/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 edgeConnectionTrigger(boolean connected) { + return trigger -> trigger instanceof EdgeConnectionTrigger edgeTrigger + && edge.getId().equals(edgeTrigger.getEdgeId()) + && edgeTrigger.isConnected() == connected; + } + +} diff --git a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java index f278153ed8..59b114a3eb 100644 --- a/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java @@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.ApiUsageStateValue; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.exception.TenantNotFoundException; import org.thingsboard.server.common.data.id.ApiUsageStateId; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; @@ -89,6 +90,7 @@ import java.util.stream.Stream; import static com.google.common.util.concurrent.Futures.immediateFailedFuture; import static com.google.common.util.concurrent.Futures.immediateFuture; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; @@ -1151,6 +1153,26 @@ class DefaultTelemetrySubscriptionServiceTest { then(deviceStateManager).shouldHaveNoInteractions(); } + /* --- Subscription forwarding --- */ + + @Test + void shouldSkipSubscriptionForwardWhenTenantWasDeleted() { + // GIVEN the tenant was deleted concurrently, so partition resolution fails + given(partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId)) + .willThrow(new TenantNotFoundException(tenantId)); + + // WHEN forwarding a subscription update (e.g. from an in-flight async save callback) + // THEN it must not propagate the exception + assertThatNoException().isThrownBy(() -> telemetryService.forwardToSubscriptionManagerService( + tenantId, entityId, + sm -> sm.onAttributesUpdate(tenantId, entityId, AttributeScope.SERVER_SCOPE.name(), List.of(), TbCallback.EMPTY), + () -> null)); + + // AND nothing is forwarded, since there is no partition to route to and no subscribers to notify + then(subscriptionManagerService).shouldHaveNoInteractions(); + then(clusterService).shouldHaveNoInteractions(); + } + // used to emulate versions returned by save APIs private static List listOfNNumbers(int N) { return LongStream.range(0, N).boxed().toList(); diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java index 7e58a2cb9f..2237e06f43 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java @@ -19,8 +19,17 @@ import io.grpc.HttpConnectProxiedSocketAddress; import io.grpc.ManagedChannel; import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; +import io.grpc.netty.shaded.io.netty.channel.Channel; +import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.epoll.Epoll; +import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.epoll.EpollSocketChannel; +import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioSocketChannel; import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder; +import io.grpc.netty.shaded.io.netty.util.concurrent.DefaultThreadFactory; import io.grpc.stub.StreamObserver; +import jakarta.annotation.PreDestroy; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; @@ -84,8 +93,14 @@ public class EdgeGrpcClient implements EdgeRpcClient { private ManagedChannel channel; + private final EventLoopGroup workerGroup = createWorkerGroup(); + private StreamObserver inputStream; + private volatile boolean connected; + + private volatile boolean streamActive; + private static final ReentrantLock uplinkMsgLock = new ReentrantLock(); @Override @@ -95,7 +110,11 @@ public class EdgeGrpcClient implements EdgeRpcClient { Consumer onEdgeUpdate, Consumer onDownlink, Consumer onError) { + connected = false; + streamActive = false; NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort) + .eventLoopGroup(workerGroup) + .channelType(channelType()) .maxInboundMessageSize(maxInboundMessageSize) .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS) .keepAliveTimeout(keepAliveTimeoutSec, TimeUnit.SECONDS) @@ -131,6 +150,7 @@ public class EdgeGrpcClient implements EdgeRpcClient { EdgeRpcServiceGrpc.EdgeRpcServiceStub stub = EdgeRpcServiceGrpc.newStub(channel); log.info("[{}] Sending a connect request to the TB!", edgeKey); this.inputStream = stub.withCompression("gzip").handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onDownlink, onError)); + streamActive = true; this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.CONNECT_RPC_MESSAGE) .setConnectRequestMsg(ConnectRequestMsg.newBuilder() @@ -146,6 +166,20 @@ public class EdgeGrpcClient implements EdgeRpcClient { return EdgeVersionComparator.getNewestEdgeVersion(); } + private static EventLoopGroup createWorkerGroup() { + DefaultThreadFactory threadFactory = new DefaultThreadFactory("edge-grpc-worker", true); + return Epoll.isAvailable() ? new EpollEventLoopGroup(1, threadFactory) : new NioEventLoopGroup(1, threadFactory); + } + + private static Class channelType() { + return Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class; + } + + @PreDestroy + public void destroy() { + workerGroup.shutdownGracefully(); + } + private StreamObserver initOutputStream(String edgeKey, Consumer onUplinkResponse, Consumer onEdgeUpdate, @@ -162,8 +196,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { serverMaxInboundMessageSize = connectResponseMsg.getMaxInboundMessageSize(); } log.info("[{}] Configuration received: {}", edgeKey, connectResponseMsg.getConfiguration()); + connected = true; onEdgeUpdate.accept(connectResponseMsg.getConfiguration()); } else { + connected = false; log.error("[{}] Failed to establish the connection! Code: {}. Error message: {}.", edgeKey, connectResponseMsg.getResponseCode(), connectResponseMsg.getErrorMsg()); try { EdgeGrpcClient.this.disconnect(true); @@ -186,6 +222,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void onError(Throwable t) { + connected = false; + streamActive = false; log.warn("[{}] Stream was terminated due to error:", edgeKey, t); try { EdgeGrpcClient.this.disconnect(true); @@ -197,6 +235,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void onCompleted() { + connected = false; + streamActive = false; log.info("[{}] Stream was closed and completed successfully!", edgeKey); } }; @@ -204,6 +244,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void disconnect(boolean onError) throws InterruptedException { + connected = false; + streamActive = false; if (!onError) { try { if (inputStream != null) { @@ -236,10 +278,19 @@ public class EdgeGrpcClient implements EdgeRpcClient { } } + @Override + public boolean isConnected() { + return connected; + } + @Override public void sendUplinkMsg(UplinkMsg msg) { uplinkMsgLock.lock(); try { + if (!streamActive) { + log.debug("Uplink msg is skipped, the cloud session is not established: {}", msg); + return; + } this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) .setUplinkMsg(msg) @@ -253,6 +304,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { public void sendSyncRequestMsg(boolean fullSyncRequired) { uplinkMsgLock.lock(); try { + if (!streamActive) { + log.debug("Sync request msg is skipped, the cloud session is not established"); + return; + } SyncRequestMsg syncRequestMsg = SyncRequestMsg.newBuilder() .setFullSync(fullSyncRequired) .build(); @@ -269,6 +324,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { public void sendDownlinkResponseMsg(DownlinkResponseMsg downlinkResponseMsg) { uplinkMsgLock.lock(); try { + if (!streamActive) { + log.debug("Downlink response msg is skipped, the cloud session is not established: {}", downlinkResponseMsg); + return; + } this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) .setDownlinkResponseMsg(downlinkResponseMsg) diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java index a1a7b5462f..55bdcac04e 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java @@ -34,6 +34,8 @@ public interface EdgeRpcClient { void disconnect(boolean onError) throws InterruptedException; + boolean isConnected(); + void sendSyncRequestMsg(boolean fullSyncRequired); void sendUplinkMsg(UplinkMsg uplinkMsg); diff --git a/common/edge-api/src/test/java/org/thingsboard/edge/rpc/EdgeGrpcClientLeakTest.java b/common/edge-api/src/test/java/org/thingsboard/edge/rpc/EdgeGrpcClientLeakTest.java new file mode 100644 index 0000000000..9fc66ed33e --- /dev/null +++ b/common/edge-api/src/test/java/org/thingsboard/edge/rpc/EdgeGrpcClientLeakTest.java @@ -0,0 +1,168 @@ +/** + * 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.edge.rpc; + +import io.grpc.Server; +import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import io.grpc.netty.shaded.io.netty.buffer.PooledByteBufAllocator; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.gen.edge.v1.ConnectResponseCode; +import org.thingsboard.server.gen.edge.v1.ConnectResponseMsg; +import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; +import org.thingsboard.server.gen.edge.v1.RequestMsg; +import org.thingsboard.server.gen.edge.v1.RequestMsgType; +import org.thingsboard.server.gen.edge.v1.ResponseMsg; +import org.thingsboard.server.gen.edge.v1.UplinkMsg; + +import java.lang.reflect.Method; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +import static org.junit.jupiter.api.Assertions.fail; + +class EdgeGrpcClientLeakTest { + + // The window in which the default shared event loop group would be destroyed after the channel + // terminates (SharedResourceHolder delays destruction by 1 second). If EdgeGrpcClient ever goes + // back to the shared group, writes after this window hit a terminated executor and every buffer + // committed to grpc-netty's WriteQueue is pinned forever (4112 bytes per message, silent after + // the first RejectedExecutionException). + private static final long SHARED_GROUP_DEATH_WINDOW_MS = 3000; + private static final int MSG_COUNT = 50; + private static final long AWAIT_TIMEOUT_MS = 15_000; + + private Server server; + private EdgeGrpcClient client; + + @Test + void uplinksSentAfterTransportDeathDoNotPinPooledBuffers() throws Exception { + server = NettyServerBuilder.forPort(0) + .addService(new EdgeRpcServiceGrpc.EdgeRpcServiceImplBase() { + @Override + public StreamObserver handleMsgs(StreamObserver outputStream) { + return new StreamObserver<>() { + @Override + public void onNext(RequestMsg requestMsg) { + if (requestMsg.hasConnectRequestMsg()) { + outputStream.onNext(ResponseMsg.newBuilder() + .setConnectResponseMsg(ConnectResponseMsg.newBuilder() + .setResponseCode(ConnectResponseCode.ACCEPTED) + .build()) + .build()); + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + } + }) + .build() + .start(); + + client = new EdgeGrpcClient(); + ReflectionTestUtils.setField(client, "rpcHost", "localhost"); + ReflectionTestUtils.setField(client, "rpcPort", server.getPort()); + ReflectionTestUtils.setField(client, "timeoutSecs", 1); + ReflectionTestUtils.setField(client, "keepAliveTimeSec", 10); + ReflectionTestUtils.setField(client, "keepAliveTimeoutSec", 5); + ReflectionTestUtils.setField(client, "maxInboundMessageSize", 4194304); + + client.connect("leakTest", "leakTest", msg -> {}, cfg -> {}, msg -> {}, e -> {}); + await("client to connect", () -> client.isConnected()); + + server.shutdownNow(); + server.awaitTermination(10, TimeUnit.SECONDS); + await("client to observe the transport death", () -> !client.isConnected()); + Thread.sleep(SHARED_GROUP_DEATH_WINDOW_MS); + + long baseline = pinnedBytes(); + @SuppressWarnings("unchecked") + StreamObserver inputStream = (StreamObserver) ReflectionTestUtils.getField(client, "inputStream"); + RequestMsg uplink = RequestMsg.newBuilder() + .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) + .setUplinkMsg(UplinkMsg.newBuilder().setUplinkMsgId(1).build()) + .build(); + // Bypasses the connected gate on purpose: this models the check-then-act straggler (and the + // pre-gate retry loop) writing to a stream whose transport is already gone. Exceptions are + // swallowed the same way the production retry loop survives them. + for (int i = 0; i < MSG_COUNT; i++) { + try { + inputStream.onNext(uplink); + } catch (RuntimeException ignored) { + } + } + + long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS; + while (pinnedBytes() > baseline) { + if (System.currentTimeMillis() > deadline) { + fail("Pinned pooled memory did not return to baseline: " + (pinnedBytes() - baseline) + + " bytes retained after " + MSG_COUNT + " uplinks to a dead stream"); + } + Thread.sleep(50); + } + } + + @AfterEach + void tearDown() throws Exception { + if (client != null) { + client.disconnect(true); + client.destroy(); + } + if (server != null) { + server.shutdownNow(); + } + } + + private void await(String what, BooleanSupplier condition) throws InterruptedException { + long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS; + while (!condition.getAsBoolean()) { + if (System.currentTimeMillis() > deadline) { + fail("Timed out waiting for " + what); + } + Thread.sleep(50); + } + } + + // grpc-netty builds its own PooledByteBufAllocator instances instead of using + // PooledByteBufAllocator.DEFAULT, and the factory that owns them is package private - so they + // have to be pulled out reflectively. Both variants are checked so the assertion holds no matter + // which one the transport picks on this platform/version. + private static long pinnedBytes() { + try { + Class utils = Class.forName("io.grpc.netty.shaded.io.grpc.netty.Utils"); + Method getByteBufAllocator = utils.getDeclaredMethod("getByteBufAllocator", boolean.class); + getByteBufAllocator.setAccessible(true); + long total = 0; + for (boolean forceHeapBuffer : new boolean[]{false, true}) { + PooledByteBufAllocator pooled = (PooledByteBufAllocator) getByteBufAllocator.invoke(null, forceHeapBuffer); + total += pooled.pinnedDirectMemory() + pooled.pinnedHeapMemory(); + } + return total; + } catch (Exception e) { + throw new IllegalStateException("Failed to read the gRPC allocator metrics", e); + } + } + +} 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) {