Browse Source

Fix fetch pack size

pull/7041/head
Andrii Shvaika 4 years ago
parent
commit
3258a987e3
  1. 4
      application/src/main/java/org/thingsboard/server/service/partition/AbstractPartitionBasedService.java
  2. 5
      application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java

4
application/src/main/java/org/thingsboard/server/service/partition/AbstractPartitionBasedService.java

@ -119,7 +119,9 @@ public abstract class AbstractPartitionBasedService<T extends EntityId> extends
// We no longer manage current partition of entities; // We no longer manage current partition of entities;
removedPartitions.forEach(partition -> { removedPartitions.forEach(partition -> {
Set<T> entities = partitionedEntities.remove(partition); Set<T> entities = partitionedEntities.remove(partition);
entities.forEach(this::cleanupEntityOnPartitionRemoval); if (entities != null) {
entities.forEach(this::cleanupEntityOnPartitionRemoval);
}
List<ListenableFuture<?>> fetchTasks = partitionedFetchTasks.remove(partition); List<ListenableFuture<?>> fetchTasks = partitionedFetchTasks.remove(partition);
if (fetchTasks != null) { if (fetchTasks != null) {
fetchTasks.forEach(f -> f.cancel(true)); fetchTasks.forEach(f -> f.cancel(true));

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

@ -50,7 +50,6 @@ import org.thingsboard.server.common.data.page.PageDataIterable;
import org.thingsboard.server.common.data.query.EntityData; import org.thingsboard.server.common.data.query.EntityData;
import org.thingsboard.server.common.data.query.EntityDataPageLink; import org.thingsboard.server.common.data.query.EntityDataPageLink;
import org.thingsboard.server.common.data.query.EntityDataQuery; import org.thingsboard.server.common.data.query.EntityDataQuery;
import org.thingsboard.server.common.data.query.EntityDataSortOrder;
import org.thingsboard.server.common.data.query.EntityKey; import org.thingsboard.server.common.data.query.EntityKey;
import org.thingsboard.server.common.data.query.EntityKeyType; import org.thingsboard.server.common.data.query.EntityKeyType;
import org.thingsboard.server.common.data.query.EntityListFilter; import org.thingsboard.server.common.data.query.EntityListFilter;
@ -340,7 +339,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
@Override @Override
protected Map<TopicPartitionInfo, List<ListenableFuture<?>>> onAddedPartitions(Set<TopicPartitionInfo> addedPartitions) { protected Map<TopicPartitionInfo, List<ListenableFuture<?>>> onAddedPartitions(Set<TopicPartitionInfo> addedPartitions) {
var result = new HashMap<TopicPartitionInfo, List<ListenableFuture<?>>>(); var result = new HashMap<TopicPartitionInfo, List<ListenableFuture<?>>>();
PageDataIterable<DeviceIdInfo> deviceIdInfos = new PageDataIterable<>(deviceService::findDeviceIdInfos, 1); PageDataIterable<DeviceIdInfo> deviceIdInfos = new PageDataIterable<>(deviceService::findDeviceIdInfos, initFetchPackSize);
Map<TopicPartitionInfo, List<DeviceIdInfo>> tpiDeviceMap = new HashMap<>(); Map<TopicPartitionInfo, List<DeviceIdInfo>> tpiDeviceMap = new HashMap<>();
for (DeviceIdInfo idInfo : deviceIdInfos) { for (DeviceIdInfo idInfo : deviceIdInfos) {
@ -569,7 +568,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
ef.setEntityList(deviceIds.stream().map(DeviceIdInfo::getDeviceId).map(DeviceId::getId).map(UUID::toString).collect(Collectors.toList())); ef.setEntityList(deviceIds.stream().map(DeviceIdInfo::getDeviceId).map(DeviceId::getId).map(UUID::toString).collect(Collectors.toList()));
EntityDataQuery query = new EntityDataQuery(ef, EntityDataQuery query = new EntityDataQuery(ef,
new EntityDataPageLink(deviceIds.size(), 0, null, new EntityDataSortOrder(new EntityKey(EntityKeyType.ENTITY_FIELD, "id"))), new EntityDataPageLink(deviceIds.size(), 0, null, null),
PERSISTENT_ENTITY_FIELDS, PERSISTENT_ENTITY_FIELDS,
persistToTelemetry ? PERSISTENT_TELEMETRY_KEYS : PERSISTENT_ATTRIBUTE_KEYS, Collections.emptyList()); persistToTelemetry ? PERSISTENT_TELEMETRY_KEYS : PERSISTENT_ATTRIBUTE_KEYS, Collections.emptyList());
PageData<EntityData> queryResult = entityQueryRepository.findEntityDataByQueryInternal(query); PageData<EntityData> queryResult = entityQueryRepository.findEntityDataByQueryInternal(query);

Loading…
Cancel
Save