From 517fd3a7d3fb3aa7297631db14eb209e80058a01 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Tue, 9 Dec 2025 18:14:31 +0200 Subject: [PATCH 01/10] closing device sessions after tenant deletion --- .../client/MqttClientConnectionTest.java | 40 +++++++++++++++++++ .../service/DefaultTransportService.java | 20 ++++++++++ 2 files changed, 60 insertions(+) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java index 2c91bc85f1..36ee145c65 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java @@ -15,10 +15,20 @@ */ package org.thingsboard.server.transport.mqtt.mqttv3.client; +import org.awaitility.Awaitility; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.security.Authority; +import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; +import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; + +import java.util.concurrent.TimeUnit; @DaoSqlTest public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { @@ -45,4 +55,34 @@ public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { public void testClientWithWrongClientIdAndEmptyUsernamePassword() throws Exception { processClientWithWrongClientIdAndEmptyUsernamePasswordTest(); } + + @Test + public void testClientShouldBeDisconnectedAfterTenantDeletion() throws Exception { + loginSysAdmin(); + Tenant tenant = new Tenant(); + tenant.setTitle("Mqtt Client Connection Test Tenant"); + Tenant savedTenant = saveTenant(tenant); + + User tenantAdmin = new User(); + tenantAdmin.setAuthority(Authority.TENANT_ADMIN); + tenantAdmin.setTenantId(savedTenant.getTenantId()); + tenantAdmin.setEmail("mqttTestClient@thingsboard.org"); + createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD); + + savedDevice = createDevice(RandomStringUtils.randomAlphabetic(10), "default", false); + DeviceCredentials deviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class); + accessToken = deviceCredentials.getCredentialsId(); + + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(accessToken); + Assert.assertTrue(client.isConnected()); + + loginSysAdmin(); + deleteTenant(savedTenant.getId()); + + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> !client.isConnected()); + } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 0f8d7df294..8ed799a620 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -45,6 +45,7 @@ import org.thingsboard.server.common.data.ResourceType; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.device.data.PowerMode; +import org.thingsboard.server.common.data.exception.TenantNotFoundException; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; @@ -931,6 +932,7 @@ public class DefaultTransportService extends TransportActivityManager implements TenantId tenantId = TenantId.fromUUID(entityUuid); rateLimitService.remove(tenantId); partitionService.removeTenant(tenantId); + onTenantDeleted(tenantId); } else if (EntityType.DEVICE.equals(entityType)) { rateLimitService.remove(new DeviceId(entityUuid)); onDeviceDeleted(new DeviceId(entityUuid)); @@ -966,6 +968,18 @@ public class DefaultTransportService extends TransportActivityManager implements } } + private void onTenantDeleted(TenantId tenantId) { + sessions.forEach((id, md) -> { + TenantId sessionTenantId = getTenantId(md.getSessionInfo()); + if (sessionTenantId.equals(tenantId)) { + DeviceId sessionDeviceId = getDeviceId(md.getSessionInfo()); + transportCallbackExecutor.submit(() -> { + md.getListener().onDeviceDeleted(sessionDeviceId); + }); + eventPublisher.publishEvent(new DeviceDeletedEvent(sessionDeviceId)); + } + }); + } public void onProfileUpdate(DeviceProfile deviceProfile) { long deviceProfileIdMSB = deviceProfile.getId().getId().getMostSignificantBits(); @@ -1107,6 +1121,12 @@ public class DefaultTransportService extends TransportActivityManager implements TopicPartitionInfo tpi; try { tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + } catch (TenantNotFoundException e) { + log.warn("Ignoring message delivery because Tenant with ID [{}] not found, routingKey [{}], msg [{}].", tenantId, routingKey, msg); + if (callback != null) { + callback.onError(e); + } + return; } catch (Exception e) { log.warn("Failed to send message to core. Tenant with ID [{}], routingKey [{}], msg [{}]. Message delivery aborted.", tenantId, routingKey, msg, e); if (callback != null) { From 123db956bd967448ebb55e52df736a7e870115b5 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Fri, 12 Dec 2025 16:53:18 +0200 Subject: [PATCH 02/10] terminating sessions before removing tenant EntityTransportRateLimits to avoid creating on new limit --- .../common/transport/service/DefaultTransportService.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 8ed799a620..c9e84e8735 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -930,9 +930,9 @@ public class DefaultTransportService extends TransportActivityManager implements tenantProfileCache.remove(new TenantProfileId(entityUuid)); } else if (EntityType.TENANT.equals(entityType)) { TenantId tenantId = TenantId.fromUUID(entityUuid); + onDevicesDeleted(tenantId); rateLimitService.remove(tenantId); partitionService.removeTenant(tenantId); - onTenantDeleted(tenantId); } else if (EntityType.DEVICE.equals(entityType)) { rateLimitService.remove(new DeviceId(entityUuid)); onDeviceDeleted(new DeviceId(entityUuid)); @@ -968,7 +968,7 @@ public class DefaultTransportService extends TransportActivityManager implements } } - private void onTenantDeleted(TenantId tenantId) { + private void onDevicesDeleted(TenantId tenantId) { sessions.forEach((id, md) -> { TenantId sessionTenantId = getTenantId(md.getSessionInfo()); if (sessionTenantId.equals(tenantId)) { From 94e875e264859807fd8e831e3a7bd9ee95d672bd Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Mon, 5 Jan 2026 15:53:56 +0200 Subject: [PATCH 03/10] cleanup redundant messages to core services when closing session white tenant deletion --- .../callback/AbstractSyncSessionCallback.java | 5 + .../coap/client/DefaultCoapClientContext.java | 40 +++++-- .../coap/CoapSessionListenerTest.java | 103 ++++++++++++++++++ .../transport/http/DeviceApiController.java | 4 + .../lwm2m/server/LwM2mSessionMsgListener.java | 5 + .../transport/mqtt/MqttTransportHandler.java | 5 + .../AbstractGatewayDeviceSessionContext.java | 5 + .../AbstractGatewaySessionHandler.java | 18 ++- .../transport/snmp/SnmpTransportContext.java | 12 +- .../snmp/session/DeviceSessionContext.java | 5 + .../common/transport/SessionMsgListener.java | 2 + .../service/DefaultTransportService.java | 6 +- 12 files changed, 194 insertions(+), 16 deletions(-) create mode 100644 common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java index a1bfb004b8..125491e18d 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java @@ -57,6 +57,11 @@ public abstract class AbstractSyncSessionCallback implements SessionMsgListener } + @Override + public void onTenantDeleted(DeviceId deviceId) { + + } + @Override public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg toDeviceRequest) { logUnsupportedCommandMessage(toDeviceRequest); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java index 5c17cb75e8..dab977f29c 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java @@ -560,6 +560,12 @@ public class DefaultCoapClientContext implements CoapClientContext { cancelAttributeSubscription(state); } + @Override + public void onTenantDeleted(DeviceId deviceId) { + cancelRpcSubscription(state, false); + cancelAttributeSubscription(state, false); + } + @Override public void onRemoteSessionCloseCommand(UUID sessionId, TransportProtos.SessionCloseNotificationProto sessionCloseNotification) { log.trace("[{}] Received the remote command to close the session: {}", sessionId, sessionCloseNotification.getMessage()); @@ -812,35 +818,49 @@ public class DefaultCoapClientContext implements CoapClientContext { } private void cancelRpcSubscription(TbCoapClientState state) { + cancelRpcSubscription(state, true); + } + + private void cancelRpcSubscription(TbCoapClientState state, boolean notifyCore) { if (state.getRpc() != null) { clientsByToken.remove(state.getRpc().getToken()); CoapExchange exchange = state.getRpc().getExchange(); state.setRpc(null); - transportService.process(state.getSession(), - TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), - new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + if (notifyCore) { + transportService.process(state.getSession(), + TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), + new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + } if (state.getAttrs() == null) { - closeAndCleanup(state); + closeAndCleanup(state, notifyCore); } } } private void cancelAttributeSubscription(TbCoapClientState state) { + cancelAttributeSubscription(state, true); + } + + private void cancelAttributeSubscription(TbCoapClientState state, boolean notifyCore) { if (state.getAttrs() != null) { clientsByToken.remove(state.getAttrs().getToken()); CoapExchange exchange = state.getAttrs().getExchange(); state.setAttrs(null); - transportService.process(state.getSession(), - TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), - new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + if (notifyCore) { + transportService.process(state.getSession(), + TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), + new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + } if (state.getRpc() == null) { - closeAndCleanup(state); + closeAndCleanup(state, notifyCore); } } } - private void closeAndCleanup(TbCoapClientState state) { - transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); + private void closeAndCleanup(TbCoapClientState state, boolean notifyCore) { + if (notifyCore) { + transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); + } transportService.deregisterSession(state.getSession()); state.setSession(null); state.setConfiguration(null); diff --git a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java new file mode 100644 index 0000000000..ff6a5ca348 --- /dev/null +++ b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java @@ -0,0 +1,103 @@ +package org.thingsboard.server.transport.coap; + +import org.eclipse.californium.core.server.resources.CoapExchange; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.thingsboard.server.coapserver.CoapServerContext; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.transport.TransportDeviceProfileCache; +import org.thingsboard.server.common.transport.TransportService; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.transport.coap.client.DefaultCoapClientContext; +import org.thingsboard.server.transport.coap.client.TbCoapClientState; +import org.thingsboard.server.transport.coap.client.TbCoapObservationState; + +import static org.mockito.ArgumentMatchers.any; +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; + +@ExtendWith(MockitoExtension.class) +public class CoapSessionListenerTest { + + @Mock + CoapServerContext config; + @Mock CoapTransportContext transportContext; + @Mock + TransportService transportService; + @Mock + TransportDeviceProfileCache profileCache; + @Mock + PartitionService partitionService; + + private DefaultCoapClientContext coapClientContext; + + @BeforeEach + void setUp() { + coapClientContext = new DefaultCoapClientContext(config, transportContext, transportService, profileCache, partitionService); + } + + @Test + void onTenantDeleted_shouldNotInteractWithCore() { + // Arrange + TbCoapClientState state = mock(TbCoapClientState.class); + + TbCoapObservationState rpcObs = mock(TbCoapObservationState.class); + when(rpcObs.getToken()).thenReturn("rpc-token"); + when(rpcObs.getExchange()).thenReturn(mock(CoapExchange.class)); + + TbCoapObservationState attrsObs = mock(TbCoapObservationState.class); + when(attrsObs.getToken()).thenReturn("attr-token"); + when(attrsObs.getExchange()).thenReturn(mock(CoapExchange.class)); + when(state.getRpc()).thenReturn(rpcObs); + when(state.getAttrs()).thenReturn(attrsObs); + + DefaultCoapClientContext.CoapSessionListener listener = coapClientContext.new CoapSessionListener(state); + + // Act + listener.onTenantDeleted(mock(DeviceId.class)); + + // Assert + verify(transportService, never()) + .process( + any(TransportProtos.SessionInfoProto.class), + any(TransportProtos.SessionEventMsg.class), + any() + ); + } + + @Test + void onTenantDeleted_shouldDeregisterSession() { + // GIVEN + TbCoapClientState state = mock(TbCoapClientState.class); + + TbCoapObservationState rpcObs = mock(TbCoapObservationState.class); + when(rpcObs.getToken()).thenReturn("rpc-token"); + when(rpcObs.getExchange()).thenReturn(mock(CoapExchange.class)); + + when(state.getRpc()).thenReturn(rpcObs); + when(state.getAttrs()).thenReturn(null); + + DefaultCoapClientContext.CoapSessionListener listener = coapClientContext.new CoapSessionListener(state); + + // WHEN + listener.onTenantDeleted(mock(DeviceId.class)); + + // THEN + verify(transportService, never()) + .process( + any(TransportProtos.SessionInfoProto.class), + any(TransportProtos.SessionEventMsg.class), + any() + ); + verify(transportService, times(1)) + .deregisterSession(any()); + } + +} diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index 069509787c..a91c89adbb 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java @@ -618,6 +618,10 @@ public class DeviceApiController implements TbTransportService { responseWriter.setResult(new ResponseEntity<>("Device was deleted!", HttpStatus.FORBIDDEN)); } + @Override + public void onTenantDeleted(DeviceId deviceId) { + onDeviceDeleted(deviceId); + } } private static MediaType parseMediaType(String contentType) { diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java index dc26fc1108..b49b02d3a5 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java @@ -114,4 +114,9 @@ public class LwM2mSessionMsgListener implements GenericFutureListener 0; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index a2ad982765..865788e9ac 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -227,6 +227,10 @@ public abstract class AbstractGatewaySessionHandler { TenantId sessionTenantId = getTenantId(md.getSessionInfo()); if (sessionTenantId.equals(tenantId)) { DeviceId sessionDeviceId = getDeviceId(md.getSessionInfo()); transportCallbackExecutor.submit(() -> { - md.getListener().onDeviceDeleted(sessionDeviceId); + md.getListener().onTenantDeleted(sessionDeviceId); }); eventPublisher.publishEvent(new DeviceDeletedEvent(sessionDeviceId)); } From 36c94de83674e44049863a4544edd225fee048bc Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Tue, 6 Jan 2026 11:01:15 +0200 Subject: [PATCH 04/10] license fix --- .../transport/coap/CoapSessionListenerTest.java | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java index ff6a5ca348..1dd3b2790e 100644 --- a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java +++ b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2025 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.transport.coap; import org.eclipse.californium.core.server.resources.CoapExchange; From da714692762b8b18eb4a9f02b89f817dd2bc60e9 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Wed, 7 Jan 2026 17:36:22 +0200 Subject: [PATCH 05/10] excluded messages to core while tenant deletion, added integration tests --- .../client/CoapClientIntegrationTest.java | 60 +++++++++++++++++++ .../client/MqttClientConnectionTest.java | 15 +++++ .../coap/client/DefaultCoapClientContext.java | 25 +++----- .../coap/client/TbCoapClientState.java | 8 +++ .../coap/CoapSessionListenerTest.java | 46 ++------------ .../transport/mqtt/MqttTransportHandler.java | 6 +- .../mqtt/session/DeviceSessionCtx.java | 9 +++ .../mqtt/MqttTransportHandlerTest.java | 14 +++++ .../transport/snmp/SnmpTransportContext.java | 10 +--- .../snmp/session/DeviceSessionContext.java | 12 +++- .../transport/session/SessionCloseReason.java | 22 +++++++ 11 files changed, 159 insertions(+), 68 deletions(-) create mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java index 3845cc0c3c..5404e0ef31 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java @@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.awaitility.Awaitility; import org.eclipse.californium.core.CoapHandler; import org.eclipse.californium.core.CoapObserveRelation; import org.eclipse.californium.core.CoapResponse; @@ -27,13 +28,23 @@ import org.eclipse.californium.core.coap.MediaTypeRegistry; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.query.AliasEntityId; import org.thingsboard.server.common.data.query.EntityKey; import org.thingsboard.server.common.data.query.EntityKeyType; import org.thingsboard.server.common.data.query.SingleEntityFilter; +import org.thingsboard.server.common.data.security.Authority; +import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.msg.session.FeatureType; +import org.thingsboard.server.common.transport.limits.TransportRateLimitService; +import org.thingsboard.server.common.transport.service.DefaultTransportService; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.coap.AbstractCoapIntegrationTest; import org.thingsboard.server.transport.coap.CoapTestCallback; @@ -56,6 +67,10 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.common.data.query.EntityKeyType.CLIENT_ATTRIBUTE; import static org.thingsboard.server.common.data.query.EntityKeyType.SHARED_ATTRIBUTE; @@ -67,6 +82,11 @@ public class CoapClientIntegrationTest extends AbstractCoapIntegrationTest { private static final List EXPECTED_KEYS = Arrays.asList("key1", "key2", "key3", "key4", "key5"); private static final String DEVICE_RESPONSE = "{\"value1\":\"A\",\"value2\":\"B\"}"; + @Autowired + DefaultTransportService defaultTransportService; + @MockitoSpyBean + TransportRateLimitService rateLimitService; + @Before public void beforeTest() throws Exception { CoapTestConfigProperties configProperties = CoapTestConfigProperties.builder() @@ -304,4 +324,44 @@ public class CoapClientIntegrationTest extends AbstractCoapIntegrationTest { private List getEntityKeys(EntityKeyType scope) { return CoapClientIntegrationTest.EXPECTED_KEYS.stream().map(key -> new EntityKey(scope, key)).collect(Collectors.toList()); } + + @Test + public void testCancelAttributeSubscriptionAfterTenantDeletion() throws Exception { + loginSysAdmin(); + Tenant tenant = new Tenant(); + tenant.setTitle("Mqtt Client Connection Test Tenant"); + Tenant savedTenant = saveTenant(tenant); + + User tenantAdmin = new User(); + tenantAdmin.setAuthority(Authority.TENANT_ADMIN); + tenantAdmin.setTenantId(savedTenant.getTenantId()); + tenantAdmin.setEmail("mqttTestClient@thingsboard.org"); + createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD); + + savedDevice = createDevice(RandomStringUtils.randomAlphabetic(10), "default"); + DeviceCredentials deviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class); + accessToken = deviceCredentials.getCredentialsId(); + + client = new CoapTestClient(accessToken, FeatureType.ATTRIBUTES); + CoapTestCallback callbackCoap = new CoapTestCallback(); + + client.getObserveRelation(callbackCoap); + String awaitAlias = "await Subscribe To AttributesUpdates (client.getObserveRelation)"; + await(awaitAlias) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> CoAP.ResponseCode.CONTENT.equals(callbackCoap.getResponseCode()) && + callbackCoap.getObserve() != null && 0 == callbackCoap.getObserve()); + + loginSysAdmin(); + Mockito.clearInvocations(rateLimitService); + deleteTenant(savedTenant.getId()); + + Awaitility.await("awaiting defaultTransportService.sessions is empty") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> defaultTransportService.sessions.isEmpty()); + + verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java index 36ee145c65..e4ab8c3acc 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java @@ -19,20 +19,31 @@ import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.data.security.DeviceCredentials; +import org.thingsboard.server.common.transport.limits.TransportRateLimitService; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; import java.util.concurrent.TimeUnit; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + @DaoSqlTest public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { + @MockitoSpyBean + TransportRateLimitService rateLimitService; + @Before public void beforeTest() throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() @@ -79,10 +90,14 @@ public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { Assert.assertTrue(client.isConnected()); loginSysAdmin(); + + Mockito.clearInvocations(rateLimitService); deleteTenant(savedTenant.getId()); Awaitility.await() .atMost(10, TimeUnit.SECONDS) .until(() -> !client.isConnected()); + + verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); } } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java index dab977f29c..c50a753c5c 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java @@ -55,6 +55,7 @@ import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.SessionInfoCreator; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.transport.coap.CoapSessionMsgType; @@ -562,8 +563,8 @@ public class DefaultCoapClientContext implements CoapClientContext { @Override public void onTenantDeleted(DeviceId deviceId) { - cancelRpcSubscription(state, false); - cancelAttributeSubscription(state, false); + state.setSessionCloseReason(SessionCloseReason.TENANT_DELETED); + onDeviceDeleted(deviceId); } @Override @@ -818,47 +819,39 @@ public class DefaultCoapClientContext implements CoapClientContext { } private void cancelRpcSubscription(TbCoapClientState state) { - cancelRpcSubscription(state, true); - } - - private void cancelRpcSubscription(TbCoapClientState state, boolean notifyCore) { if (state.getRpc() != null) { clientsByToken.remove(state.getRpc().getToken()); CoapExchange exchange = state.getRpc().getExchange(); state.setRpc(null); - if (notifyCore) { + if (state.shouldNotifyCore()) { transportService.process(state.getSession(), TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); } if (state.getAttrs() == null) { - closeAndCleanup(state, notifyCore); + closeAndCleanup(state); } } } private void cancelAttributeSubscription(TbCoapClientState state) { - cancelAttributeSubscription(state, true); - } - - private void cancelAttributeSubscription(TbCoapClientState state, boolean notifyCore) { if (state.getAttrs() != null) { clientsByToken.remove(state.getAttrs().getToken()); CoapExchange exchange = state.getAttrs().getExchange(); state.setAttrs(null); - if (notifyCore) { + if (state.shouldNotifyCore()) { transportService.process(state.getSession(), TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); } if (state.getRpc() == null) { - closeAndCleanup(state, notifyCore); + closeAndCleanup(state); } } } - private void closeAndCleanup(TbCoapClientState state, boolean notifyCore) { - if (notifyCore) { + private void closeAndCleanup(TbCoapClientState state) { + if (state.shouldNotifyCore()) { transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); } transportService.deregisterSession(state.getSession()); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java index d804a74283..ab6599a6cc 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java @@ -25,6 +25,7 @@ import org.thingsboard.server.common.data.device.data.PowerMode; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.transport.coap.TransportConfigurationContainer; import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; @@ -72,6 +73,9 @@ public class TbCoapClientState { @Getter @Setter private Future sleepTask; + @Getter + @Setter + private SessionCloseReason sessionCloseReason; private boolean firstEdrxDownlink = true; @@ -149,4 +153,8 @@ public class TbCoapClientState { this.missedAttributeUpdates = null; return result; } + + public boolean shouldNotifyCore() { + return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + } } diff --git a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java index 1dd3b2790e..771a38e932 100644 --- a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java +++ b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java @@ -25,17 +25,13 @@ import org.thingsboard.server.coapserver.CoapServerContext; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.transport.TransportDeviceProfileCache; import org.thingsboard.server.common.transport.TransportService; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.transport.coap.client.DefaultCoapClientContext; import org.thingsboard.server.transport.coap.client.TbCoapClientState; import org.thingsboard.server.transport.coap.client.TbCoapObservationState; -import static org.mockito.ArgumentMatchers.any; 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.verifyNoInteractions; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -43,7 +39,8 @@ public class CoapSessionListenerTest { @Mock CoapServerContext config; - @Mock CoapTransportContext transportContext; + @Mock + CoapTransportContext transportContext; @Mock TransportService transportService; @Mock @@ -60,7 +57,7 @@ public class CoapSessionListenerTest { @Test void onTenantDeleted_shouldNotInteractWithCore() { - // Arrange + // GIVEN TbCoapClientState state = mock(TbCoapClientState.class); TbCoapObservationState rpcObs = mock(TbCoapObservationState.class); @@ -75,44 +72,11 @@ public class CoapSessionListenerTest { DefaultCoapClientContext.CoapSessionListener listener = coapClientContext.new CoapSessionListener(state); - // Act - listener.onTenantDeleted(mock(DeviceId.class)); - - // Assert - verify(transportService, never()) - .process( - any(TransportProtos.SessionInfoProto.class), - any(TransportProtos.SessionEventMsg.class), - any() - ); - } - - @Test - void onTenantDeleted_shouldDeregisterSession() { - // GIVEN - TbCoapClientState state = mock(TbCoapClientState.class); - - TbCoapObservationState rpcObs = mock(TbCoapObservationState.class); - when(rpcObs.getToken()).thenReturn("rpc-token"); - when(rpcObs.getExchange()).thenReturn(mock(CoapExchange.class)); - - when(state.getRpc()).thenReturn(rpcObs); - when(state.getAttrs()).thenReturn(null); - - DefaultCoapClientContext.CoapSessionListener listener = coapClientContext.new CoapSessionListener(state); - // WHEN listener.onTenantDeleted(mock(DeviceId.class)); // THEN - verify(transportService, never()) - .process( - any(TransportProtos.SessionInfoProto.class), - any(TransportProtos.SessionEventMsg.class), - any() - ); - verify(transportService, times(1)) - .deregisterSession(any()); + verifyNoInteractions(transportService, partitionService); } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index b995bdbdb3..d3322be667 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -73,6 +73,7 @@ import org.thingsboard.server.common.transport.auth.SessionInfoCreator; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; import org.thingsboard.server.common.transport.service.SessionMetaData; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.common.transport.util.SslUtil; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceResponseMsg; @@ -1306,7 +1307,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void doDisconnect() { if (deviceSessionCtx.isConnected()) { log.debug("[{}] Client disconnected!", sessionId); - transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); + if (deviceSessionCtx.shouldNotifyCore()) { + transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); + } transportService.deregisterSession(deviceSessionCtx.getSessionInfo()); if (gatewaySessionHandler != null) { gatewaySessionHandler.onDevicesDisconnect(); @@ -1600,6 +1603,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @Override public void onTenantDeleted(DeviceId deviceId) { + deviceSessionCtx.setSessionCloseReason(SessionCloseReason.TENANT_DELETED); onDeviceDeleted(deviceId); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java index 1152cbae79..9e6e8442ae 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java @@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadCo import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.transport.mqtt.MqttTransportContext; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.transport.mqtt.TopicType; import org.thingsboard.server.transport.mqtt.adaptors.BackwardCompatibilityAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; @@ -72,6 +73,10 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext { @Setter private boolean provisionOnly = false; + @Getter + @Setter + private SessionCloseReason sessionCloseReason; + @Getter @Setter private MqttVersion mqttVersion; @@ -133,6 +138,10 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext { return payloadType.equals(TransportPayloadType.JSON); } + public boolean shouldNotifyCore() { + return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + } + @Override public void setDeviceProfile(DeviceProfile deviceProfile) { super.setDeviceProfile(deviceProfile); diff --git a/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/MqttTransportHandlerTest.java b/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/MqttTransportHandlerTest.java index 0beddac946..32628bf472 100644 --- a/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/MqttTransportHandlerTest.java +++ b/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/MqttTransportHandlerTest.java @@ -42,14 +42,17 @@ import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.device.profile.DeviceProfileData; import org.thingsboard.server.common.data.device.profile.JsonTransportPayloadConfiguration; import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.transport.TransportService; +import org.thingsboard.server.common.transport.limits.TransportRateLimitService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.util.List; +import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -72,6 +75,7 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @Slf4j @@ -95,12 +99,15 @@ public class MqttTransportHandlerTest { @Spy TransportService transportService; + @Spy + TransportRateLimitService transportRateLimitService; @BeforeEach public void setUp() throws Exception { lenient().doReturn(MSG_QUEUE_LIMIT).when(context).getMessageQueueSizePerDeviceLimit(); lenient().doReturn(transportService).when(context).getTransportService(); + lenient().doReturn(transportRateLimitService).when(context).getRateLimitService(); handler = spy(new MqttTransportHandler(context, sslHandler)); lenient().doReturn(IP_ADDR).when(handler).getAddress(any()); @@ -252,4 +259,11 @@ public class MqttTransportHandlerTest { verify(transportService, times(1)).process(any(), (TransportProtos.PostTelemetryMsg) any(), eq(expectedMd), any()); } + @Test + public void givenOnTenantDeleted_thenNoMessagesToCore() { + handler.deviceSessionCtx.setChannel(ctx); + handler.onTenantDeleted(new DeviceId(UUID.randomUUID())); + verifyNoInteractions(transportService, transportRateLimitService); + } + } \ No newline at end of file diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index 78c4b333af..729d86db34 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -201,10 +201,6 @@ public class SnmpTransportContext extends TransportContext { } private void destroyDeviceSession(DeviceSessionContext sessionContext) { - destroyDeviceSession(sessionContext, true); - } - - private void destroyDeviceSession(DeviceSessionContext sessionContext, boolean notifyCore) { if (sessionContext == null) return; log.info("Destroying SNMP device session for device {}", sessionContext.getDevice().getId()); sessionContext.close(); @@ -212,7 +208,7 @@ public class SnmpTransportContext extends TransportContext { transportService.deregisterSession(sessionContext.getSessionInfo()); snmpTransportService.cancelQueryingTasks(sessionContext); sessions.remove(sessionContext.getDeviceId()); - if (notifyCore) { + if (sessionContext.shouldNotifyCore()) { transportService.lifecycleEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), ComponentLifecycleEvent.STOPPED, true, null); } log.trace("Unregistered and removed session"); @@ -298,10 +294,6 @@ public class SnmpTransportContext extends TransportContext { destroyDeviceSession(sessionContext); } - public void onDeviceDeleted(DeviceSessionContext sessionContext, boolean notifyCore) { - destroyDeviceSession(sessionContext, notifyCore); - } - public void onDeviceProfileUpdated(DeviceProfile deviceProfile, DeviceSessionContext sessionContext) { log.debug("Handling device profile {} update event for device {}", deviceProfile.getId(), sessionContext.getDeviceId()); updateDeviceSession(sessionContext, sessionContext.getDevice(), deviceProfile); diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java index 47982320dc..80b5df124e 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java @@ -34,6 +34,7 @@ import org.thingsboard.server.common.transport.SessionMsgListener; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.service.DefaultTransportService; import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg; @@ -75,6 +76,10 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Getter private final List queryingTasks = new LinkedList<>(); + @Getter + @Setter + private SessionCloseReason sessionCloseReason; + @Builder public DeviceSessionContext(TenantId tenantId, Device device, DeviceProfile deviceProfile, String token, SnmpDeviceProfileTransportConfiguration profileTransportConfiguration, @@ -110,7 +115,8 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Override public void onTenantDeleted(DeviceId deviceId) { - snmpTransportContext.onDeviceDeleted(this, false); + sessionCloseReason = SessionCloseReason.TENANT_DELETED; + snmpTransportContext.onDeviceDeleted(this); } @Override @@ -134,6 +140,10 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S return token; } + public boolean shouldNotifyCore() { + return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + } + @Override public int nextMsgId() { return msgIdSeq.incrementAndGet(); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java new file mode 100644 index 0000000000..d6e2c2052b --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2025 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.common.transport.session; + +public enum SessionCloseReason { + + DEFAULT, TENANT_DELETED + +} From 6bb0e8b5444c090d9ebc1bd23116ce7df53e8bc4 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Thu, 8 Jan 2026 10:39:34 +0200 Subject: [PATCH 06/10] excluded messages to core while tenant deletion, added integration tests --- .../client/CoapClientIntegrationTest.java | 2 +- .../client/MqttClientConnectionTest.java | 67 +++++++++++++++++++ .../lwm2m/server/LwM2mSessionMsgListener.java | 3 +- .../lwm2m/server/client/LwM2mClient.java | 10 +++ .../session/DefaultLwM2MSessionManager.java | 9 ++- .../server/session/LwM2MSessionManager.java | 1 + .../uplink/DefaultLwM2mUplinkMsgHandler.java | 10 ++- .../server/uplink/LwM2mUplinkMsgHandler.java | 2 + .../transport/mqtt/MqttTransportHandler.java | 15 +++-- .../AbstractGatewayDeviceSessionContext.java | 4 +- .../AbstractGatewaySessionHandler.java | 10 ++- .../mqtt/session/DeviceSessionCtx.java | 9 --- .../snmp/session/DeviceSessionContext.java | 10 +-- .../session/DeviceAwareSessionContext.java | 7 ++ 14 files changed, 127 insertions(+), 32 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java index 5404e0ef31..1ac5685d82 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java @@ -347,7 +347,7 @@ public class CoapClientIntegrationTest extends AbstractCoapIntegrationTest { CoapTestCallback callbackCoap = new CoapTestCallback(); client.getObserveRelation(callbackCoap); - String awaitAlias = "await Subscribe To AttributesUpdates (client.getObserveRelation)"; + String awaitAlias = "await subscribe to attributes updates"; await(awaitAlias) .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) .until(() -> CoAP.ResponseCode.CONTENT.equals(callbackCoap.getResponseCode()) && diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java index e4ab8c3acc..e37c41d4cb 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java @@ -15,11 +15,15 @@ */ package org.thingsboard.server.transport.mqtt.mqttv3.client; +import com.google.gson.JsonObject; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.mqtt.MqttQoS; import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.thingsboard.server.common.data.Tenant; @@ -27,6 +31,8 @@ import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.transport.limits.TransportRateLimitService; +import org.thingsboard.server.common.transport.service.DefaultTransportService; +import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; @@ -41,6 +47,10 @@ import static org.mockito.Mockito.verify; @DaoSqlTest public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { + @Autowired + DefaultTransportService defaultTransportService; + @Autowired + DeviceService deviceService; @MockitoSpyBean TransportRateLimitService rateLimitService; @@ -94,10 +104,67 @@ public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { Mockito.clearInvocations(rateLimitService); deleteTenant(savedTenant.getId()); + Awaitility.await("awaiting defaultTransportService.sessions is empty") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> defaultTransportService.sessions.isEmpty()); + Awaitility.await() .atMost(10, TimeUnit.SECONDS) .until(() -> !client.isConnected()); verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); } + + @Test + public void testGatewayShouldBeDisconnectedAfterTenantDeletion() throws Exception { + loginSysAdmin(); + Tenant tenant = new Tenant(); + tenant.setTitle("Mqtt Client Connection Test Tenant"); + Tenant savedTenant = saveTenant(tenant); + + User tenantAdmin = new User(); + tenantAdmin.setAuthority(Authority.TENANT_ADMIN); + tenantAdmin.setTenantId(savedTenant.getTenantId()); + tenantAdmin.setEmail("mqttTestClient@thingsboard.org"); + createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD); + + savedDevice = createDevice(RandomStringUtils.randomAlphabetic(10), "default", true); + DeviceCredentials deviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class); + accessToken = deviceCredentials.getCredentialsId(); + + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(accessToken); + + String deviceName = "gateway_device_" + RandomStringUtils.randomAlphabetic(5); + client.publish("v1/gateway/connect", createGatewayConnectPayload(deviceName).toString().getBytes()); + + Awaitility.await("awaiting till device is created") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> deviceService.findDeviceByTenantIdAndName(savedTenant.getTenantId(), deviceName) != null); + + loginSysAdmin(); + + Mockito.clearInvocations(rateLimitService); + deleteTenant(savedTenant.getId()); + + Awaitility.await("awaiting defaultTransportService.sessions is empty") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> defaultTransportService.sessions.isEmpty()); + + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> !client.isConnected()); + + verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); + } + + protected JsonObject createGatewayConnectPayload(String deviceName){ + JsonObject payload = new JsonObject(); + payload.addProperty("device", deviceName); + return payload; + } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java index b49b02d3a5..1314946f34 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java @@ -117,6 +117,7 @@ public class LwM2mSessionMsgListener implements GenericFutureListener supportedClientObjects; + @Getter + @Setter + private SessionCloseReason sessionCloseReason; + public Object clone() throws CloneNotSupportedException { return super.clone(); } @@ -445,6 +451,10 @@ public class LwM2mClient { return this.supportedClientObjects != null ? this.supportedClientObjects.get(objectid) : null; } + public boolean shouldNotifyCore() { + return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + } + private void setSupportedClientObjects(){ if (this.registration.getSupportedObject() != null && this.registration.getSupportedObject().size() > 0) { this.supportedClientObjects = this.registration.getSupportedObject(); diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java index 285b3da751..5d95067e7b 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java @@ -65,7 +65,14 @@ public class DefaultLwM2MSessionManager implements LwM2MSessionManager { @Override public void deregister(TransportProtos.SessionInfoProto sessionInfo) { - transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); + deregister(sessionInfo, true); + } + + @Override + public void deregister(TransportProtos.SessionInfoProto sessionInfo, boolean notifyCore) { + if (notifyCore) { + transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); + } transportService.deregisterSession(sessionInfo); } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java index 110f8ec976..050abf9953 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java @@ -23,5 +23,6 @@ public interface LwM2MSessionManager { void deregister(TransportProtos.SessionInfoProto sessionInfo); + void deregister(TransportProtos.SessionInfoProto sessionInfo, boolean notifyCore); } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java index b7e56c139a..4c6d62bd39 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java @@ -65,6 +65,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.ota.OtaPackageUtil; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; import org.thingsboard.server.queue.util.TbLwM2mTransportComponent; @@ -293,7 +294,7 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl clientContext.unregister(client, registration); SessionInfoProto sessionInfo = client.getSession(); if (sessionInfo != null) { - sessionManager.deregister(sessionInfo); + sessionManager.deregister(sessionInfo, client.shouldNotifyCore()); sessionStore.remove(registration.getEndpoint()); log.info("Client close session: [{}] unReg [{}] name [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType()); } else { @@ -447,6 +448,13 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl clearAndUnregister(clientContext.getClientByDeviceId(deviceId.getId())); } + @Override + public void onTenantDelete(DeviceId deviceId) { + LwM2mClient client = clientContext.getClientByDeviceId(deviceId.getId()); + client.setSessionCloseReason(SessionCloseReason.TENANT_DELETED); + clearAndUnregister(client); + } + @Override public void onResourceUpdate(TransportProtos.ResourceUpdateMsg resourceUpdateMsgOpt) { String idVer = resourceUpdateMsgOpt.getResourceKey(); diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java index 0c28372bce..5a2a432347 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java @@ -59,6 +59,8 @@ public interface LwM2mUplinkMsgHandler { void onDeviceDelete(DeviceId deviceId); + void onTenantDelete(DeviceId deviceId); + void onResourceUpdate(TransportProtos.ResourceUpdateMsg resourceUpdateMsgOpt); void onResourceDelete(TransportProtos.ResourceDeleteMsg resourceDeleteMsgOpt); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index d3322be667..9b3cc795af 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -1307,18 +1307,21 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void doDisconnect() { if (deviceSessionCtx.isConnected()) { log.debug("[{}] Client disconnected!", sessionId); - if (deviceSessionCtx.shouldNotifyCore()) { + boolean notifyCore = deviceSessionCtx.shouldNotifyCore(); + if (notifyCore) { transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); } transportService.deregisterSession(deviceSessionCtx.getSessionInfo()); if (gatewaySessionHandler != null) { - gatewaySessionHandler.onDevicesDisconnect(); + gatewaySessionHandler.onDevicesDisconnect(notifyCore); } if (sparkplugSessionHandler != null) { - // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp() - sparkplugSessionHandler.sendSparkplugStateOnTelemetry(deviceSessionCtx.getSessionInfo(), - deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime()); - sparkplugSessionHandler.onDevicesDisconnect(); + if (notifyCore) { + // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp() + sparkplugSessionHandler.sendSparkplugStateOnTelemetry(deviceSessionCtx.getSessionInfo(), + deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime()); + } + sparkplugSessionHandler.onDevicesDisconnect(notifyCore); } deviceSessionCtx.setDisconnected(); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index fee6ab13fa..78c2b98143 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -27,6 +27,7 @@ import org.thingsboard.server.common.transport.SessionMsgListener; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; @@ -147,7 +148,8 @@ public abstract class AbstractGatewayDeviceSessionContext { @@ -207,7 +211,7 @@ public abstract class AbstractGatewaySessionHandler deregisterSession(deviceName, ctx, notifyCore)); } catch (Exception e) { log.error("Gateway disconnect failure", e); } @@ -814,7 +818,7 @@ public abstract class AbstractGatewaySessionHandler queryingTasks = new LinkedList<>(); - @Getter - @Setter - private SessionCloseReason sessionCloseReason; - @Builder public DeviceSessionContext(TenantId tenantId, Device device, DeviceProfile deviceProfile, String token, SnmpDeviceProfileTransportConfiguration profileTransportConfiguration, @@ -115,7 +111,7 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Override public void onTenantDeleted(DeviceId deviceId) { - sessionCloseReason = SessionCloseReason.TENANT_DELETED; + setSessionCloseReason(SessionCloseReason.TENANT_DELETED); snmpTransportContext.onDeviceDeleted(this); } @@ -140,10 +136,6 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S return token; } - public boolean shouldNotifyCore() { - return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; - } - @Override public int nextMsgId() { return msgIdSeq.incrementAndGet(); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java index 008c410e56..f338749138 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java @@ -50,6 +50,9 @@ public abstract class DeviceAwareSessionContext implements SessionContext { @Getter @Setter protected volatile TransportProtos.SessionInfoProto sessionInfo; + @Getter + @Setter + protected SessionCloseReason sessionCloseReason; @Setter private volatile boolean connected; @@ -87,6 +90,10 @@ public abstract class DeviceAwareSessionContext implements SessionContext { this.connected = false; } + public boolean shouldNotifyCore() { + return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + } + public boolean isSparkplug() { DeviceProfileTransportConfiguration transportConfiguration = this.deviceProfile.getProfileData().getTransportConfiguration(); if (transportConfiguration instanceof MqttDeviceProfileTransportConfiguration) { From 25223e75a994d14b69cad9b1644d214c3b08f15f Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Thu, 8 Jan 2026 11:13:39 +0200 Subject: [PATCH 07/10] refactoring --- .../transport/mqtt/MqttTransportHandler.java | 4 +-- .../AbstractGatewayDeviceSessionContext.java | 2 +- .../AbstractGatewaySessionHandler.java | 28 ++++--------------- 3 files changed, 9 insertions(+), 25 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 9b3cc795af..4efe03de17 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -1313,7 +1313,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } transportService.deregisterSession(deviceSessionCtx.getSessionInfo()); if (gatewaySessionHandler != null) { - gatewaySessionHandler.onDevicesDisconnect(notifyCore); + gatewaySessionHandler.onDevicesDisconnect(); } if (sparkplugSessionHandler != null) { if (notifyCore) { @@ -1321,7 +1321,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement sparkplugSessionHandler.sendSparkplugStateOnTelemetry(deviceSessionCtx.getSessionInfo(), deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime()); } - sparkplugSessionHandler.onDevicesDisconnect(notifyCore); + sparkplugSessionHandler.onDevicesDisconnect(); } deviceSessionCtx.setDisconnected(); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index 78c2b98143..c955dc942c 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -149,7 +149,7 @@ public abstract class AbstractGatewayDeviceSessionContext { @@ -211,7 +207,7 @@ public abstract class AbstractGatewaySessionHandler deregisterSession(deviceName, ctx, notifyCore)); + devices.forEach(this::deregisterSession); } catch (Exception e) { log.error("Gateway disconnect failure", e); } @@ -231,11 +227,7 @@ public abstract class AbstractGatewaySessionHandler Date: Thu, 4 Jun 2026 19:07:25 +0300 Subject: [PATCH 08/10] fixed pr comments --- .../coap/client/DefaultCoapClientContext.java | 6 ++++++ .../transport/coap/client/TbCoapClientState.java | 14 +++++++++++++- .../lwm2m/server/LwM2mSessionMsgListener.java | 2 +- .../transport/lwm2m/server/client/LwM2mClient.java | 2 +- .../uplink/DefaultLwM2mUplinkMsgHandler.java | 10 +++++++--- .../transport/mqtt/MqttTransportHandler.java | 4 +++- .../session/AbstractGatewaySessionHandler.java | 2 +- .../transport/mqtt/MqttTransportHandlerTest.java | 11 +++++++++++ .../transport/service/DefaultTransportService.java | 7 +++++-- .../session/DeviceAwareSessionContext.java | 4 ++-- .../transport/session/SessionCloseReason.java | 6 +++++- 11 files changed, 55 insertions(+), 13 deletions(-) diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java index c50a753c5c..6ed2a5cc14 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java @@ -50,6 +50,7 @@ import org.thingsboard.server.common.transport.DeviceDeletedEvent; import org.thingsboard.server.common.transport.DeviceProfileUpdatedEvent; import org.thingsboard.server.common.transport.DeviceUpdatedEvent; import org.thingsboard.server.common.transport.SessionMsgListener; +import org.thingsboard.server.common.transport.TenantDeletedEvent; import org.thingsboard.server.common.transport.TransportDeviceProfileCache; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; @@ -142,6 +143,11 @@ public class DefaultCoapClientContext implements CoapClientContext { clients.remove(event.getDeviceId()); } + @EventListener(TenantDeletedEvent.class) + public void onApplicationEvent(TenantDeletedEvent event) { + clients.values().removeIf(state -> event.getTenantId().equals(state.getTenantId())); + } + @Override public boolean registerAttributeObservation(TbCoapClientState clientState, String token, CoapExchange exchange) { return registerFeatureObservation(clientState, token, exchange, FeatureType.ATTRIBUTES); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java index ab6599a6cc..3e7e4258b8 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java @@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.device.data.CoapDeviceTransportConfigu import org.thingsboard.server.common.data.device.data.PowerMode; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; @@ -34,6 +35,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.concurrent.Future; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -155,6 +157,16 @@ public class TbCoapClientState { } public boolean shouldNotifyCore() { - return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + return SessionCloseReason.shouldNotifyCore(sessionCloseReason); + } + + public TenantId getTenantId() { + if (credentials != null && credentials.getDeviceInfo() != null) { + return credentials.getDeviceInfo().getTenantId(); + } + if (session != null) { + return TenantId.fromUUID(new UUID(session.getTenantIdMSB(), session.getTenantIdLSB())); + } + return null; } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java index 1314946f34..f79b961913 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java @@ -117,7 +117,7 @@ public class LwM2mSessionMsgListener implements GenericFutureListener { TenantId sessionTenantId = getTenantId(md.getSessionInfo()); if (sessionTenantId.equals(tenantId)) { @@ -976,9 +977,11 @@ public class DefaultTransportService extends TransportActivityManager implements transportCallbackExecutor.submit(() -> { md.getListener().onTenantDeleted(sessionDeviceId); }); - eventPublisher.publishEvent(new DeviceDeletedEvent(sessionDeviceId)); } }); + // notify transports holding per-device state (e.g. CoAP's client states) to clean it up in one shot, + // since no per-device delete messages are sent during tenant deletion + eventPublisher.publishEvent(new TenantDeletedEvent(tenantId)); } public void onProfileUpdate(DeviceProfile deviceProfile) { diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java index f338749138..5ac3b6200b 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java @@ -52,7 +52,7 @@ public abstract class DeviceAwareSessionContext implements SessionContext { protected volatile TransportProtos.SessionInfoProto sessionInfo; @Getter @Setter - protected SessionCloseReason sessionCloseReason; + protected volatile SessionCloseReason sessionCloseReason; @Setter private volatile boolean connected; @@ -91,7 +91,7 @@ public abstract class DeviceAwareSessionContext implements SessionContext { } public boolean shouldNotifyCore() { - return sessionCloseReason == null || sessionCloseReason != SessionCloseReason.TENANT_DELETED; + return SessionCloseReason.shouldNotifyCore(sessionCloseReason); } public boolean isSparkplug() { diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java index d6e2c2052b..d6caa8bbf9 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java @@ -17,6 +17,10 @@ package org.thingsboard.server.common.transport.session; public enum SessionCloseReason { - DEFAULT, TENANT_DELETED + TENANT_DELETED; + + public static boolean shouldNotifyCore(SessionCloseReason reason) { + return reason != TENANT_DELETED; + } } From aaa9a3ac199f946a6a87606833a8a3acefec8ce7 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Fri, 5 Jun 2026 14:54:10 +0300 Subject: [PATCH 09/10] added missed event --- .../common/transport/TenantDeletedEvent.java | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java new file mode 100644 index 0000000000..017e673399 --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java @@ -0,0 +1,32 @@ +/** + * Copyright © 2016-2025 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.common.transport; + +import lombok.Getter; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.queue.discovery.event.TbApplicationEvent; + +public final class TenantDeletedEvent extends TbApplicationEvent { + + private static final long serialVersionUID = -3801673609913423334L; + @Getter + private final TenantId tenantId; + + public TenantDeletedEvent(TenantId tenantId) { + super(new Object()); + this.tenantId = tenantId; + } +} From e7c4062e215fe6f7a54752cec4326c749fd0cd91 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Fri, 5 Jun 2026 16:57:57 +0300 Subject: [PATCH 10/10] fixed license --- .../server/transport/coap/CoapSessionListenerTest.java | 2 +- .../thingsboard/server/common/transport/TenantDeletedEvent.java | 2 +- .../server/common/transport/session/SessionCloseReason.java | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java index 771a38e932..e28d880a87 100644 --- a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java +++ b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java @@ -1,5 +1,5 @@ /** - * Copyright © 2016-2025 The Thingsboard Authors + * 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. diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java index 017e673399..0587fd9aa6 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TenantDeletedEvent.java @@ -1,5 +1,5 @@ /** - * Copyright © 2016-2025 The Thingsboard Authors + * 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. diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java index d6caa8bbf9..cd729f58ac 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java @@ -1,5 +1,5 @@ /** - * Copyright © 2016-2025 The Thingsboard Authors + * 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.