Merge 31bc775035 into be70822bc6

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<SubscriptionManagerService> toSubscriptionManagerService,
                                                        Supplier<TransportProtos.ToCoreMsg> 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());

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<Long> listOfNNumbers(int N) {
         return LongStream.range(0, N).boxed().toList();