Browse Source

Propagate inactivity timeout on scheduled executor and enforce whole-second profile timeout

- Run profile inactivity timeout propagation on the single-threaded scheduled executor instead of the notifications consumer thread, so it stays off the consumer thread and serializes with the periodic state checker; a supersede guard drops stale tasks
- Reject device profile inactivity timeout values that are not a whole number of seconds
- Remove @Lazy on the device profile cache field
- Set inactivityTimeoutMs directly in the device profile form instead of behind a redundant profileData guard
pull/15527/head
Oleksandra Matviienko 1 month ago
parent
commit
7e3f2d72bf
  1. 30
      application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java
  2. 8
      application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java
  3. 3
      dao/src/main/java/org/thingsboard/server/dao/service/validator/DeviceProfileDataValidator.java
  4. 10
      dao/src/test/java/org/thingsboard/server/dao/service/validator/DeviceProfileDataValidatorTest.java
  5. 6
      ui-ngx/src/app/modules/home/components/profile/device-profile.component.ts

30
application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java

@ -175,7 +175,6 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
@Lazy
private TelemetrySubscriptionService tsSubService;
@Autowired
@Lazy
private TbDeviceProfileCache deviceProfileCache;
@Value("#{${state.defaultInactivityTimeoutInSec} * 1000}")
@ -376,18 +375,27 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
if (previousResolved != null && previousResolved == resolvedNew) {
return;
}
deviceStates.forEach((deviceId, stateData) -> {
if (!profileId.equals(stateData.getDeviceProfileId())) {
return;
}
if (stateData.isInactivityTimeoutOverridden()) {
return;
}
if (stateData.getState().getInactivityTimeout() == resolvedNew) {
// Run the per-device propagation on the single-threaded scheduledExecutor instead of the notifications
// consumer thread: it keeps the consumer responsive, serializes with checkStates() so the iteration never
// races the periodic checker over the same DeviceState, and preserves event order so devices converge on
// the latest resolved timeout. The guard below drops a task whose value was already superseded.
scheduledExecutor.submit(() -> {
if (!Objects.equals(profileResolvedInactivityTimeoutMs.get(profileId), resolvedNew)) {
return;
}
stateData.getState().setInactivityTimeout(resolvedNew);
checkAndUpdateState(deviceId, stateData);
deviceStates.forEach((deviceId, stateData) -> {
if (!profileId.equals(stateData.getDeviceProfileId())) {
return;
}
if (stateData.isInactivityTimeoutOverridden()) {
return;
}
if (stateData.getState().getInactivityTimeout() == resolvedNew) {
return;
}
stateData.getState().setInactivityTimeout(resolvedNew);
checkAndUpdateState(deviceId, stateData);
});
});
}

8
application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java

@ -17,6 +17,7 @@ package org.thingsboard.server.service.state;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@ -159,6 +160,13 @@ class DefaultDeviceStateServiceTest {
deviceStateCallbackExecutor = MoreExecutors.newDirectExecutorService();
ReflectionTestUtils.setField(service, "deviceStateCallbackExecutor", deviceStateCallbackExecutor);
ListeningScheduledExecutorService scheduledExecutor = mock(ListeningScheduledExecutorService.class);
lenient().doAnswer(inv -> {
((Runnable) inv.getArgument(0)).run();
return Futures.immediateVoidFuture();
}).when(scheduledExecutor).submit(any(Runnable.class));
ReflectionTestUtils.setField(service, "scheduledExecutor", scheduledExecutor);
lenient().when(partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(tpi);
ConcurrentMap<TopicPartitionInfo, Set<DeviceId>> partitionedEntities = new ConcurrentHashMap<>();

3
dao/src/main/java/org/thingsboard/server/dao/service/validator/DeviceProfileDataValidator.java

@ -210,6 +210,9 @@ public class DeviceProfileDataValidator extends AbstractHasOtaPackageValidator<D
if (inactivityTimeoutMs != null && inactivityTimeoutMs <= 0) {
throw new DataValidationException("Device profile inactivity timeout must be greater than 0!");
}
if (inactivityTimeoutMs != null && inactivityTimeoutMs % 1000 != 0) {
throw new DataValidationException("Device profile inactivity timeout must be specified in whole seconds!");
}
if (deviceProfile.getDefaultRuleChainId() != null) {
validateRuleChain(tenantId, deviceProfile.getTenantId(), deviceProfile.getDefaultRuleChainId());

10
dao/src/test/java/org/thingsboard/server/dao/service/validator/DeviceProfileDataValidatorTest.java

@ -134,10 +134,18 @@ class DeviceProfileDataValidatorTest {
@Test
void testValidate_InactivityTimeoutMs_Positive_Ok() {
DeviceProfile deviceProfile = baseDefaultProfile();
deviceProfile.getProfileData().setInactivityTimeoutMs(1L);
deviceProfile.getProfileData().setInactivityTimeoutMs(1000L);
validator.validateDataImpl(tenantId, deviceProfile);
}
@Test
void testValidate_InactivityTimeoutMs_NotMultipleOfSecond_Error() {
DeviceProfile deviceProfile = baseDefaultProfile();
deviceProfile.getProfileData().setInactivityTimeoutMs(1500L);
assertThatThrownBy(() -> validator.validateDataImpl(tenantId, deviceProfile))
.hasMessageContaining("Device profile inactivity timeout must be specified in whole seconds");
}
@Test
void testValidate_InactivityTimeoutMs_Zero_Error() {
DeviceProfile deviceProfile = baseDefaultProfile();

6
ui-ngx/src/app/modules/home/components/profile/device-profile.component.ts

@ -232,10 +232,8 @@ export class DeviceProfileComponent extends EntityComponent<DeviceProfile> {
if (formValue.defaultEdgeRuleChainId) {
formValue.defaultEdgeRuleChainId = new RuleChainId(formValue.defaultEdgeRuleChainId);
}
if (formValue.profileData) {
const sec = formValue.inactivityTimeoutSec;
formValue.profileData.inactivityTimeoutMs = sec && sec > 0 ? sec * SECOND : null;
}
const sec = formValue.inactivityTimeoutSec;
formValue.profileData.inactivityTimeoutMs = sec && sec > 0 ? sec * SECOND : null;
delete formValue.inactivityTimeoutSec;
const deviceProvisionConfiguration: DeviceProvisionConfiguration = formValue.profileData.provisionConfiguration;
formValue.provisionType = deviceProvisionConfiguration.type;

Loading…
Cancel
Save