From a6a6112ecc81092add19940df5d709eb0d35f893 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Mon, 24 Feb 2025 12:11:05 +0200 Subject: [PATCH] Cleanup for EDQS --- .../service/edqs/DefaultEdqsService.java | 8 +-- .../src/main/resources/thingsboard.yml | 10 +++ common/edqs/pom.xml | 6 +- .../server/edqs/repo/TenantRepo.java | 17 ++--- .../edqs/state/KafkaEdqsStateService.java | 1 - .../server/queue/edqs/EdqsQueue.java | 4 +- .../queue/environment/DistributedLock.java | 2 +- .../environment/DistributedLockService.java | 2 +- .../DummyDistributedLockService.java | 6 +- .../environment/ZkDistributedLockService.java | 6 +- .../queue/kafka/TbKafkaConsumerTemplate.java | 2 +- .../common/stats/DummyMessagesStats.java | 54 --------------- .../thingsboard/common/util/JacksonUtil.java | 4 -- .../attributes/CachedAttributesService.java | 2 +- .../server/dao/entity/BaseEntityService.java | 6 +- .../server/dao/entity/EntityDaoRegistry.java | 13 ++-- .../server/dao/model/ModelConstants.java | 2 - .../dao/model/sql/AlarmTypeCompositeKey.java | 33 --------- .../server/dao/sql/rule/JpaRuleNodeDao.java | 5 -- .../dao/sql/rule/RuleNodeRepository.java | 3 - .../dao/sql/user/JpaUserAuthSettingsDao.java | 16 ----- .../widget/WidgetsBundleWidgetRepository.java | 7 -- .../sqlts/latest/TsKvLatestRepository.java | 7 -- edqs/pom.xml | 48 ------------- edqs/src/main/resources/edqs.yml | 67 +++---------------- 25 files changed, 53 insertions(+), 278 deletions(-) delete mode 100644 common/stats/src/main/java/org/thingsboard/server/common/stats/DummyMessagesStats.java delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/model/sql/AlarmTypeCompositeKey.java diff --git a/application/src/main/java/org/thingsboard/server/service/edqs/DefaultEdqsService.java b/application/src/main/java/org/thingsboard/server/service/edqs/DefaultEdqsService.java index a6a9b9bb23..92905dc336 100644 --- a/application/src/main/java/org/thingsboard/server/service/edqs/DefaultEdqsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edqs/DefaultEdqsService.java @@ -55,9 +55,9 @@ import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.msg.edqs.EdqsService; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.dao.attributes.AttributesService; -import org.thingsboard.server.edqs.util.EdqsConverter; import org.thingsboard.server.edqs.processor.EdqsProducer; import org.thingsboard.server.edqs.state.EdqsPartitionService; +import org.thingsboard.server.edqs.util.EdqsConverter; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.EdqsEventMsg; import org.thingsboard.server.gen.transport.TransportProtos.EdqsRequestMsg; @@ -103,7 +103,7 @@ public class DefaultEdqsService implements EdqsService { private EdqsProducer eventsProducer; private TbQueueRequestTemplate, TbProtoQueueMsg> requestTemplate; private ExecutorService executor; - private DistributedLock syncLock; + private DistributedLock syncLock; @PostConstruct private void init() { @@ -313,14 +313,14 @@ public class DefaultEdqsService implements EdqsService { .flatMap(KvEntry::getJsonValue) .map(value -> JacksonUtil.fromString(value, EdqsSyncState.class)) .orElse(null); - log.info("getSyncState = {}", state); + log.info("EDQS sync state: {}", state); return state; } @SneakyThrows private void saveSyncState(EdqsSyncStatus status) { EdqsSyncState state = new EdqsSyncState(status); - log.info("saveSyncState {}", state); + log.info("New EDQS sync state: {}", state); attributesService.save(TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID, AttributeScope.SERVER_SCOPE, new BaseAttributeKvEntry( new JsonDataEntry("edqsSyncState", JacksonUtil.toString(state)), System.currentTimeMillis())).get(30, TimeUnit.SECONDS); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 46ab8f25de..e12c6aee2d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1586,6 +1586,16 @@ queue: - key: max.poll.records # Amount of records to be returned in a single poll. For Housekeeper reprocessing topic, we should consume messages (tasks) one by one value: "${TB_QUEUE_KAFKA_HOUSEKEEPER_REPROCESSING_MAX_POLL_RECORDS:1}" + edqs.events: + # Key-value properties for Kafka consumer for edqs.events topic + - key: max.poll.records + # Max poll records for edqs.events topic + value: "${TB_QUEUE_KAFKA_EDQS_EVENTS_MAX_POLL_RECORDS:512}" + edqs.state: + # Key-value properties for Kafka consumer for edqs.state topic + - key: max.poll.records + # Max poll records for edqs.state topic + value: "${TB_QUEUE_KAFKA_EDQS_STATE_MAX_POLL_RECORDS:512}" other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section, you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms diff --git a/common/edqs/pom.xml b/common/edqs/pom.xml index 2abd1286b8..e5f5b759ca 100644 --- a/common/edqs/pom.xml +++ b/common/edqs/pom.xml @@ -27,7 +27,7 @@ edqs jar - Thingsboard Server EDQS API + ThingsBoard EDQS API https://thingsboard.io @@ -72,10 +72,6 @@ org.springframework.boot spring-boot-starter-web - - org.apache.kafka - kafka-clients - com.github.ben-manes.caffeine caffeine diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/repo/TenantRepo.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/repo/TenantRepo.java index 9003eb4da2..aaf91bc37b 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/repo/TenantRepo.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/repo/TenantRepo.java @@ -84,7 +84,7 @@ import static org.thingsboard.server.edqs.util.RepositoryUtils.resolveEntityType @Slf4j public class TenantRepo { - public static final Comparator> CREATED_TIME_COMPARATOR = Comparator.comparingLong(o -> o.getFields() != null ? o.getFields().getCreatedTime() : 0); // FIXME: fields may be null at first + public static final Comparator> CREATED_TIME_COMPARATOR = Comparator.comparingLong(ed -> ed.getFields().getCreatedTime()); public static final Comparator> CREATED_TIME_AND_ID_COMPARATOR = CREATED_TIME_COMPARATOR .thenComparing(EntityData::getId); public static final Comparator> CREATED_TIME_AND_ID_DESC_COMPARATOR = CREATED_TIME_AND_ID_COMPARATOR.reversed(); @@ -186,7 +186,11 @@ public class TenantRepo { EntityData entityData = getOrCreate(entityType, entityId); processFields(fields); - entityData.setFields(entity.getFields()); + EntityFields oldFields = entityData.getFields(); + entityData.setFields(fields); + if (oldFields == null) { + getEntitySet(entityType).add(entityData); + } UUID newCustomerId = fields.getCustomerId(); UUID oldCustomerId = entityData.getCustomerId(); @@ -291,7 +295,6 @@ public class TenantRepo { return getEntityMap(entityType).computeIfAbsent(entityId, id -> { log.debug("[{}] Adding {} {}", tenantId, entityType, id); EntityData entityData = constructEntityData(entityType, entityId); - getEntitySet(entityType).add(entityData); edqsStatsService.ifPresent(statService -> statService.reportEvent(tenantId, ObjectType.fromEntityType(entityType), EdqsEventType.UPDATED)); return entityData; }); @@ -331,9 +334,6 @@ public class TenantRepo { EdqsDataQuery query = RepositoryUtils.toNewQuery(oldQuery); log.info("[{}][{}] findEntityDataByQuery: {}", tenantId, customerId, query); QueryContext ctx = buildContext(customerId, query.getEntityFilter(), ignorePermissionCheck); - if (ctx == null) { - return PageData.emptyPageData(); - } EntityQueryProcessor queryProcessor = EntityQueryProcessorFactory.create(this, ctx, query); return sortAndConvert(query, queryProcessor.processQuery(), ctx); } @@ -342,9 +342,6 @@ public class TenantRepo { EdqsQuery query = RepositoryUtils.toNewQuery(oldQuery); log.info("[{}][{}] countEntitiesByQuery: {}", tenantId, customerId, query); QueryContext ctx = buildContext(customerId, query.getEntityFilter(), ignorePermissionCheck); - if (ctx == null) { - return 0; - } EntityQueryProcessor queryProcessor = EntityQueryProcessorFactory.create(this, ctx, query); return queryProcessor.count(); } @@ -383,7 +380,7 @@ public class TenantRepo { // IMPLEMENTATION THAT IS BASED ON TIM SORT (For offset + query.getPageSize() > totalSize / 2) // data.sort(comparator); // var result = data.subList(offset, endIndex); - log.debug("EDQ Sorted in {}", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTs)); + log.trace("EDQ Sorted in {}", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTs)); return new PageData<>(toQueryResult(result, query, ctx), totalPages, totalSize, totalSize > requiredSize); } } diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java index 05eebb188c..c9bae14265 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java @@ -169,7 +169,6 @@ public class KafkaEdqsStateService implements EdqsStateService { ready = true; // once true - always true, not to change readiness status on each repartitioning } } - log.error("ready: {}", ready); return ready != null && ready; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/edqs/EdqsQueue.java b/common/queue/src/main/java/org/thingsboard/server/queue/edqs/EdqsQueue.java index c773ea4e93..45126186a3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/edqs/EdqsQueue.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/edqs/EdqsQueue.java @@ -24,8 +24,8 @@ public enum EdqsQueue { STATE("edqs.state", true, true); private final String topic; - private final boolean readFromBeginning; // read from the beginning of the topic, instead of the latest committed offset - private final boolean stopWhenRead; // stop consuming when reached an empty msg pack + private final boolean readFromBeginning; + private final boolean stopWhenRead; EdqsQueue(String topic, boolean readFromBeginning, boolean stopWhenRead) { this.topic = topic; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLock.java b/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLock.java index b69b98dd4f..9e4b681dab 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLock.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLock.java @@ -15,7 +15,7 @@ */ package org.thingsboard.server.queue.environment; -public interface DistributedLock { +public interface DistributedLock { void lock(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLockService.java b/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLockService.java index 61a21cca7c..17b0f84978 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLockService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/environment/DistributedLockService.java @@ -17,6 +17,6 @@ package org.thingsboard.server.queue.environment; public interface DistributedLockService { - DistributedLock getLock(String key); + DistributedLock getLock(String key); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/environment/DummyDistributedLockService.java b/common/queue/src/main/java/org/thingsboard/server/queue/environment/DummyDistributedLockService.java index 03b6d0f2d9..d289c5cb34 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/environment/DummyDistributedLockService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/environment/DummyDistributedLockService.java @@ -27,12 +27,12 @@ import java.util.concurrent.locks.ReentrantLock; public class DummyDistributedLockService implements DistributedLockService { @Override - public DistributedLock getLock(String key) { - return new DummyDistributedLock<>(); + public DistributedLock getLock(String key) { + return new DummyDistributedLock(); } @RequiredArgsConstructor - private static class DummyDistributedLock implements DistributedLock { + private static class DummyDistributedLock implements DistributedLock { private final ReentrantLock lock = new ReentrantLock(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/environment/ZkDistributedLockService.java b/common/queue/src/main/java/org/thingsboard/server/queue/environment/ZkDistributedLockService.java index 07b8b503f9..c0b9b4d8d3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/environment/ZkDistributedLockService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/environment/ZkDistributedLockService.java @@ -33,12 +33,12 @@ public class ZkDistributedLockService implements DistributedLockService { private final ZkDiscoveryService zkDiscoveryService; @Override - public DistributedLock getLock(String key) { - return new ZkDistributedLock<>(key); + public DistributedLock getLock(String key) { + return new ZkDistributedLock(key); } @RequiredArgsConstructor - private class ZkDistributedLock implements DistributedLock { + private class ZkDistributedLock implements DistributedLock { private final InterProcessLock interProcessLock; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java index 9b580c762e..6d4ed1073c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java @@ -53,7 +53,7 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue private final String groupId; private final boolean readFromBeginning; // reset offset to beginning - private final boolean stopWhenRead; // stop consuming when reached an empty msg pack + private final boolean stopWhenRead; // stop consuming when reached end offset remembered on start private int readCount; private Map endOffsets; // needed if stopWhenRead is true diff --git a/common/stats/src/main/java/org/thingsboard/server/common/stats/DummyMessagesStats.java b/common/stats/src/main/java/org/thingsboard/server/common/stats/DummyMessagesStats.java deleted file mode 100644 index 7847860dcd..0000000000 --- a/common/stats/src/main/java/org/thingsboard/server/common/stats/DummyMessagesStats.java +++ /dev/null @@ -1,54 +0,0 @@ -/** - * Copyright © 2016-2024 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.common.stats; - -public class DummyMessagesStats implements MessagesStats { - @Override - public void incrementTotal(int amount) { - - } - - @Override - public void incrementSuccessful(int amount) { - - } - - @Override - public void incrementFailed(int amount) { - - } - - @Override - public int getTotal() { - return 0; - } - - @Override - public int getSuccessful() { - return 0; - } - - @Override - public int getFailed() { - return 0; - } - - @Override - public void reset() { - - } - -} diff --git a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java index 8dc98377d3..fe0e019537 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java +++ b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java @@ -559,10 +559,6 @@ public class JacksonUtil { } } - public static JsonNode getValueByPath(ObjectNode node, String path) { - return node.at("/" + path.replace(".", "/")); - } - @Data public static class JsonNodeProcessingTask { private final String path; diff --git a/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java b/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java index 3c12ae3209..63371d97e4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java @@ -281,7 +281,7 @@ public class CachedAttributesService implements AttributesService { String key = deleted.getValue(); if (scope != null && key != null) { cache.evict(new AttributeCacheKey(scope, entityId, key)); - // FIXME: version is Long.MAX_VALUE because we expect that the entity is deleted and there won't be any attributes after this + // using version as Long.MAX_VALUE because we expect that the entity is deleted and there won't be any attributes after this edqsService.onDelete(tenantId, ObjectType.ATTRIBUTE_KV, new AttributeKv(entityId, scope, key, Long.MAX_VALUE)); } }); diff --git a/dao/src/main/java/org/thingsboard/server/dao/entity/BaseEntityService.java b/dao/src/main/java/org/thingsboard/server/dao/entity/BaseEntityService.java index 1a0f3be1e7..fb8078dd99 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entity/BaseEntityService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entity/BaseEntityService.java @@ -95,7 +95,7 @@ public class BaseEntityService extends AbstractEntityService implements EntitySe validateId(customerId, id -> INCORRECT_CUSTOMER_ID + id); validateEntityCountQuery(query); - if (edqsService.isApiEnabled() && validForEdqs(query)) { // TODO: separate boolean param whether to use in dashboards; but sync to edqs - always + if (edqsService.isApiEnabled() && validForEdqs(query)) { EdqsRequest request = EdqsRequest.builder() .entityCountQuery(query) .build(); @@ -112,7 +112,7 @@ public class BaseEntityService extends AbstractEntityService implements EntitySe validateId(customerId, id -> INCORRECT_CUSTOMER_ID + id); validateEntityDataQuery(query); - if (edqsService.isApiEnabled() && validForEdqs(query)) { // TODO: separate boolean param whether to use in dashboards; but sync to edqs - always + if (edqsService.isApiEnabled() && validForEdqs(query)) { EdqsRequest request = EdqsRequest.builder() .entityDataQuery(query) .build(); @@ -135,7 +135,7 @@ public class BaseEntityService extends AbstractEntityService implements EntitySe return new PageData<>(result, entityDataByQuery.getTotalPages(), entityDataByQuery.getTotalElements(), entityDataByQuery.hasNext()); } - private boolean validForEdqs(EntityCountQuery query) { + private boolean validForEdqs(EntityCountQuery query) { // for compatibility with PE return true; } diff --git a/dao/src/main/java/org/thingsboard/server/dao/entity/EntityDaoRegistry.java b/dao/src/main/java/org/thingsboard/server/dao/entity/EntityDaoRegistry.java index 86887e94ec..46e6a51830 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entity/EntityDaoRegistry.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entity/EntityDaoRegistry.java @@ -29,18 +29,19 @@ import java.util.Map; @SuppressWarnings({"unchecked"}) public class EntityDaoRegistry { - private final Map> entityDaos = new EnumMap<>(EntityType.class); + private final Map> daos = new EnumMap<>(EntityType.class); - private EntityDaoRegistry(List> entityDaos) { - entityDaos.forEach(dao -> { - if (dao.getEntityType() != null) { - this.entityDaos.put(dao.getEntityType(), dao); + private EntityDaoRegistry(List> daos) { + daos.forEach(dao -> { + EntityType entityType = dao.getEntityType(); + if (entityType != null) { + this.daos.put(entityType, dao); } }); } public Dao getDao(EntityType entityType) { - Dao dao = (Dao) entityDaos.get(entityType); + Dao dao = (Dao) daos.get(entityType); if (dao == null) { throw new IllegalArgumentException("Missing dao for entity type " + entityType); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java index f762be0429..d404c3128c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java @@ -281,8 +281,6 @@ public class ModelConstants { public static final String ALARM_COMMENT_TYPE = "type"; public static final String ALARM_COMMENT_COMMENT = "comment"; - public static final String ALARM_TYPES_TABLE_NAME = "alarm_types"; - /** * Entity relation constants. */ diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/AlarmTypeCompositeKey.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/AlarmTypeCompositeKey.java deleted file mode 100644 index 60f5232142..0000000000 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/AlarmTypeCompositeKey.java +++ /dev/null @@ -1,33 +0,0 @@ -/** - * Copyright © 2016-2024 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.dao.model.sql; - -import lombok.AllArgsConstructor; -import lombok.Data; -import lombok.NoArgsConstructor; - -import java.io.Serializable; -import java.util.UUID; - -@Data -@NoArgsConstructor -@AllArgsConstructor -public class AlarmTypeCompositeKey implements Serializable { - - private UUID tenantId; - private String type; - -} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/JpaRuleNodeDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/JpaRuleNodeDao.java index 43087b8652..3bf9ab2963 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/JpaRuleNodeDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/JpaRuleNodeDao.java @@ -107,11 +107,6 @@ public class JpaRuleNodeDao extends JpaAbstractDao imp ruleNodeRepository.deleteAllById(ruleNodeIds.stream().map(RuleNodeId::getId).collect(Collectors.toList())); } - @Override - public PageData findAllByTenantId(TenantId tenantId, PageLink pageLink) { - return DaoUtil.toPageData(ruleNodeRepository.findByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink))); - } - @Override public EntityType getEntityType() { return EntityType.RULE_NODE; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java index bf19d61a9d..264ea88326 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java @@ -63,7 +63,4 @@ public interface RuleNodeRepository extends JpaRepository @Query("DELETE FROM RuleNodeEntity e where e.id in :ids") void deleteByIdIn(@Param("ids") List ids); - @Query("SELECT n FROM RuleNodeEntity n WHERE n.ruleChainId IN (SELECT rc.id FROM RuleChainEntity rc WHERE rc.tenantId = :tenantId)") - Page findByTenantId(@Param("tenantId") UUID tenantId, Pageable pageable); - } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserAuthSettingsDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserAuthSettingsDao.java index 32b6acf08b..307cadce91 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserAuthSettingsDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserAuthSettingsDao.java @@ -18,12 +18,8 @@ package org.thingsboard.server.dao.sql.user; import lombok.RequiredArgsConstructor; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; -import org.thingsboard.server.common.data.page.PageData; -import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.security.UserAuthSettings; -import org.thingsboard.server.common.data.security.model.mfa.account.AccountTwoFaSettings; import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.TenantEntityDao; import org.thingsboard.server.dao.model.sql.UserAuthSettingsEntity; @@ -50,18 +46,6 @@ public class JpaUserAuthSettingsDao extends JpaAbstractDao findAllByTenantId(TenantId tenantId, PageLink pageLink) { - PageData data = DaoUtil.toPageData(repository.findByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink))); - data.getData().forEach(settings -> { - AccountTwoFaSettings twoFaSettings = settings.getTwoFaSettings(); - if (twoFaSettings != null && twoFaSettings.getConfigs() != null) { - twoFaSettings.getConfigs().values().forEach(config -> config.setSerializeHiddenFields(true)); - } - }); - return data; - } - @Override protected Class getEntityClass() { return UserAuthSettingsEntity.class; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/widget/WidgetsBundleWidgetRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/widget/WidgetsBundleWidgetRepository.java index 3a024e1687..89533106f3 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/widget/WidgetsBundleWidgetRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/widget/WidgetsBundleWidgetRepository.java @@ -15,11 +15,7 @@ */ package org.thingsboard.server.dao.sql.widget; -import org.springframework.data.domain.Page; -import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; -import org.springframework.data.jpa.repository.Query; -import org.springframework.data.repository.query.Param; import org.thingsboard.server.dao.model.sql.WidgetsBundleWidgetCompositeKey; import org.thingsboard.server.dao.model.sql.WidgetsBundleWidgetEntity; @@ -30,7 +26,4 @@ public interface WidgetsBundleWidgetRepository extends JpaRepository findAllByWidgetsBundleId(UUID widgetsBundleId); - @Query("SELECT w FROM WidgetsBundleWidgetEntity w WHERE w.widgetsBundleId IN (SELECT b.id FROM WidgetsBundleEntity b WHERE b.tenantId = :tenantId)") - Page findByTenantId(@Param("tenantId") UUID tenantId, Pageable pageable); - } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/TsKvLatestRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/TsKvLatestRepository.java index 2c97ba30ba..8b44e2c9ad 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/TsKvLatestRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/TsKvLatestRepository.java @@ -15,12 +15,9 @@ */ package org.thingsboard.server.dao.sqlts.latest; -import org.springframework.data.domain.Page; -import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; -import org.thingsboard.server.dao.model.sql.AttributeKvEntity; import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestCompositeKey; import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity; @@ -44,10 +41,6 @@ public interface TsKvLatestRepository extends JpaRepository findAllKeysByEntityIds(@Param("entityIds") List entityIds); - @Query("SELECT e FROM TsKvLatestEntity e ORDER BY e.entityId ASC, e.key ASC") - Page findAll(Pageable pageable); - - @Query(value = "SELECT entity_id, key, ts, bool_v, str_v, long_v, dbl_v, json_v, version FROM ts_kv_latest WHERE (entity_id, key) > " + "(:entityId, :key) ORDER BY entity_id, key LIMIT :batchSize", nativeQuery = true) List findNextBatch(@Param("entityId") UUID entityId, diff --git a/edqs/pom.xml b/edqs/pom.xml index eb824cb77f..8eb6edfb23 100644 --- a/edqs/pom.xml +++ b/edqs/pom.xml @@ -28,8 +28,6 @@ ThingsBoard Entity Data Query Service Application https://thingsboard.io - ThingsBoard Professional Edition: IoT Platform - Device management, data collection, processing and visualization - UTF-8 @@ -90,16 +88,6 @@ exe provided - - org.thingsboard - tools - test - - - org.springframework.security - spring-security-test - test - org.springframework.boot spring-boot-starter-test @@ -115,42 +103,6 @@ awaitility test - - org.dbunit - dbunit - test - - - com.github.springtestdbunit - spring-test-dbunit - test - - - org.testcontainers - cassandra - test - - - org.testcontainers - postgresql - test - - - org.testcontainers - jdbc - test - - - org.java-websocket - Java-WebSocket - test - - - org.eclipse.milo - sdk-server - ${milo.version} - test - org.assertj assertj-core diff --git a/edqs/src/main/resources/edqs.yml b/edqs/src/main/resources/edqs.yml index 3d30c5c6fa..80771c2265 100644 --- a/edqs/src/main/resources/edqs.yml +++ b/edqs/src/main/resources/edqs.yml @@ -132,29 +132,17 @@ queue: # Key-value properties for Kafka consumer per specific topic, e.g. tb_ota_package is a topic name for ota, tb_rule_engine.sq is a topic name for default SequentialByOriginator queue. # Check TB_QUEUE_CORE_OTA_TOPIC and TB_QUEUE_RE_SQ_TOPIC params consumer-properties-per-topic: - tb_ota_package: - # Key-value properties for Kafka consumer per specific topic, e.g. tb_ota_package is a topic name for ota, tb_rule_engine.sq is a topic name for default SequentialByOriginator queue. Check TB_QUEUE_CORE_OTA_TOPIC and TB_QUEUE_RE_SQ_TOPIC params + edqs.events: + # Key-value properties for Kafka consumer for edqs.events topic - key: max.poll.records - # Example of specific consumer properties value per topic - value: "${TB_QUEUE_KAFKA_OTA_MAX_POLL_RECORDS:10}" - tb_version_control: - # Example of specific consumer properties value per topic for VC - - key: max.poll.interval.ms - # Example of specific consumer properties value per topic for VC - value: "${TB_QUEUE_KAFKA_VC_MAX_POLL_INTERVAL_MS:600000}" - # tb_rule_engine.sq: - # - key: max.poll.records - # value: "${TB_QUEUE_KAFKA_SQ_MAX_POLL_RECORDS:1024}" - tb_housekeeper: - # Consumer properties for Housekeeper tasks topic + # Max poll records for edqs.events topic + value: "${TB_QUEUE_KAFKA_EDQS_EVENTS_MAX_POLL_RECORDS:512}" + edqs.state: + # Key-value properties for Kafka consumer for edqs.state topic - key: max.poll.records - # Amount of records to be returned in a single poll. For Housekeeper tasks topic, we should consume messages (tasks) one by one - value: "${TB_QUEUE_KAFKA_HOUSEKEEPER_MAX_POLL_RECORDS:1}" - tb_housekeeper.reprocessing: - # Consumer properties for Housekeeper reprocessing topic - - key: max.poll.records - # Amount of records to be returned in a single poll. For Housekeeper reprocessing topic, we should consume messages (tasks) one by one - value: "${TB_QUEUE_KAFKA_HOUSEKEEPER_REPROCESSING_MAX_POLL_RECORDS:1}" + # Max poll records for edqs.state topic + value: "${TB_QUEUE_KAFKA_EDQS_STATE_MAX_POLL_RECORDS:512}" + other-inline: "${TB_QUEUE_KAFKA_OTHER_PROPERTIES:}" # In this section you can specify custom parameters (semicolon separated) for Kafka consumer/producer/admin # Example "metrics.recording.level:INFO;metrics.sample.window.ms:30000" other: # DEPRECATED. In this section, you can specify custom parameters for Kafka consumer/producer and expose the env variables to configure outside # - key: "request.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_request.timeout.ms @@ -162,26 +150,6 @@ queue: # - key: "session.timeout.ms" # refer to https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_session.timeout.ms # value: "${TB_QUEUE_KAFKA_SESSION_TIMEOUT_MS:10000}" # (10 seconds) topic-properties: - # Kafka properties for Rule Engine - rule-engine: "${TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" - # Kafka properties for Core topics - core: "${TB_QUEUE_KAFKA_CORE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" - # Kafka properties for Transport Api topics - transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" - # Kafka properties for Notifications topics - notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" - # Kafka properties for JS Executor topics - js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:104857600;partitions:100;min.insync.replicas:1}" - # Kafka properties for OTA updates topic - ota-updates: "${TB_QUEUE_KAFKA_OTA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" - # Kafka properties for Version Control topic - version-control: "${TB_QUEUE_KAFKA_VC_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" - # Kafka properties for Integration Api topics - integration-api: "${TB_QUEUE_KAFKA_INTEGRATION_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" - # Kafka properties for Housekeeper tasks topic - housekeeper: "${TB_QUEUE_KAFKA_HOUSEKEEPER_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" - # Kafka properties for Housekeeper reprocessing topic; retention.ms is set to 90 days; partitions is set to 1 since only one reprocessing service is running at a time - housekeeper-reprocessing: "${TB_QUEUE_KAFKA_HOUSEKEEPER_REPROCESSING_TOPIC_PROPERTIES:retention.ms:7776000000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" # Kafka properties for EDQS events topics. Partitions number must be the same as queue.edqs.partitions edqs-events: "${TB_QUEUE_KAFKA_EDQS_EVENTS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:-1;partitions:12;min.insync.replicas:1}" # Kafka properties for EDQS requests topic (default: 3 minutes retention). Partitions number must be the same as queue.edqs.partitions @@ -197,23 +165,6 @@ queue: kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 - transport_api: - # Topic used to consume api requests from transport microservices - requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb_transport.api.requests}" - # Topic used to produce api responses to transport microservices - responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb_transport.api.responses}" - # Maximum pending api requests from transport microservices to be handled by server - max_pending_requests: "${TB_QUEUE_TRANSPORT_MAX_PENDING_REQUESTS:10000}" - # Maximum timeout in milliseconds to handle api request from transport microservice by server - max_requests_timeout: "${TB_QUEUE_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" - # Amount of threads used to invoke callbacks - max_callback_threads: "${TB_QUEUE_TRANSPORT_MAX_CALLBACK_THREADS:100}" - # Amount of threads used for transport API requests - max_core_handler_threads: "${TB_QUEUE_TRANSPORT_MAX_CORE_HANDLER_THREADS:16}" - # Interval in milliseconds to poll api requests from transport microservices - request_poll_interval: "${TB_QUEUE_TRANSPORT_REQUEST_POLL_INTERVAL_MS:25}" - # Interval in milliseconds to poll api response from transport microservices - response_poll_interval: "${TB_QUEUE_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" # General service parameters service: