From 46a58ca82bb17c4434de39c52e34203dfd8fd417 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 3 Jun 2025 12:47:55 +0300 Subject: [PATCH 1/7] Edqs - VersionStore - Use local cache instead of caffeine to reduce memory heap size --- .../server/edqs/processor/EdqsProcessor.java | 1 + .../server/edqs/util/VersionsStore.java | 47 +++++++++++++++---- 2 files changed, 39 insertions(+), 9 deletions(-) diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java index 510d2c3a41..0e74cb98fa 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java @@ -277,6 +277,7 @@ public class EdqsProcessor implements TbQueueHandler, eventConsumer.awaitStop(); responseTemplate.stop(); stateService.stop(); + versionsStore.shutdown(); } } diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java index ba3263eec2..9d4c67c4c2 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java @@ -15,31 +15,35 @@ */ package org.thingsboard.server.edqs.util; -import com.github.benmanes.caffeine.cache.Cache; -import com.github.benmanes.caffeine.cache.Caffeine; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.edqs.EdqsObjectKey; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @Slf4j public class VersionsStore { - private final Cache versions; + private final ConcurrentMap> versions = new ConcurrentHashMap<>(); + private final long expirationMillis; + private final ScheduledExecutorService cleaner = Executors.newSingleThreadScheduledExecutor(); public VersionsStore(int ttlMinutes) { - this.versions = Caffeine.newBuilder() - .expireAfterWrite(ttlMinutes, TimeUnit.MINUTES) - .build(); + this.expirationMillis = TimeUnit.MINUTES.toMillis(ttlMinutes); + startCleanupTask(); } public boolean isNew(EdqsObjectKey key, Long version) { AtomicBoolean isNew = new AtomicBoolean(false); - versions.asMap().compute(key, (k, prevVersion) -> { - if (prevVersion == null || prevVersion <= version) { + versions.compute(key, (k, prevVersion) -> { + if (prevVersion == null || prevVersion.value <= version) { isNew.set(true); - return version; + return new TimedValue<>(version); } else { log.debug("[{}] Version {} is outdated, the latest is {}", key, version, prevVersion); return prevVersion; @@ -48,4 +52,29 @@ public class VersionsStore { return isNew.get(); } + private void startCleanupTask() { + cleaner.scheduleAtFixedRate(() -> { + long now = System.currentTimeMillis(); + for (Map.Entry> entry : versions.entrySet()) { + if (now - entry.getValue().lastUpdated > expirationMillis) { + versions.remove(entry.getKey(), entry.getValue()); + } + } + }, expirationMillis, expirationMillis, TimeUnit.MILLISECONDS); + } + + public void shutdown() { + cleaner.shutdown(); + } + + private static class TimedValue { + private final long lastUpdated; + private final V value; + + public TimedValue(V value) { + this.value = value; + this.lastUpdated = System.currentTimeMillis(); + } + } + } From ccdcbc635043bfb05628612d0179e03cfd9bfb18 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 3 Jun 2025 15:47:19 +0300 Subject: [PATCH 2/7] VersionsStore - use long intead of Long to decrease heap size --- .../thingsboard/server/edqs/util/VersionsStore.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java index 9d4c67c4c2..c8c8f76761 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java @@ -29,7 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean; @Slf4j public class VersionsStore { - private final ConcurrentMap> versions = new ConcurrentHashMap<>(); + private final ConcurrentMap versions = new ConcurrentHashMap<>(); private final long expirationMillis; private final ScheduledExecutorService cleaner = Executors.newSingleThreadScheduledExecutor(); @@ -43,7 +43,7 @@ public class VersionsStore { versions.compute(key, (k, prevVersion) -> { if (prevVersion == null || prevVersion.value <= version) { isNew.set(true); - return new TimedValue<>(version); + return new TimedValue(version); } else { log.debug("[{}] Version {} is outdated, the latest is {}", key, version, prevVersion); return prevVersion; @@ -55,7 +55,7 @@ public class VersionsStore { private void startCleanupTask() { cleaner.scheduleAtFixedRate(() -> { long now = System.currentTimeMillis(); - for (Map.Entry> entry : versions.entrySet()) { + for (Map.Entry entry : versions.entrySet()) { if (now - entry.getValue().lastUpdated > expirationMillis) { versions.remove(entry.getKey(), entry.getValue()); } @@ -67,11 +67,11 @@ public class VersionsStore { cleaner.shutdown(); } - private static class TimedValue { + private static class TimedValue { private final long lastUpdated; - private final V value; + private final long value; - public TimedValue(V value) { + public TimedValue(long value) { this.value = value; this.lastUpdated = System.currentTimeMillis(); } From 1d5c4ac7ab5f978fb05339cc6e897d40c2e8fbc1 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 3 Jun 2025 15:48:26 +0300 Subject: [PATCH 3/7] VersionsStore - added try/catch for cleanup task --- .../thingsboard/server/edqs/util/VersionsStore.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java index c8c8f76761..f348e9cf9e 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/util/VersionsStore.java @@ -54,11 +54,15 @@ public class VersionsStore { private void startCleanupTask() { cleaner.scheduleAtFixedRate(() -> { - long now = System.currentTimeMillis(); - for (Map.Entry entry : versions.entrySet()) { - if (now - entry.getValue().lastUpdated > expirationMillis) { - versions.remove(entry.getKey(), entry.getValue()); + try { + long now = System.currentTimeMillis(); + for (Map.Entry entry : versions.entrySet()) { + if (now - entry.getValue().lastUpdated > expirationMillis) { + versions.remove(entry.getKey(), entry.getValue()); + } } + } catch (Exception e) { + log.error("Cleanup task failed", e); } }, expirationMillis, expirationMillis, TimeUnit.MILLISECONDS); } From 36a2b3f66624702adb12df9b5911e6536043090f Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Wed, 4 Jun 2025 13:13:29 +0300 Subject: [PATCH 4/7] KafkaEdqsStateService - added versionsStore.shutdown() --- .../org/thingsboard/server/edqs/state/KafkaEdqsStateService.java | 1 + 1 file changed, 1 insertion(+) 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 66bbb7a68a..7e2e99e662 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 @@ -224,6 +224,7 @@ public class KafkaEdqsStateService implements EdqsStateService { stateConsumer.awaitStop(); eventsToBackupConsumer.stop(); stateProducer.stop(); + versionsStore.shutdown(); } } From cf4ab4fd09ba68535bacb39dd0ba8b7e612f3009 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 4 Jun 2025 13:45:02 +0300 Subject: [PATCH 5/7] added tests for toTbelCfArg method --- .../state/SingleValueArgumentEntryTest.java | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java index 2c48ed9167..5d035efb26 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java @@ -17,8 +17,15 @@ package org.thingsboard.server.service.cf.ctx.state; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.thingsboard.script.api.tbel.TbelCfArg; +import org.thingsboard.script.api.tbel.TbelCfSingleValueArg; +import org.thingsboard.server.common.data.kv.JsonDataEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -73,4 +80,34 @@ public class SingleValueArgumentEntryTest { void testUpdateEntryWhenValueWasNotChanged() { assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 18, new LongDataEntry("key", 11L), 364L))).isTrue(); } + + @Test + void testToTbelCfArgWhenJsonIsObject() { + entry = new SingleValueArgumentEntry(ts, new JsonDataEntry("key", "{\"test\": 10}"), 370L); + TbelCfArg tbelCfArg = entry.toTbelCfArg(); + assertThat(tbelCfArg).isNotNull(); + assertThat(tbelCfArg).isInstanceOf(TbelCfSingleValueArg.class); + + TbelCfSingleValueArg singleValueArg = (TbelCfSingleValueArg) tbelCfArg; + + assertThat(singleValueArg.getValue()).isInstanceOf(Map.class); + Map expectedMap = Map.of("test", 10); + assertThat(singleValueArg.getValue()).isEqualTo(expectedMap); + } + + @Test + void testToTbelCfArgWhenJsonIsArray() { + entry = new SingleValueArgumentEntry(ts, new JsonDataEntry("key", "[{\"test\": 10}, {\"test2\": 20}]"), 371L); + TbelCfArg tbelCfArg = entry.toTbelCfArg(); + assertThat(tbelCfArg).isNotNull(); + assertThat(tbelCfArg).isInstanceOf(TbelCfSingleValueArg.class); + + TbelCfSingleValueArg singleValueArg = (TbelCfSingleValueArg) tbelCfArg; + + assertThat(singleValueArg.getValue()).isInstanceOf(List.class); + List> expectedList = new ArrayList<>(); + expectedList.add(Map.of("test", 10)); + expectedList.add(Map.of("test2", 20)); + assertThat(singleValueArg.getValue()).isEqualTo(expectedList); + } } \ No newline at end of file From 6727a3c9eac98fe6e6a5a66bac50d689ac3cefca Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 4 Jun 2025 13:51:35 +0300 Subject: [PATCH 6/7] added new line to the end of the file --- .../service/cf/ctx/state/SingleValueArgumentEntryTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java index 5d035efb26..50cac8a6fe 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java @@ -110,4 +110,4 @@ public class SingleValueArgumentEntryTest { expectedList.add(Map.of("test2", 20)); assertThat(singleValueArg.getValue()).isEqualTo(expectedList); } -} \ No newline at end of file +} From 16d204632a0710e931cf7cac78e2dffa53dfc759 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 4 Jun 2025 14:23:39 +0300 Subject: [PATCH 7/7] Add backward compatibility for RateLimitsNotificationInfo --- .../org/thingsboard/server/common/data/limit/LimitedApi.java | 1 + .../server/dao/notification/DefaultNotifications.java | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java index ef839247ab..3dc063ccca 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java @@ -43,6 +43,7 @@ public enum LimitedApi { RateLimitUtil.merge( DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantCoreRateLimits, DefaultTenantProfileConfiguration::getCassandraWriteQueryTenantRuleEngineRateLimits), "Monolith telemetry Cassandra write queries", true), + CASSANDRA_QUERIES(null, true), // left for backward compatibility with RateLimitsNotificationInfo EDGE_EVENTS(DefaultTenantProfileConfiguration::getEdgeEventRateLimits, "Edge events", true), EDGE_EVENTS_PER_EDGE(DefaultTenantProfileConfiguration::getEdgeEventRateLimitsPerEdge, "Edge events per edge", false), EDGE_UPLINK_MESSAGES(DefaultTenantProfileConfiguration::getEdgeUplinkMessagesRateLimits, "Edge uplink messages", true), diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java index 9b8b8b255e..efd69a4e61 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java @@ -33,7 +33,6 @@ import org.thingsboard.server.common.data.notification.rule.DefaultNotificationR import org.thingsboard.server.common.data.notification.rule.EscalatedNotificationRuleRecipientsConfig; import org.thingsboard.server.common.data.notification.rule.NotificationRule; import org.thingsboard.server.common.data.notification.rule.NotificationRuleConfig; -import org.thingsboard.server.common.data.notification.rule.trigger.ResourcesShortageTrigger.Resource; import org.thingsboard.server.common.data.notification.rule.trigger.config.AlarmAssignmentNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.AlarmCommentNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.AlarmNotificationRuleTriggerConfig;