@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.ApiUsageStateValue;
import org.thingsboard.server.common.data.AttributeScope ;
import org.thingsboard.server.common.data.AttributeScope ;
import org.thingsboard.server.common.data.EntityType ;
import org.thingsboard.server.common.data.EntityType ;
import org.thingsboard.server.common.data.EntityView ;
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.ApiUsageStateId ;
import org.thingsboard.server.common.data.id.CustomerId ;
import org.thingsboard.server.common.data.id.CustomerId ;
import org.thingsboard.server.common.data.id.DeviceId ;
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.immediateFailedFuture ;
import static com.google.common.util.concurrent.Futures.immediateFuture ;
import static com.google.common.util.concurrent.Futures.immediateFuture ;
import static org.assertj.core.api.Assertions.assertThat ;
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.assertj.core.api.Assertions.assertThatThrownBy ;
import static org.mockito.ArgumentMatchers.any ;
import static org.mockito.ArgumentMatchers.any ;
import static org.mockito.ArgumentMatchers.eq ;
import static org.mockito.ArgumentMatchers.eq ;
@ -1155,6 +1157,26 @@ class DefaultTelemetrySubscriptionServiceTest {
then ( deviceStateManager ) . shouldHaveNoInteractions ( ) ;
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
// used to emulate versions returned by save APIs
private static List < Long > listOfNNumbers ( int N ) {
private static List < Long > listOfNNumbers ( int N ) {
return LongStream . range ( 0 , N ) . boxed ( ) . toList ( ) ;
return LongStream . range ( 0 , N ) . boxed ( ) . toList ( ) ;