From 0bbec75e7550eb427f9f62b67d2001cec14cd8c4 Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Wed, 28 May 2025 18:16:54 +0300 Subject: [PATCH 01/16] Fixed channel disconnection --- .../src/main/java/org/thingsboard/mqtt/MqttClientImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java index 543553951d..65c195ea03 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java @@ -462,7 +462,7 @@ final class MqttClientImpl implements MqttClient { MqttMessage message = new MqttMessage(new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0)); ChannelFuture channelFuture = this.sendAndFlushPacket(message); eventLoop.schedule(() -> { - if (!channelFuture.isDone()) { + if (channel.isOpen()) { this.channel.close(); } }, 500, TimeUnit.MILLISECONDS); From e112077cb0da9dbbd74dab20aa076c8997a3794a Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Thu, 29 May 2025 13:52:08 +0300 Subject: [PATCH 02/16] fixed --- .../org/thingsboard/mqtt/MqttClientImpl.java | 15 ++++++++-- .../org/thingsboard/mqtt/MqttClientTest.java | 30 +++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java index 65c195ea03..5e2c5d44cf 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java @@ -456,16 +456,25 @@ final class MqttClientImpl implements MqttClient { @Override public void disconnect() { + if (disconnected) { + return; + } + log.trace("[{}] Disconnecting from server", channel != null ? channel.id() : "UNKNOWN"); - disconnected = true; if (this.channel != null) { MqttMessage message = new MqttMessage(new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0)); - ChannelFuture channelFuture = this.sendAndFlushPacket(message); + + sendAndFlushPacket(message).addListener((ChannelFutureListener) future -> { + future.channel().close(); + disconnected = true; + }); + eventLoop.schedule(() -> { if (channel.isOpen()) { this.channel.close(); + disconnected = true; } - }, 500, TimeUnit.MILLISECONDS); + }, 1, TimeUnit.SECONDS); } } diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java index 1481b354ee..a65c9fb4c7 100644 --- a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java @@ -119,6 +119,36 @@ class MqttClientTest { assertThat(client.isConnected()).isTrue(); } + @Test + void testDisconnectFromBroker() { + // GIVEN + var clientConfig = new MqttClientConfig(); + clientConfig.setOwnerId("Test[ConnectToBroker]"); + clientConfig.setClientId("connect"); + + client = MqttClient.create(clientConfig, null, handlerExecutor); + + // WHEN + Promise connectFuture = client.connect(broker.getHost(), broker.getMqttPort()); + + // THEN + assertThat(connectFuture).isNotNull(); + + Awaitility.await("waiting for client to connect") + .atMost(Duration.ofSeconds(10L)) + .until(connectFuture::isDone); + + assertThat(connectFuture.isSuccess()).isTrue(); + + // WHEN + client.disconnect(); + + // THEN + Awaitility.await("waiting for client to disconnect") + .atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> assertThat(client.isConnected()).isFalse()); + } + @Test void testDisconnectDueToKeepAliveIfNoActivity() { // GIVEN From 9786e0a2f884d77dbdc55a7c4410795f79325d3a Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Mon, 2 Jun 2025 16:21:35 +0300 Subject: [PATCH 03/16] Resolved PR comments --- .../org/thingsboard/mqtt/MqttClientImpl.java | 6 ++++-- .../org/thingsboard/mqtt/MqttClientTest.java | 16 +++------------- 2 files changed, 7 insertions(+), 15 deletions(-) diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java index 5e2c5d44cf..801470284b 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java @@ -96,6 +96,8 @@ final class MqttClientImpl implements MqttClient { private final ListeningExecutor handlerExecutor; + private final static int DISCONNECT_FALLBACK_DELAY_SECS = 1; + /** * Construct the MqttClientImpl with default config */ @@ -468,13 +470,13 @@ final class MqttClientImpl implements MqttClient { future.channel().close(); disconnected = true; }); - eventLoop.schedule(() -> { if (channel.isOpen()) { + log.trace("[{}] Channel still open after {} second; forcing close now", channel.id(), DISCONNECT_FALLBACK_DELAY_SECS); this.channel.close(); disconnected = true; } - }, 1, TimeUnit.SECONDS); + }, DISCONNECT_FALLBACK_DELAY_SECS, TimeUnit.SECONDS); } } diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java index a65c9fb4c7..60e625aa8d 100644 --- a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java @@ -123,22 +123,12 @@ class MqttClientTest { void testDisconnectFromBroker() { // GIVEN var clientConfig = new MqttClientConfig(); - clientConfig.setOwnerId("Test[ConnectToBroker]"); - clientConfig.setClientId("connect"); + clientConfig.setOwnerId("Test[Disconnect]"); + clientConfig.setClientId("disconnect"); client = MqttClient.create(clientConfig, null, handlerExecutor); - // WHEN - Promise connectFuture = client.connect(broker.getHost(), broker.getMqttPort()); - - // THEN - assertThat(connectFuture).isNotNull(); - - Awaitility.await("waiting for client to connect") - .atMost(Duration.ofSeconds(10L)) - .until(connectFuture::isDone); - - assertThat(connectFuture.isSuccess()).isTrue(); + connect(broker.getHost(), broker.getMqttPort()); // WHEN client.disconnect(); From 16a1d65c286a9bdf1b0a76f3d713413a2d4fc20b Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Mon, 2 Jun 2025 18:02:07 +0300 Subject: [PATCH 04/16] Removed test --- .../server/msa/connectivity/MqttClientTest.java | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java index 1b1ed9bf0f..eb30ef8b9c 100644 --- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java @@ -556,22 +556,6 @@ public class MqttClientTest extends AbstractContainerTest { assertThat(provisionResponse.get("status").asText()).isEqualTo("NOT_FOUND"); } - @Test - public void regularDisconnect() throws Exception { - DeviceCredentials deviceCredentials = testRestClient.getDeviceCredentialsByDeviceId(device.getId()); - - MqttMessageListener listener = new MqttMessageListener(); - MqttClient mqttClient = getMqttClient(deviceCredentials, listener, MqttVersion.MQTT_5); - final List returnCodeByteValue = new ArrayList<>(); - MqttClientCallback callbackForDisconnectWithReturnCode = getCallbackWrapperForDisconnectWithReturnCode(returnCodeByteValue); - mqttClient.setCallback(callbackForDisconnectWithReturnCode); - mqttClient.disconnect(); - Thread.sleep(1000); - assertThat(returnCodeByteValue.size()).isEqualTo(1); - MqttReasonCodes.Disconnect returnCode = MqttReasonCodes.Disconnect.valueOf(returnCodeByteValue.get(0)); - assertThat(returnCode).isEqualTo(MqttReasonCodes.Disconnect.NORMAL_DISCONNECT); - } - @Test public void clientSessionTakenOverDisconnect() throws Exception { DeviceCredentials deviceCredentials = testRestClient.getDeviceCredentialsByDeviceId(device.getId()); From 46a58ca82bb17c4434de39c52e34203dfd8fd417 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 3 Jun 2025 12:47:55 +0300 Subject: [PATCH 05/16] 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 06/16] 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 07/16] 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 9b09c6542bdd41bbdcb88bd555dec4a32c12e662 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 4 Jun 2025 11:00:36 +0300 Subject: [PATCH 08/16] fixed error when json passed as argument --- .../cf/ctx/state/SingleValueArgumentEntry.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index bfe9eed24f..bdbda2309e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -16,9 +16,11 @@ package org.thingsboard.server.service.cf.ctx.state; import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.core.type.TypeReference; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelCfSingleValueArg; import org.thingsboard.server.common.data.kv.AttributeKvEntry; @@ -90,7 +92,14 @@ public class SingleValueArgumentEntry implements ArgumentEntry { @Override public TbelCfArg toTbelCfArg() { - return new TbelCfSingleValueArg(ts, kvEntryValue.getValue()); + Object value; + try { + value = JacksonUtil.readValue(kvEntryValue.getValueAsString(), new TypeReference<>() { + }); + } catch (Exception e) { + value = kvEntryValue.getValue(); + } + return new TbelCfSingleValueArg(ts, value); } @Override From 7fb1a4f20a114694fae28c3efaa9a715bd15fb6b Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 4 Jun 2025 12:33:35 +0300 Subject: [PATCH 09/16] added check for jsonDataEntry --- .../cf/ctx/state/SingleValueArgumentEntry.java | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index bdbda2309e..3b858e81b2 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -25,6 +25,7 @@ import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelCfSingleValueArg; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BasicKvEntry; +import org.thingsboard.server.common.data.kv.JsonDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.util.ProtoUtils; @@ -92,12 +93,13 @@ public class SingleValueArgumentEntry implements ArgumentEntry { @Override public TbelCfArg toTbelCfArg() { - Object value; - try { - value = JacksonUtil.readValue(kvEntryValue.getValueAsString(), new TypeReference<>() { - }); - } catch (Exception e) { - value = kvEntryValue.getValue(); + Object value = kvEntryValue.getValue(); + if (kvEntryValue instanceof JsonDataEntry) { + try { + value = JacksonUtil.readValue(kvEntryValue.getValueAsString(), new TypeReference<>() { + }); + } catch (Exception e) { + } } return new TbelCfSingleValueArg(ts, value); } From 49b3081d416ca42f016fe82f13c70ebc26d7d613 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 4 Jun 2025 12:45:17 +0300 Subject: [PATCH 10/16] Proper rate limit exception for Cassandra queries --- .../service/ws/DefaultWebSocketService.java | 4 ++-- .../dao/timeseries/BaseTimeseriesService.java | 8 ++++---- .../util/AbstractBufferedRateExecutor.java | 11 ++++++----- .../dao/util/TenantRateLimitException.java | 19 ------------------- 4 files changed, 12 insertions(+), 30 deletions(-) delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/util/TenantRateLimitException.java diff --git a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java index cbe6663663..283e3baf76 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java @@ -36,6 +36,7 @@ import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TenantProfile; +import org.thingsboard.server.common.data.exception.RateLimitExceededException; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; @@ -52,7 +53,6 @@ import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.dao.timeseries.TimeseriesService; -import org.thingsboard.server.dao.util.TenantRateLimitException; import org.thingsboard.server.exception.UnauthorizedException; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -742,7 +742,7 @@ public class DefaultWebSocketService implements WebSocketService { @Override public void onFailure(Throwable e) { - if (e instanceof TenantRateLimitException || e.getCause() instanceof TenantRateLimitException) { + if (e instanceof RateLimitExceededException || e.getCause() instanceof RateLimitExceededException) { log.trace("[{}] Tenant rate limit detected for subscription: [{}]:{}", sessionRef.getSecurityCtx().getTenantId(), entityId, cmd); } else { log.info(FAILED_TO_FETCH_DATA, e); diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java index 9eefcaae1e..cecf4ab587 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java @@ -205,8 +205,8 @@ public class BaseTimeseriesService implements TimeseriesService { ListenableFuture dpsFuture = saveTs ? Futures.transform(Futures.allAsList(tsFutures), SUM_ALL_INTEGERS, MoreExecutors.directExecutor()) : Futures.immediateFuture(0); ListenableFuture> versionsFuture = saveLatest ? Futures.allAsList(latestFutures) : Futures.immediateFuture(null); return Futures.whenAllComplete(dpsFuture, versionsFuture).call(() -> { - Integer dataPoints = Futures.getUnchecked(dpsFuture); - List versions = Futures.getUnchecked(versionsFuture); + Integer dataPoints = dpsFuture.get(); + List versions = versionsFuture.get(); return TimeseriesSaveResult.of(dataPoints, versions); }, MoreExecutors.directExecutor()); } @@ -298,13 +298,13 @@ public class BaseTimeseriesService implements TimeseriesService { long interval = query.getInterval(); if (interval < 1) { throw new IncorrectParameterException("Invalid TsKvQuery: 'interval' must be greater than 0, but got " + interval + - ". Please check your query parameters and ensure 'endTs' is greater than 'startTs' or increase 'interval'."); + ". Please check your query parameters and ensure 'endTs' is greater than 'startTs' or increase 'interval'."); } long step = Math.max(interval, 1000); long intervalCounts = (query.getEndTs() - query.getStartTs()) / step; if (intervalCounts > maxTsIntervals || intervalCounts < 0) { throw new IncorrectParameterException("Incorrect TsKvQuery. Number of intervals is to high - " + intervalCounts + ". " + - "Please increase 'interval' parameter for your query or reduce the time range of the query."); + "Please increase 'interval' parameter for your query or reduce the time range of the query."); } } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java b/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java index cbcf3e81ec..4d691db31d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java +++ b/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java @@ -32,6 +32,7 @@ import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cache.limits.RateLimitService; +import org.thingsboard.server.common.data.exception.RateLimitExceededException; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.limit.LimitedApi; import org.thingsboard.server.common.msg.queue.ServiceType; @@ -66,7 +67,7 @@ public abstract class AbstractBufferedRateExecutor> queue; private final ExecutorService dispatcherExecutor; private final ExecutorService callbackExecutor; @@ -124,7 +125,7 @@ public abstract class AbstractBufferedRateExecutor 0 - || rateLimitedTenantsCount > 0 - || concurrencyLevel.get() > 0 - || stats.getStatsCounters().stream().anyMatch(counter -> counter.get() > 0) + || rateLimitedTenantsCount > 0 + || concurrencyLevel.get() > 0 + || stats.getStatsCounters().stream().anyMatch(counter -> counter.get() > 0) ) { StringBuilder statsBuilder = new StringBuilder(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/util/TenantRateLimitException.java b/dao/src/main/java/org/thingsboard/server/dao/util/TenantRateLimitException.java deleted file mode 100644 index 3d79af980d..0000000000 --- a/dao/src/main/java/org/thingsboard/server/dao/util/TenantRateLimitException.java +++ /dev/null @@ -1,19 +0,0 @@ -/** - * Copyright © 2016-2025 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.util; - -public class TenantRateLimitException extends Exception { -} From 36a2b3f66624702adb12df9b5911e6536043090f Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Wed, 4 Jun 2025 13:13:29 +0300 Subject: [PATCH 11/16] 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 12/16] 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 13/16] 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 14/16] 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; From 957965b351427d92cd73e5c2df06df36170c91ca Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 4 Jun 2025 15:39:14 +0300 Subject: [PATCH 15/16] Improvements for task processing --- .../server/service/job/JobManagerTest.java | 6 +++--- .../server/common/data/job/task/DummyTaskResult.java | 11 +++++++---- .../server/common/data/job/task/TaskResult.java | 10 ++++++---- .../thingsboard/server/queue/task/TaskProcessor.java | 2 ++ .../thingsboard/server/dao/job/DefaultJobService.java | 10 ++++++++-- 5 files changed, 26 insertions(+), 13 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/service/job/JobManagerTest.java b/application/src/test/java/org/thingsboard/server/service/job/JobManagerTest.java index b23722afd5..8da1be43f1 100644 --- a/application/src/test/java/org/thingsboard/server/service/job/JobManagerTest.java +++ b/application/src/test/java/org/thingsboard/server/service/job/JobManagerTest.java @@ -89,7 +89,7 @@ public class JobManagerTest extends AbstractControllerTest { @Test public void testSubmitJob_allTasksSuccessful() { - int tasksCount = 5; + int tasksCount = 7; JobId jobId = submitJob(DummyJobConfiguration.builder() .successfulTasksCount(tasksCount) .taskProcessingTimeMs(1000) @@ -154,10 +154,10 @@ public class JobManagerTest extends AbstractControllerTest { @Test public void testCancelJob_whileRunning() throws Exception { - int tasksCount = 100; + int tasksCount = 200; JobId jobId = submitJob(DummyJobConfiguration.builder() .successfulTasksCount(tasksCount) - .taskProcessingTimeMs(100) + .taskProcessingTimeMs(50) .build()).getId(); Thread.sleep(500); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/job/task/DummyTaskResult.java b/common/data/src/main/java/org/thingsboard/server/common/data/job/task/DummyTaskResult.java index 1988f13eb0..5b913af3e5 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/job/task/DummyTaskResult.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/job/task/DummyTaskResult.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.common.data.job.task; +import lombok.Builder; import lombok.Data; import lombok.EqualsAndHashCode; import lombok.NoArgsConstructor; @@ -25,22 +26,25 @@ import org.thingsboard.server.common.data.job.JobType; @Data @EqualsAndHashCode(callSuper = true) @NoArgsConstructor -@SuperBuilder @ToString(callSuper = true) public class DummyTaskResult extends TaskResult { private DummyTaskFailure failure; + @Builder + private DummyTaskResult(boolean success, boolean discarded, DummyTaskFailure failure) { + super(success, discarded); + this.failure = failure; + } + public static DummyTaskResult success(DummyTask task) { return DummyTaskResult.builder() - .key(task.getKey()) .success(true) .build(); } public static DummyTaskResult failed(DummyTask task, Throwable error) { return DummyTaskResult.builder() - .key(task.getKey()) .failure(DummyTaskFailure.builder() .error(error.getMessage()) .number(task.getNumber()) @@ -51,7 +55,6 @@ public class DummyTaskResult extends TaskResult { public static DummyTaskResult discarded(DummyTask task) { return DummyTaskResult.builder() - .key(task.getKey()) .discarded(true) .build(); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/job/task/TaskResult.java b/common/data/src/main/java/org/thingsboard/server/common/data/job/task/TaskResult.java index 21303a55fe..da3c8252eb 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/job/task/TaskResult.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/job/task/TaskResult.java @@ -20,16 +20,12 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonSubTypes.Type; import com.fasterxml.jackson.annotation.JsonTypeInfo; -import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; -import lombok.experimental.SuperBuilder; import org.thingsboard.server.common.data.job.JobType; @Data -@AllArgsConstructor @NoArgsConstructor -@SuperBuilder @JsonIgnoreProperties(ignoreUnknown = true) @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "jobType") @JsonSubTypes({ @@ -40,6 +36,12 @@ public abstract class TaskResult { private String key; private boolean success; private boolean discarded; + private long finishTs; + + protected TaskResult(boolean success, boolean discarded) { + this.success = success; + this.discarded = discarded; + } @JsonIgnore public abstract JobType getJobType(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/task/TaskProcessor.java b/common/queue/src/main/java/org/thingsboard/server/queue/task/TaskProcessor.java index 62ca19a05f..33c52859ca 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/task/TaskProcessor.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/task/TaskProcessor.java @@ -232,6 +232,8 @@ public abstract class TaskProcessor, R extends TaskResult> { } private void reportTaskResult(T task, R result) { + result.setKey(task.getKey()); + result.setFinishTs(System.currentTimeMillis()); statsService.reportTaskResult(task.getTenantId(), task.getJobId(), result); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/job/DefaultJobService.java b/dao/src/main/java/org/thingsboard/server/dao/job/DefaultJobService.java index 153e95a404..360aa0063b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/job/DefaultJobService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/job/DefaultJobService.java @@ -69,7 +69,6 @@ public class DefaultJobService extends AbstractEntityService implements JobServi job.setStatus(QUEUED); } else { job.setStatus(PENDING); - job.getResult().setStartTs(System.currentTimeMillis()); } return saveJob(tenantId, job, true, null); } @@ -125,6 +124,7 @@ public class DefaultJobService extends AbstractEntityService implements JobServi } boolean publishEvent = false; + long lastFinishTs = 0; for (TaskResult taskResult : jobStats.getTaskResults()) { if (!taskResult.getKey().equals(job.getConfiguration().getTasksKey())) { log.debug("Ignoring task result {} with outdated key {}", taskResult, job.getConfiguration().getTasksKey()); @@ -140,6 +140,9 @@ public class DefaultJobService extends AbstractEntityService implements JobServi publishEvent = true; } } + if (taskResult.getFinishTs() > lastFinishTs) { + lastFinishTs = taskResult.getFinishTs(); + } } if (job.getStatus() == RUNNING) { @@ -153,7 +156,7 @@ public class DefaultJobService extends AbstractEntityService implements JobServi job.setStatus(COMPLETED); publishEvent = true; } - result.setFinishTs(System.currentTimeMillis()); + result.setFinishTs(lastFinishTs); job.getConfiguration().setToReprocess(null); } } @@ -166,6 +169,9 @@ public class DefaultJobService extends AbstractEntityService implements JobServi if (!Job.SUPPORTED_ENTITY_TYPES.contains(job.getEntityId().getEntityType())) { throw new IllegalArgumentException("Unsupported entity type " + job.getEntityId().getEntityType()); } + if (job.getStatus() == PENDING) { + job.getResult().setStartTs(System.currentTimeMillis()); + } job = jobDao.save(tenantId, job); if (publishEvent) { From 11fc6358b4b5226ca9ec092c7a33d3b47ae36e5e Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Thu, 5 Jun 2025 15:03:19 +0300 Subject: [PATCH 16/16] Fix EdqsState.isApiReady --- .../org/thingsboard/server/common/data/edqs/EdqsState.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/edqs/EdqsState.java b/common/data/src/main/java/org/thingsboard/server/common/data/edqs/EdqsState.java index 3df7fc92fe..1e890da961 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/edqs/EdqsState.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/edqs/EdqsState.java @@ -20,7 +20,8 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; -import org.apache.commons.lang3.BooleanUtils; + +import static org.apache.commons.lang3.BooleanUtils.toBooleanDefaultIfNull; @Getter @NoArgsConstructor @@ -34,14 +35,14 @@ public class EdqsState { private EdqsApiMode apiMode; public boolean updateEdqsReady(boolean ready) { - boolean changed = BooleanUtils.toBooleanDefaultIfNull(this.edqsReady, false) != ready; + boolean changed = toBooleanDefaultIfNull(this.edqsReady, false) != ready; this.edqsReady = ready; return changed; } @JsonIgnore public boolean isApiReady() { - return edqsReady && syncStatus == EdqsSyncStatus.FINISHED; + return toBooleanDefaultIfNull(edqsReady, false) && syncStatus == EdqsSyncStatus.FINISHED; } @JsonIgnore