From 0bbec75e7550eb427f9f62b67d2001cec14cd8c4 Mon Sep 17 00:00:00 2001 From: Artem Barysh Date: Wed, 28 May 2025 18:16:54 +0300 Subject: [PATCH 1/7] 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 2/7] 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 3/7] 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 4/7] 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 9b09c6542bdd41bbdcb88bd555dec4a32c12e662 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 4 Jun 2025 11:00:36 +0300 Subject: [PATCH 5/7] 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 6/7] 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 7/7] 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 { -}