diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index 286260cf8e..ac18ec71d1 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -386,13 +386,14 @@ public class DefaultDeviceStateService extends TbApplicationEventListener submitNextPage(addedPartitions, tenant, pageLink, scheduledExecutor)); + scheduledExecutor.submit(() -> processPageAndSubmitNextPage(addedPartitions, tenant, pageLink, scheduledExecutor)); } return true; } - private void submitNextPage(final Set addedPartitions, final Tenant tenant, final PageLink pageLink, final ExecutorService executor) { + private void processPageAndSubmitNextPage(final Set addedPartitions, final Tenant tenant, final PageLink pageLink, final ExecutorService executor) { + log.trace("[{}] Process page {} from {}", tenant, pageLink.getPage(), pageLink.getPageSize()); List> fetchFutures = new ArrayList<>(); PageData page = deviceService.findDevicesByTenantId(tenant.getId(), pageLink); for (Device device : page.getData()) { @@ -433,7 +434,8 @@ public class DefaultDeviceStateService extends TbApplicationEventListener submitNextPage(addedPartitions, tenant, nextPageLink, executor)); + log.trace("[{}] Submit next page {} from {}", tenant, nextPageLink.getPage(), nextPageLink.getPageSize()); + executor.submit(() -> processPageAndSubmitNextPage(addedPartitions, tenant, nextPageLink, executor)); } } @@ -449,8 +451,13 @@ public class DefaultDeviceStateService extends TbApplicationEventListener ConcurrentHashMap.newKeySet()).add(state.getDeviceId()); - deviceStates.put(state.getDeviceId(), state); + Set deviceIds = partitionedDevices.get(tpi); + if (deviceIds != null) { + deviceIds.add(state.getDeviceId()); + deviceStates.put(state.getDeviceId(), state); + } else { + new RuntimeException("Device belongs to external partition " + tpi.getFullTopicName() + "!"); + } } void updateInactivityStateIfExpired() {