Cleanup for EDQS

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<ToEdqsMsg>, TbProtoQueueMsg<FromEdqsMsg>> requestTemplate;
     private ExecutorService executor;
-    private DistributedLock<EdqsSyncState> 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);

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

common/edqs/pom.xml

@@ -27,7 +27,7 @@
     <artifactId>edqs</artifactId>
     <packaging>jar</packaging>
 
-    <name>Thingsboard Server EDQS API</name>
+    <name>ThingsBoard EDQS API</name>
     <url>https://thingsboard.io</url>
 
     <properties>
@@ -72,10 +72,6 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-        </dependency>
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>
             <artifactId>caffeine</artifactId>

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<EntityData<?>> CREATED_TIME_COMPARATOR = Comparator.comparingLong(o -> o.getFields() != null ? o.getFields().getCreatedTime() : 0); // FIXME: fields may be null at first
+    public static final Comparator<EntityData<?>> CREATED_TIME_COMPARATOR = Comparator.comparingLong(ed -> ed.getFields().getCreatedTime());
     public static final Comparator<EntityData<?>> CREATED_TIME_AND_ID_COMPARATOR = CREATED_TIME_COMPARATOR
             .thenComparing(EntityData::getId);
     public static final Comparator<EntityData<?>> 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);
         }
     }

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;
     }
 

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;

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<I> {
+public interface DistributedLock {
 
     void lock();
 

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 {
 
-    <I> DistributedLock<I> getLock(String key);
+    DistributedLock getLock(String key);
 
 }

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 <I> DistributedLock<I> getLock(String key) {
-        return new DummyDistributedLock<>();
+    public DistributedLock getLock(String key) {
+        return new DummyDistributedLock();
     }
 
     @RequiredArgsConstructor
-    private static class DummyDistributedLock<I> implements DistributedLock<I> {
+    private static class DummyDistributedLock implements DistributedLock {
 
         private final ReentrantLock lock = new ReentrantLock();
 

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 <I> DistributedLock<I> getLock(String key) {
-        return new ZkDistributedLock<>(key);
+    public DistributedLock getLock(String key) {
+        return new ZkDistributedLock(key);
     }
 
     @RequiredArgsConstructor
-    private class ZkDistributedLock<I> implements DistributedLock<I> {
+    private class ZkDistributedLock implements DistributedLock {
 
         private final InterProcessLock interProcessLock;
 

common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java

@@ -53,7 +53,7 @@ public class TbKafkaConsumerTemplate<T extends TbQueueMsg> 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<Integer, Long> endOffsets; // needed if stopWhenRead is true
 

common/stats/src/main/java/org/thingsboard/server/common/stats/DummyMessagesStats.java

@@ -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() {
-
-    }
-
-}

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;

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));
             }
         });

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;
     }
 

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<EntityType, Dao<?>> entityDaos = new EnumMap<>(EntityType.class);
+    private final Map<EntityType, Dao<?>> daos = new EnumMap<>(EntityType.class);
 
-    private EntityDaoRegistry(List<Dao<?>> entityDaos) {
-        entityDaos.forEach(dao -> {
-            if (dao.getEntityType() != null) {
-                this.entityDaos.put(dao.getEntityType(), dao);
+    private EntityDaoRegistry(List<Dao<?>> daos) {
+        daos.forEach(dao -> {
+            EntityType entityType = dao.getEntityType();
+            if (entityType != null) {
+                this.daos.put(entityType, dao);
             }
         });
     }
 
     public <T> Dao<T> getDao(EntityType entityType) {
-        Dao<T> dao = (Dao<T>) entityDaos.get(entityType);
+        Dao<T> dao = (Dao<T>) daos.get(entityType);
         if (dao == null) {
             throw new IllegalArgumentException("Missing dao for entity type " + entityType);
         }

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.
      */

dao/src/main/java/org/thingsboard/server/dao/model/sql/AlarmTypeCompositeKey.java

@@ -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;
-
-}

dao/src/main/java/org/thingsboard/server/dao/sql/rule/JpaRuleNodeDao.java

@@ -107,11 +107,6 @@ public class JpaRuleNodeDao extends JpaAbstractDao<RuleNodeEntity, RuleNode> imp
         ruleNodeRepository.deleteAllById(ruleNodeIds.stream().map(RuleNodeId::getId).collect(Collectors.toList()));
     }
 
-    @Override
-    public PageData<RuleNode> findAllByTenantId(TenantId tenantId, PageLink pageLink) {
-        return DaoUtil.toPageData(ruleNodeRepository.findByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink)));
-    }
-
     @Override
     public EntityType getEntityType() {
         return EntityType.RULE_NODE;

dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java

@@ -63,7 +63,4 @@ public interface RuleNodeRepository extends JpaRepository<RuleNodeEntity, UUID>
     @Query("DELETE FROM RuleNodeEntity e where e.id in :ids")
     void deleteByIdIn(@Param("ids") List<UUID> ids);
 
-    @Query("SELECT n FROM RuleNodeEntity n WHERE n.ruleChainId IN (SELECT rc.id FROM RuleChainEntity rc WHERE rc.tenantId = :tenantId)")
-    Page<RuleNodeEntity> findByTenantId(@Param("tenantId") UUID tenantId, Pageable pageable);
-
 }

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<UserAuthSettingsEntit
         repository.deleteByUserId(userId.getId());
     }
 
-    @Override
-    public PageData<UserAuthSettings> findAllByTenantId(TenantId tenantId, PageLink pageLink) {
-        PageData<UserAuthSettings> 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<UserAuthSettingsEntity> getEntityClass() {
         return UserAuthSettingsEntity.class;

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<WidgetsBund
 
     List<WidgetsBundleWidgetEntity> findAllByWidgetsBundleId(UUID widgetsBundleId);
 
-    @Query("SELECT w FROM WidgetsBundleWidgetEntity w WHERE w.widgetsBundleId IN (SELECT b.id FROM WidgetsBundleEntity b WHERE b.tenantId = :tenantId)")
-    Page<WidgetsBundleWidgetEntity> findByTenantId(@Param("tenantId") UUID tenantId, Pageable pageable);
-
 }

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<TsKvLatestEntity, Ts
             "WHERE ts_kv_latest.entity_id IN :entityIds ORDER BY key_dictionary.key", nativeQuery = true)
     List<String> findAllKeysByEntityIds(@Param("entityIds") List<UUID> entityIds);
 
-    @Query("SELECT e FROM TsKvLatestEntity e ORDER BY e.entityId ASC, e.key ASC")
-    Page<TsKvLatestEntity> 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<TsKvLatestEntity> findNextBatch(@Param("entityId") UUID entityId,

edqs/pom.xml

@@ -28,8 +28,6 @@
 
     <name>ThingsBoard Entity Data Query Service Application</name>
     <url>https://thingsboard.io</url>
-    <description>ThingsBoard Professional Edition: IoT Platform - Device management, data collection, processing and visualization
-    </description>
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -90,16 +88,6 @@
             <type>exe</type>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>org.thingsboard</groupId>
-            <artifactId>tools</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.security</groupId>
-            <artifactId>spring-security-test</artifactId>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
@@ -115,42 +103,6 @@
             <artifactId>awaitility</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.dbunit</groupId>
-            <artifactId>dbunit</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.github.springtestdbunit</groupId>
-            <artifactId>spring-test-dbunit</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>cassandra</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>postgresql</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.testcontainers</groupId>
-            <artifactId>jdbc</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.java-websocket</groupId>
-            <artifactId>Java-WebSocket</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.eclipse.milo</groupId>
-            <artifactId>sdk-server</artifactId>
-            <version>${milo.version}</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>

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: