From 31bc775035d738b1cbfa1a3f988787a741e459e4 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Tue, 7 Jul 2026 14:58:23 +0300 Subject: [PATCH] Skip subscription forwarding when tenant no longer exists A device-state attribute save can complete after its tenant was deleted. The post-save WS-update callback then resolves an evicted tenant profile via partitionService.resolve -> getRoutingInfo and throws TenantNotFoundException uncaught on the ws-callback thread. Guard resolve() in forwardToSubscriptionManagerService: a deleted tenant has no partition to route to and no subscribers to notify, so skip the forward instead of propagating. Mirrors the existing tenant-gone handling in DefaultDeviceStateService.checkStates(). --- .../AbstractSubscriptionService.java | 11 +++++++++- ...faultTelemetrySubscriptionServiceTest.java | 22 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) 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/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionServiceTest.java index dcfe9bd2bd..958a41449f 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; @@ -1155,6 +1157,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();