tsData) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
tsService.saveTimeseries(TimeseriesSaveRequest.builder()
diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSslTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSslTest.java
new file mode 100644
index 0000000000..04a1b87833
--- /dev/null
+++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSslTest.java
@@ -0,0 +1,272 @@
+/**
+ * Copyright © 2016-2026 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.service.edge.rpc;
+
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
+import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
+import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.bouncycastle.openssl.jcajce.JcaPEMWriter;
+import org.bouncycastle.openssl.jcajce.JcePEMEncryptorBuilder;
+import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+import org.bouncycastle.util.io.pem.PemObject;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.server.controller.AbstractWebTest;
+import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc;
+
+import java.io.ByteArrayInputStream;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.PrivateKey;
+import java.security.Security;
+import java.security.cert.X509Certificate;
+import java.security.spec.ECGenParameterSpec;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Tests for Edge gRPC SSL setup using the production {@link EdgeGrpcService#setupSsl} method.
+ *
+ * Covers:
+ * 1. Separate cert and key PEM inputs
+ * 2. Combined PEM (cert + key in one file)
+ * 3. Encrypted private key with password
+ * 4. Missing key in combined PEM → error
+ *
+ * Each scenario is parameterized across key types: RSA-2048, RSA-4096, EC P-256, EC P-384.
+ */
+class EdgeGrpcSslTest {
+
+ static {
+ if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
+ Security.addProvider(new BouncyCastleProvider());
+ }
+ }
+
+ enum KeyType {
+ RSA_2048("RSA", 2048, null, "SHA256withRSA"),
+ RSA_4096("RSA", 4096, null, "SHA256withRSA"),
+ EC_P256("EC", 256, "secp256r1", "SHA256withECDSA"),
+ EC_P384("EC", 384, "secp384r1", "SHA384withECDSA");
+
+ final String algorithm;
+ final int size;
+ final String curve;
+ final String sigAlg;
+
+ KeyType(String algorithm, int size, String curve, String sigAlg) {
+ this.algorithm = algorithm;
+ this.size = size;
+ this.curve = curve;
+ this.sigAlg = sigAlg;
+ }
+
+ KeyPair generateKeyPair() throws Exception {
+ KeyPairGenerator kpg = KeyPairGenerator.getInstance(algorithm);
+ if (curve != null) {
+ kpg.initialize(new ECGenParameterSpec(curve));
+ } else {
+ kpg.initialize(size);
+ }
+ return kpg.generateKeyPair();
+ }
+ }
+
+ private final List tempFiles = new ArrayList<>();
+ private Server server;
+ private ManagedChannel channel;
+
+ @AfterEach
+ void cleanup() throws Exception {
+ if (channel != null) {
+ channel.shutdownNow().awaitTermination(2, TimeUnit.SECONDS);
+ }
+ if (server != null) {
+ server.shutdownNow().awaitTermination(2, TimeUnit.SECONDS);
+ }
+ for (Path p : tempFiles) {
+ Files.deleteIfExists(p);
+ }
+ }
+
+ @ParameterizedTest(name = "separateCertAndKey_{0}")
+ @EnumSource(KeyType.class)
+ void separateCertAndKey(KeyType keyType) throws Exception {
+ KeyPair kp = keyType.generateKeyPair();
+ X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg);
+
+ Path certFile = writeTempPem("cert", cert);
+ Path keyFile = writeTempPem("key", kp.getPrivate());
+
+ server = startServer(certFile.toString(), keyFile.toString(), null);
+ assertTlsConnectivity(cert);
+ }
+
+ @ParameterizedTest(name = "combinedPemWithCertAndKey_{0}")
+ @EnumSource(KeyType.class)
+ void combinedPemWithCertAndKey(KeyType keyType) throws Exception {
+ KeyPair kp = keyType.generateKeyPair();
+ X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg);
+
+ Path combinedFile = writeTempPem("combined", cert, kp.getPrivate());
+
+ server = startServer(combinedFile.toString(), "", null);
+ assertTlsConnectivity(cert);
+ }
+
+ // RSA-only: BouncyCastle writes encrypted EC keys in traditional PEM format (BEGIN EC PRIVATE KEY),
+ // which after decryption produces a PEMKeyPair without public key info — causing PemSslCredentials
+ // to fail with "Cannot invoke SubjectPublicKeyInfo.getEncoded() because getPublicKeyInfo() is null".
+ @ParameterizedTest(name = "encryptedPrivateKey_{0}")
+ @EnumSource(value = KeyType.class, names = {"RSA_2048", "RSA_4096"})
+ void encryptedPrivateKey(KeyType keyType) throws Exception {
+ KeyPair kp = keyType.generateKeyPair();
+ X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg);
+ String password = "test-password";
+
+ Path combinedFile = writeTempPemEncrypted("enc-combined", password, cert, kp.getPrivate());
+
+ server = startServer(combinedFile.toString(), "", password);
+ assertTlsConnectivity(cert);
+ }
+
+ @ParameterizedTest(name = "combinedPemWithCertOnly_throwsException_{0}")
+ @EnumSource(KeyType.class)
+ void combinedPemWithCertOnly_throwsException(KeyType keyType) throws Exception {
+ KeyPair kp = keyType.generateKeyPair();
+ X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg);
+
+ Path certOnlyFile = writeTempPem("cert-only", cert);
+
+ assertThatThrownBy(() -> startServer(certOnlyFile.toString(), "", null))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ // --- Server startup using production EdgeGrpcService.setupSsl() ---
+
+ private Server startServer(String certFileResource, String privateKeyResource, String keyPassword) throws Exception {
+ EdgeGrpcService edgeGrpcService = new EdgeGrpcService();
+ ReflectionTestUtils.setField(edgeGrpcService, "certFileResource", certFileResource);
+ ReflectionTestUtils.setField(edgeGrpcService, "privateKeyResource", privateKeyResource);
+ ReflectionTestUtils.setField(edgeGrpcService, "keyPassword", keyPassword != null ? keyPassword : "");
+
+ NettyServerBuilder builder = NettyServerBuilder.forPort(0)
+ .addService(new EdgeRpcServiceGrpc.EdgeRpcServiceImplBase() {});
+
+ edgeGrpcService.setupSsl(builder);
+
+ return builder.build().start();
+ }
+
+ private void assertTlsConnectivity(X509Certificate trustedCert) throws Exception {
+ String certPem = toPem(trustedCert);
+ var clientSsl = GrpcSslContexts.forClient()
+ .trustManager(new ByteArrayInputStream(certPem.getBytes(StandardCharsets.UTF_8)))
+ .build();
+
+ channel = NettyChannelBuilder.forAddress("localhost", server.getPort())
+ .sslContext(clientSsl)
+ .build();
+
+ channel.getState(true); // trigger connection attempt
+ await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(50, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ var state = channel.getState(false);
+ if (state == io.grpc.ConnectivityState.TRANSIENT_FAILURE) {
+ throw new AssertionError("TLS handshake failed: channel in TRANSIENT_FAILURE");
+ }
+ assertThat(state).isEqualTo(io.grpc.ConnectivityState.READY);
+ });
+ }
+
+ // --- Cert/key generation ---
+
+ private X509Certificate generateSelfSignedCert(KeyPair kp, String sigAlg) throws Exception {
+ X500Name subject = new X500Name("CN=localhost");
+ Date now = new Date();
+ return new JcaX509CertificateConverter().getCertificate(
+ new JcaX509v3CertificateBuilder(
+ subject, BigInteger.ONE, now,
+ new Date(now.getTime() + TimeUnit.DAYS.toMillis(1)),
+ subject, kp.getPublic())
+ .build(new JcaContentSignerBuilder(sigAlg).build(kp.getPrivate())));
+ }
+
+ // --- PEM file helpers ---
+
+ private String toPem(Object obj) throws Exception {
+ java.io.StringWriter sw = new java.io.StringWriter();
+ try (JcaPEMWriter w = new JcaPEMWriter(sw)) {
+ w.writeObject(obj);
+ }
+ return sw.toString();
+ }
+
+ private Path writeTempPem(String prefix, Object... objects) throws Exception {
+ Path p = Files.createTempFile(prefix + "-", ".pem");
+ tempFiles.add(p);
+ try (JcaPEMWriter w = new JcaPEMWriter(Files.newBufferedWriter(p))) {
+ for (Object o : objects) {
+ w.writeObject(toPkcs8IfKey(o));
+ }
+ }
+ return p;
+ }
+
+ private Path writeTempPemEncrypted(String prefix, String password, Object... objects) throws Exception {
+ Path p = Files.createTempFile(prefix + "-", ".pem");
+ tempFiles.add(p);
+ var encryptor = new JcePEMEncryptorBuilder("AES-256-CBC")
+ .setProvider(BouncyCastleProvider.PROVIDER_NAME)
+ .build(password.toCharArray());
+ try (JcaPEMWriter w = new JcaPEMWriter(Files.newBufferedWriter(p))) {
+ for (Object o : objects) {
+ if (o instanceof PrivateKey) {
+ w.writeObject(o, encryptor);
+ } else {
+ w.writeObject(o);
+ }
+ }
+ }
+ return p;
+ }
+
+ private Object toPkcs8IfKey(Object o) {
+ if (o instanceof PrivateKey pk) {
+ return new PemObject("PRIVATE KEY", pk.getEncoded());
+ }
+ return o;
+ }
+}
diff --git a/application/src/test/java/org/thingsboard/server/service/entitiy/EntityServiceTest.java b/application/src/test/java/org/thingsboard/server/service/entitiy/EntityServiceTest.java
index c54ffe4a07..05ab1165f6 100644
--- a/application/src/test/java/org/thingsboard/server/service/entitiy/EntityServiceTest.java
+++ b/application/src/test/java/org/thingsboard/server/service/entitiy/EntityServiceTest.java
@@ -115,6 +115,8 @@ import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -1749,13 +1751,13 @@ public class EntityServiceTest extends AbstractControllerTest {
}
@Test
- public void testFindTenantTelemetry() {
+ public void testFindTenantTelemetry() throws ExecutionException, InterruptedException, TimeoutException {
// save timeseries by sys admin
BasicTsKvEntry timeseries = new BasicTsKvEntry(42L, new DoubleDataEntry("temperature", 45.5));
- timeseriesService.save(TenantId.SYS_TENANT_ID, tenantId, timeseries);
+ timeseriesService.save(TenantId.SYS_TENANT_ID, tenantId, timeseries).get(TIMEOUT, TimeUnit.SECONDS);
AttributeKvEntry attr = new BaseAttributeKvEntry(new LongDataEntry("attr", 10L), 42L);
- attributesService.save(TenantId.SYS_TENANT_ID, tenantId, SERVER_SCOPE, List.of(attr));
+ attributesService.save(TenantId.SYS_TENANT_ID, tenantId, SERVER_SCOPE, List.of(attr)).get(TIMEOUT, TimeUnit.SECONDS);
SingleEntityFilter singleEntityFilter = new SingleEntityFilter();
singleEntityFilter.setSingleEntity(AliasEntityId.fromEntityId(tenantId));
diff --git a/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java b/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java
index cf37afad5c..0c12767b60 100644
--- a/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java
+++ b/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java
@@ -23,8 +23,8 @@ import org.junit.Test;
import org.mockito.ArgumentMatcher;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.metadata.TbGetAttributesNode;
@@ -127,10 +127,12 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
})
public class HousekeeperServiceTest extends AbstractControllerTest {
- @SpyBean
+ @MockitoSpyBean
private HousekeeperService housekeeperService;
- @SpyBean
+ @MockitoSpyBean
private HousekeeperReprocessingService housekeeperReprocessingService;
+ @MockitoSpyBean
+ private TsHistoryDeletionTaskProcessor tsHistoryDeletionTaskProcessor;
@Autowired
private EventService eventService;
@Autowired
@@ -153,8 +155,6 @@ public class HousekeeperServiceTest extends AbstractControllerTest {
private CustomerService customerService;
@Autowired
private DashboardService dashboardService;
- @SpyBean
- private TsHistoryDeletionTaskProcessor tsHistoryDeletionTaskProcessor;
private TenantId tenantId;
diff --git a/application/src/test/java/org/thingsboard/server/service/ttl/NotificationsCleanUpServiceTest.java b/application/src/test/java/org/thingsboard/server/service/ttl/NotificationsCleanUpServiceTest.java
new file mode 100644
index 0000000000..d74652325a
--- /dev/null
+++ b/application/src/test/java/org/thingsboard/server/service/ttl/NotificationsCleanUpServiceTest.java
@@ -0,0 +1,144 @@
+/**
+ * Copyright © 2016-2026 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.service.ttl;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.dao.notification.NotificationRequestDao;
+import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository;
+import org.thingsboard.server.dao.tenant.TenantService;
+import org.thingsboard.server.queue.discovery.PartitionService;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class NotificationsCleanUpServiceTest {
+
+ @Mock
+ private PartitionService partitionService;
+ @Mock
+ private SqlPartitioningRepository partitioningRepository;
+ @Mock
+ private NotificationRequestDao notificationRequestDao;
+ @Mock
+ private TenantService tenantService;
+
+ private NotificationsCleanUpService cleanUpService;
+
+ private static final int BATCH_SIZE = 3;
+
+ @BeforeEach
+ public void setUp() {
+ cleanUpService = new NotificationsCleanUpService(partitionService, partitioningRepository, notificationRequestDao, tenantService);
+ ReflectionTestUtils.setField(cleanUpService, "ttlInSec", 2592000L);
+ ReflectionTestUtils.setField(cleanUpService, "partitionSizeInHours", 168);
+ ReflectionTestUtils.setField(cleanUpService, "removalBatchSize", BATCH_SIZE);
+ }
+
+ @Test
+ public void testBatchLoopCallsDaoMultipleTimes() {
+ TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build();
+ when(partitionService.resolve(any(), any(), any())).thenReturn(myPartition);
+ when(partitioningRepository.dropPartitionsBefore(anyString(), anyLong(), anyLong()))
+ .thenReturn(System.currentTimeMillis());
+
+ TenantId tenantId = TenantId.fromUUID(UUID.randomUUID());
+ when(tenantService.findTenantsIds(any()))
+ .thenReturn(new PageData<>(List.of(tenantId), 1, 1, false));
+
+ // Sysadmin: returns 3 (full batch), then 1 (partial) -> 2 calls
+ when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(BATCH_SIZE)
+ .thenReturn(1);
+ // Tenant: returns 3, 3, 0 -> 3 calls
+ when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(BATCH_SIZE)
+ .thenReturn(BATCH_SIZE)
+ .thenReturn(0);
+
+ cleanUpService.cleanUp();
+
+ verify(notificationRequestDao, times(2))
+ .removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE));
+ verify(notificationRequestDao, times(3))
+ .removeByTenantIdAndCreatedTimeBeforeBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE));
+ }
+
+ @Test
+ public void testSkipsTenantNotOnMyPartition() {
+ TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build();
+ TopicPartitionInfo notMyPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(false).build();
+ when(partitionService.resolve(any(), eq(TenantId.SYS_TENANT_ID), eq(TenantId.SYS_TENANT_ID)))
+ .thenReturn(myPartition);
+ when(partitioningRepository.dropPartitionsBefore(anyString(), anyLong(), anyLong()))
+ .thenReturn(System.currentTimeMillis());
+
+ // Sysadmin: no records
+ when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(0);
+
+ TenantId myTenant = TenantId.fromUUID(UUID.randomUUID());
+ TenantId otherTenant = TenantId.fromUUID(UUID.randomUUID());
+ when(tenantService.findTenantsIds(any()))
+ .thenReturn(new PageData<>(List.of(myTenant, otherTenant), 2, 1, false));
+ when(partitionService.resolve(any(), eq(myTenant), eq(myTenant))).thenReturn(myPartition);
+ when(partitionService.resolve(any(), eq(otherTenant), eq(otherTenant))).thenReturn(notMyPartition);
+
+ when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(0);
+
+ cleanUpService.cleanUp();
+
+ verify(notificationRequestDao).removeByTenantIdAndCreatedTimeBeforeBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE));
+ verify(notificationRequestDao, never()).removeByTenantIdAndCreatedTimeBeforeBatch(eq(otherTenant), anyLong(), anyInt());
+ }
+
+ @Test
+ public void testNoPartitionsDropped_stillCleansUpRequests() {
+ TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build();
+ when(partitionService.resolve(any(), any(), any())).thenReturn(myPartition);
+ when(partitioningRepository.dropPartitionsBefore(anyString(), anyLong(), anyLong()))
+ .thenReturn(0L);
+
+ when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(0);
+ when(tenantService.findTenantsIds(any()))
+ .thenReturn(new PageData<>(List.of(), 0, 0, false));
+
+ cleanUpService.cleanUp();
+
+ verify(notificationRequestDao).removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE));
+ }
+
+}
diff --git a/application/src/test/java/org/thingsboard/server/service/ttl/rpc/RpcCleanUpServiceTest.java b/application/src/test/java/org/thingsboard/server/service/ttl/rpc/RpcCleanUpServiceTest.java
new file mode 100644
index 0000000000..22cc1be1dd
--- /dev/null
+++ b/application/src/test/java/org/thingsboard/server/service/ttl/rpc/RpcCleanUpServiceTest.java
@@ -0,0 +1,136 @@
+/**
+ * Copyright © 2016-2026 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.service.ttl.rpc;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.server.common.data.TenantProfile;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
+import org.thingsboard.server.common.data.tenant.profile.TenantProfileData;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.dao.rpc.RpcDao;
+import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
+import org.thingsboard.server.dao.tenant.TenantService;
+import org.thingsboard.server.queue.discovery.PartitionService;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class RpcCleanUpServiceTest {
+
+ @Mock
+ private PartitionService partitionService;
+ @Mock
+ private RpcDao rpcDao;
+ @Mock
+ private TenantService tenantService;
+ @Mock
+ private TbTenantProfileCache tenantProfileCache;
+
+ private RpcCleanUpService cleanUpService;
+
+ private static final int BATCH_SIZE = 3;
+
+ @BeforeEach
+ public void setUp() {
+ cleanUpService = new RpcCleanUpService(tenantService, partitionService, tenantProfileCache, rpcDao);
+ ReflectionTestUtils.setField(cleanUpService, "removalBatchSize", BATCH_SIZE);
+ }
+
+ @Test
+ public void testBatchLoopCallsDaoMultipleTimes() {
+ TenantId tenantId = TenantId.fromUUID(UUID.randomUUID());
+ setupTenant(tenantId, 7);
+
+ // Returns 3 (full batch), 3 (full batch), 1 (partial) -> 3 calls
+ when(rpcDao.deleteOutdatedRpcByTenantIdBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(BATCH_SIZE)
+ .thenReturn(BATCH_SIZE)
+ .thenReturn(1);
+
+ cleanUpService.cleanUp();
+
+ verify(rpcDao, times(3)).deleteOutdatedRpcByTenantIdBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE));
+ }
+
+ @Test
+ public void testSkipsTenantNotOnMyPartition() {
+ TenantId myTenant = TenantId.fromUUID(UUID.randomUUID());
+ TenantId otherTenant = TenantId.fromUUID(UUID.randomUUID());
+
+ TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build();
+ TopicPartitionInfo notMyPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(false).build();
+
+ when(tenantService.findTenantsIds(any()))
+ .thenReturn(new PageData<>(List.of(myTenant, otherTenant), 2, 1, false));
+ when(partitionService.resolve(any(), eq(myTenant), eq(myTenant))).thenReturn(myPartition);
+ when(partitionService.resolve(any(), eq(otherTenant), eq(otherTenant))).thenReturn(notMyPartition);
+
+ setupTenantProfile(myTenant, 7);
+ when(rpcDao.deleteOutdatedRpcByTenantIdBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE)))
+ .thenReturn(0);
+
+ cleanUpService.cleanUp();
+
+ verify(rpcDao).deleteOutdatedRpcByTenantIdBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE));
+ verify(rpcDao, never()).deleteOutdatedRpcByTenantIdBatch(eq(otherTenant), anyLong(), anyInt());
+ }
+
+ @Test
+ public void testSkipsTenantWithZeroTtl() {
+ TenantId tenantId = TenantId.fromUUID(UUID.randomUUID());
+ setupTenant(tenantId, 0);
+
+ cleanUpService.cleanUp();
+
+ verify(rpcDao, never()).deleteOutdatedRpcByTenantIdBatch(any(), anyLong(), anyInt());
+ }
+
+ private void setupTenant(TenantId tenantId, int rpcTtlDays) {
+ TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build();
+ when(partitionService.resolve(any(), eq(tenantId), eq(tenantId))).thenReturn(myPartition);
+ when(tenantService.findTenantsIds(any()))
+ .thenReturn(new PageData<>(List.of(tenantId), 1, 1, false));
+ setupTenantProfile(tenantId, rpcTtlDays);
+ }
+
+ private void setupTenantProfile(TenantId tenantId, int rpcTtlDays) {
+ TenantProfile profile = new TenantProfile();
+ TenantProfileData profileData = new TenantProfileData();
+ DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration();
+ config.setRpcTtlDays(rpcTtlDays);
+ profileData.setConfiguration(config);
+ profile.setProfileData(profileData);
+ when(tenantProfileCache.get(tenantId)).thenReturn(profile);
+ }
+
+}
diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java
index 9bed9bc483..35e12691c9 100644
--- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java
+++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java
@@ -171,7 +171,14 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable {
if (this.leshanClient != null) {
log.info("Stop/reboot LwM2M client {}", this.leshanClient.getEndpoint(identity));
- this.leshanClient.stop(false);
+ try {
+ this.leshanClient.stop(false);
+ } catch (Exception stopEx) {
+ // Leshan may throw NPE during CoAP observe-relation cleanup when the server
+ // reference is null (race condition in NotificationDataStore.toKey()).
+ // The client is still considered stopped at this point — proceed with restart.
+ log.warn("Exception during LwM2M client stop, proceeding with restart: {}", stopEx.getMessage());
+ }
log.info("Start after update fw LwM2M client {}", this.leshanClient.getEndpoint(identity));
this.leshanClient.start();
@@ -193,7 +200,7 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable {
} catch (Exception e) {
log.error("Error during firmware update", e);
}
- }, 0, TimeUnit.SECONDS); // start immediately, without further delay
+ }, 1, TimeUnit.SECONDS); // delay 1 sec to allow CoAP Execute response to be delivered before client stops
}
protected void setLeshanClient(LeshanClient leshanClient) {
diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationReadCollectedValueTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationReadCollectedValueTest.java
index 432fac6d1a..c61ca0b123 100644
--- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationReadCollectedValueTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationReadCollectedValueTest.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;
import org.junit.Before;
import org.junit.Test;
+import org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper;
import org.thingsboard.server.transport.lwm2m.rpc.AbstractRpcLwM2MIntegrationTest;
import java.util.concurrent.atomic.AtomicReference;
import static java.util.concurrent.TimeUnit.SECONDS;
@@ -37,6 +38,12 @@ import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID
@Slf4j
public class RpcLwm2mIntegrationReadCollectedValueTest extends AbstractRpcLwM2MIntegrationTest {
+ @Before
+ public void resetCollectedValueTimestamps() {
+ Lwm2mTestHelper.RESOURCE_ID_3303_12_5700_TS_0 = 0;
+ Lwm2mTestHelper.RESOURCE_ID_3303_12_5700_TS_1 = 0;
+ }
+
/**
* Read {"id":"/3303/12/5700"}
* Trigger a Send operation from the client with multiple values for the same resource as a payload
diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java
index a7307b2308..5f43aa7e00 100644
--- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java
@@ -420,6 +420,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
Awaitility.await()
.atMost(10, TimeUnit.SECONDS)
+ .ignoreExceptions()
.until(() -> {
List> attributes = doGetAsyncTyped(attributeValuesUrl, new TypeReference<>() {
});
diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java
index 4e4d0debf9..a9bdbb55db 100644
--- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java
@@ -191,6 +191,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
AtomicReference device = new AtomicReference<>();
await(alias + "find device [" + deviceName + "] after created")
.atMost(200, TimeUnit.SECONDS)
+ .ignoreExceptions()
.until(() -> {
device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class));
return device.get() != null;
@@ -236,6 +237,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
AtomicReference device = new AtomicReference<>();
await(alias + "find device [" + deviceName + "] after created")
.atMost(200, TimeUnit.SECONDS)
+ .ignoreExceptions()
.until(() -> {
device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class));
return device.get() != null;
diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/AbstractMqttV5ClientSparkplugAttributesTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/AbstractMqttV5ClientSparkplugAttributesTest.java
index 756c8e603c..d42adfbcf0 100644
--- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/AbstractMqttV5ClientSparkplugAttributesTest.java
+++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/AbstractMqttV5ClientSparkplugAttributesTest.java
@@ -468,6 +468,7 @@ public abstract class AbstractMqttV5ClientSparkplugAttributesTest extends Abstra
AtomicReference> actualKeys = new AtomicReference<>();
await(alias + SparkplugMessageType.NBIRTH.name())
.atMost(40, TimeUnit.SECONDS)
+ .ignoreExceptions()
.until(() -> {
actualKeys.set(doGetAsyncTyped(urlTemplate, new TypeReference<>() {
}));
@@ -483,6 +484,7 @@ public abstract class AbstractMqttV5ClientSparkplugAttributesTest extends Abstra
AtomicReference> actualKeys = new AtomicReference<>();
await(alias + SparkplugMessageType.DBIRTH.name())
.atMost(40, TimeUnit.SECONDS)
+ .ignoreExceptions()
.until(() -> {
actualKeys.set(doGetAsyncTyped(urlTemplate, new TypeReference<>() {
}));
diff --git a/application/src/test/resources/application-test.properties b/application/src/test/resources/application-test.properties
index e79289340c..7f0ab964d6 100644
--- a/application/src/test/resources/application-test.properties
+++ b/application/src/test/resources/application-test.properties
@@ -44,6 +44,7 @@ queue.transport_api.response_poll_interval=5
queue.transport.poll_interval=5
queue.core.poll-interval=5
queue.core.partitions=2
+queue.core.housekeeper.task-reprocessing-delay-ms=0
queue.rule-engine.poll-interval=5
queue.rule-engine.stats.enabled=true
diff --git a/build.sh b/build.sh
index d2a965f281..0342aa9cbf 100755
--- a/build.sh
+++ b/build.sh
@@ -31,7 +31,7 @@ java -version
#echo "Cleaning ui-ngx/node_modules" && rm -rf ui-ngx/node_modules
MAVEN_OPTS="-Xmx1024m" NODE_OPTIONS="--max_old_space_size=4096" DOCKER_CLI_EXPERIMENTAL=enabled DOCKER_BUILDKIT=0 \
-mvn -T2 license:format clean install -DskipTests \
+mvn -T6 license:format clean install -DskipTests -Dpkg.skip=true \
$PROJECTS --also-make
# \
# -Dpush-docker-amd-arm-images
diff --git a/common/actor/pom.xml b/common/actor/pom.xml
index 649e7bbe29..6179e7cc33 100644
--- a/common/actor/pom.xml
+++ b/common/actor/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/cache/pom.xml b/common/cache/pom.xml
index 3780d356b0..69ae480c4a 100644
--- a/common/cache/pom.xml
+++ b/common/cache/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/cluster-api/pom.xml b/common/cluster-api/pom.xml
index ab762668b4..f7cbe4a047 100644
--- a/common/cluster-api/pom.xml
+++ b/common/cluster-api/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/coap-server/pom.xml b/common/coap-server/pom.xml
index 577ed81dcf..9949da0956 100644
--- a/common/coap-server/pom.xml
+++ b/common/coap-server/pom.xml
@@ -22,7 +22,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/dao-api/pom.xml b/common/dao-api/pom.xml
index 7118760ae8..f89ea630ef 100644
--- a/common/dao-api/pom.xml
+++ b/common/dao-api/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/data/pom.xml b/common/data/pom.xml
index 6a25b2e83c..af0f591bf8 100644
--- a/common/data/pom.xml
+++ b/common/data/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsDeletionHousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsDeletionHousekeeperTask.java
index dea590295e..d66ec046de 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsDeletionHousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsDeletionHousekeeperTask.java
@@ -23,6 +23,7 @@ import lombok.ToString;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
+import java.io.Serial;
import java.util.List;
import java.util.UUID;
@@ -32,6 +33,9 @@ import java.util.UUID;
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class AlarmsDeletionHousekeeperTask extends HousekeeperTask {
+ @Serial
+ private static final long serialVersionUID = 9214680001573764374L;
+
private List alarms;
public AlarmsDeletionHousekeeperTask(TenantId tenantId, EntityId entityId) {
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsUnassignHousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsUnassignHousekeeperTask.java
index 0313190056..445850d387 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsUnassignHousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/AlarmsUnassignHousekeeperTask.java
@@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UserId;
+import java.io.Serial;
import java.util.List;
import java.util.UUID;
@@ -33,6 +34,9 @@ import java.util.UUID;
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class AlarmsUnassignHousekeeperTask extends HousekeeperTask {
+ @Serial
+ private static final long serialVersionUID = 9156667024462937756L;
+
private String userTitle;
private List alarms;
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/EntitiesDeletionHousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/EntitiesDeletionHousekeeperTask.java
index fe25a98a1d..c1a233b1c4 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/EntitiesDeletionHousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/EntitiesDeletionHousekeeperTask.java
@@ -23,6 +23,7 @@ import lombok.ToString;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.TenantId;
+import java.io.Serial;
import java.util.List;
import java.util.UUID;
@@ -32,6 +33,9 @@ import java.util.UUID;
@NoArgsConstructor
public class EntitiesDeletionHousekeeperTask extends HousekeeperTask {
+ @Serial
+ private static final long serialVersionUID = 9009068831061529286L;
+
private EntityType entityType;
private List entities;
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/HousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/HousekeeperTask.java
index 875ef2765f..2df7cf4dd4 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/HousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/HousekeeperTask.java
@@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
+import java.io.Serial;
import java.io.Serializable;
@JsonIgnoreProperties(ignoreUnknown = true)
@@ -45,6 +46,9 @@ import java.io.Serializable;
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class HousekeeperTask implements Serializable {
+ @Serial
+ private static final long serialVersionUID = -2585974110832225152L;
+
private TenantId tenantId;
private EntityId entityId;
private HousekeeperTaskType taskType;
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/LatestTsDeletionHousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/LatestTsDeletionHousekeeperTask.java
index cd3e94e5c6..931c2931cb 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/LatestTsDeletionHousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/LatestTsDeletionHousekeeperTask.java
@@ -23,12 +23,17 @@ import lombok.ToString;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
+import java.io.Serial;
+
@Data
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class LatestTsDeletionHousekeeperTask extends HousekeeperTask {
+ @Serial
+ private static final long serialVersionUID = 5193191938513490138L;
+
private String key;
public LatestTsDeletionHousekeeperTask(TenantId tenantId, EntityId entityId, String key) {
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TenantEntitiesDeletionHousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TenantEntitiesDeletionHousekeeperTask.java
index 443d929d8b..be7ff6f7ec 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TenantEntitiesDeletionHousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TenantEntitiesDeletionHousekeeperTask.java
@@ -23,12 +23,17 @@ import lombok.ToString;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.TenantId;
+import java.io.Serial;
+
@Data
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
@NoArgsConstructor
public class TenantEntitiesDeletionHousekeeperTask extends HousekeeperTask {
+ @Serial
+ private static final long serialVersionUID = -8033108795318393447L;
+
private EntityType entityType;
public TenantEntitiesDeletionHousekeeperTask(TenantId tenantId, EntityType entityType) {
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TsHistoryDeletionHousekeeperTask.java b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TsHistoryDeletionHousekeeperTask.java
index d9315f0ff4..b520899ca4 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TsHistoryDeletionHousekeeperTask.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/housekeeper/TsHistoryDeletionHousekeeperTask.java
@@ -23,12 +23,17 @@ import lombok.ToString;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
+import java.io.Serial;
+
@Data
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class TsHistoryDeletionHousekeeperTask extends HousekeeperTask {
+ @Serial
+ private static final long serialVersionUID = 4573851542705079043L;
+
private String key;
public TsHistoryDeletionHousekeeperTask(TenantId tenantId, EntityId entityId, String key) {
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvEntry.java b/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvEntry.java
index cb4092f433..595e1aa26b 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvEntry.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvEntry.java
@@ -37,4 +37,9 @@ public interface TsKvEntry extends KvEntry, HasVersion {
return new TsValue(getTs(), getValueAsString());
}
+ @JsonIgnore
+ default boolean isDeletedEntry() {
+ return getTs() == 0 && (getValue() == null || getValueAsString().isEmpty());
+ }
+
}
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRuleRecipientsConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRuleRecipientsConfig.java
index fda73b8059..de4e05a2cb 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRuleRecipientsConfig.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRuleRecipientsConfig.java
@@ -17,6 +17,7 @@ package org.thingsboard.server.common.data.notification.rule;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
@@ -29,7 +30,7 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
-@JsonIgnoreProperties
+@JsonIgnoreProperties(ignoreUnknown = true)
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "triggerType", visible = true, include = JsonTypeInfo.As.EXISTING_PROPERTY, defaultImpl = DefaultNotificationRuleRecipientsConfig.class)
@JsonSubTypes({
@Type(name = "ALARM", value = EscalatedNotificationRuleRecipientsConfig.class),
@@ -38,6 +39,7 @@ import java.util.UUID;
public abstract class NotificationRuleRecipientsConfig implements Serializable {
@NotNull
+ @JsonProperty("triggerType")
private NotificationRuleTriggerType triggerType;
@JsonIgnore
diff --git a/common/discovery-api/pom.xml b/common/discovery-api/pom.xml
index 83988d0ef0..ee2a7f4503 100644
--- a/common/discovery-api/pom.xml
+++ b/common/discovery-api/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/edge-api/pom.xml b/common/edge-api/pom.xml
index 45dfec2a2d..de3d622557 100644
--- a/common/edge-api/pom.xml
+++ b/common/edge-api/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto
index 913a9199e7..516d35e41a 100644
--- a/common/edge-api/src/main/proto/edge.proto
+++ b/common/edge-api/src/main/proto/edge.proto
@@ -47,8 +47,10 @@ enum EdgeVersion {
V_4_3_0 = 13;
V_4_2_1_2 = 14;
V_4_2_2 = 4220;
+ V_4_2_2_1 = 4221;
V_4_3_0_1 = 15;
V_4_3_1 = 4310;
+ V_4_3_1_1 = 4311;
V_LATEST = 99999;
}
diff --git a/common/edqs/pom.xml b/common/edqs/pom.xml
index ba1ac9040e..8790b02f70 100644
--- a/common/edqs/pom.xml
+++ b/common/edqs/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/message/pom.xml b/common/message/pom.xml
index 550ac30ffa..ce8b1e5fae 100644
--- a/common/message/pom.xml
+++ b/common/message/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/pom.xml b/common/pom.xml
index d8db7fc527..2c25414bc6 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
thingsboard
common
diff --git a/common/proto/pom.xml b/common/proto/pom.xml
index 4a67798576..00243753d2 100644
--- a/common/proto/pom.xml
+++ b/common/proto/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/queue/pom.xml b/common/queue/pom.xml
index cbfdd997d1..800071501f 100644
--- a/common/queue/pom.xml
+++ b/common/queue/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/script/pom.xml b/common/script/pom.xml
index 5830c024bc..799b493f25 100644
--- a/common/script/pom.xml
+++ b/common/script/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/script/remote-js-client/pom.xml b/common/script/remote-js-client/pom.xml
index aef83dd55a..d88fa8c397 100644
--- a/common/script/remote-js-client/pom.xml
+++ b/common/script/remote-js-client/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
script
org.thingsboard.common.script
diff --git a/common/script/script-api/pom.xml b/common/script/script-api/pom.xml
index fba5527883..7dffb3be3a 100644
--- a/common/script/script-api/pom.xml
+++ b/common/script/script-api/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
script
org.thingsboard.common.script
diff --git a/common/stats/pom.xml b/common/stats/pom.xml
index 701acc4b52..194967bf63 100644
--- a/common/stats/pom.xml
+++ b/common/stats/pom.xml
@@ -22,7 +22,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/transport/coap/pom.xml b/common/transport/coap/pom.xml
index faeb310f88..bf3316b437 100644
--- a/common/transport/coap/pom.xml
+++ b/common/transport/coap/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
transport
org.thingsboard.common.transport
diff --git a/common/transport/http/pom.xml b/common/transport/http/pom.xml
index 74bea2e70f..fe6fae32d1 100644
--- a/common/transport/http/pom.xml
+++ b/common/transport/http/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
transport
org.thingsboard.common.transport
diff --git a/common/transport/lwm2m/pom.xml b/common/transport/lwm2m/pom.xml
index 5a9dbe8a06..d9370c085d 100644
--- a/common/transport/lwm2m/pom.xml
+++ b/common/transport/lwm2m/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
transport
org.thingsboard.common.transport
diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbLwM2mRedisRegistrationStore.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbLwM2mRedisRegistrationStore.java
index 4b2ef07994..8158e82a81 100644
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbLwM2mRedisRegistrationStore.java
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbLwM2mRedisRegistrationStore.java
@@ -323,22 +323,24 @@ public class TbLwM2mRedisRegistrationStore implements RegistrationStore, Startab
@Override
public Iterator getAllRegistrations() {
- try (var connection = connectionFactory.getConnection()) {
+ try (var scanConnection = connectionFactory.getConnection();
+ var getConnection = connectionFactory.getConnection()) {
Collection list = new LinkedList<>();
ScanOptions scanOptions = ScanOptions.scanOptions().count(100).match(REG_EP + "*").build();
List> scans = new ArrayList<>();
- if (connection instanceof RedisClusterConnection) {
- ((RedisClusterConnection) connection).clusterGetNodes().forEach(node -> {
- scans.add(((RedisClusterConnection) connection).scan(node, scanOptions));
- });
+ if (scanConnection instanceof RedisClusterConnection clusterConnection) {
+ clusterConnection.clusterGetNodes().forEach(node ->
+ scans.add(clusterConnection.scan(node, scanOptions)));
} else {
- scans.add(connection.scan(scanOptions));
+ scans.add(scanConnection.scan(scanOptions));
}
scans.forEach(scan -> {
scan.forEachRemaining(key -> {
- byte[] element = connection.get(key);
- list.add(deserializeReg(element));
+ byte[] element = getConnection.get(key);
+ if (element != null) {
+ list.add(deserializeReg(element));
+ }
});
});
return list.iterator();
diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java
index 3293cd8b53..4beefb4896 100644
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java
@@ -61,21 +61,21 @@ public class TbRedisLwM2MClientStore implements TbLwM2MClientStore {
@Override
public Set getAll() {
- try (var connection = connectionFactory.getConnection()) {
+ try (var scanConnection = connectionFactory.getConnection();
+ var getConnection = connectionFactory.getConnection()) {
Set clients = new HashSet<>();
ScanOptions scanOptions = ScanOptions.scanOptions().count(100).match(CLIENT_EP + "*").build();
List> scans = new ArrayList<>();
- if (connection instanceof RedisClusterConnection) {
- ((RedisClusterConnection) connection).clusterGetNodes().forEach(node -> {
- scans.add(((RedisClusterConnection) connection).scan(node, scanOptions));
- });
+ if (scanConnection instanceof RedisClusterConnection clusterConnection) {
+ clusterConnection.clusterGetNodes().forEach(node ->
+ scans.add(clusterConnection.scan(node, scanOptions)));
} else {
- scans.add(connection.scan(scanOptions));
+ scans.add(scanConnection.scan(scanOptions));
}
scans.forEach(scan -> {
scan.forEachRemaining(key -> {
- byte[] element = connection.get(key);
+ byte[] element = getConnection.get(key);
if (element != null) {
try {
clients.add(deserialize(element));
diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MModelConfigStore.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MModelConfigStore.java
index 73b6f3c8df..31a78234d0 100644
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MModelConfigStore.java
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MModelConfigStore.java
@@ -35,22 +35,24 @@ public class TbRedisLwM2MModelConfigStore implements TbLwM2MModelConfigStore {
@Override
public List getAll() {
- try (var connection = connectionFactory.getConnection()) {
+ try (var scanConnection = connectionFactory.getConnection();
+ var getConnection = connectionFactory.getConnection()) {
List configs = new ArrayList<>();
ScanOptions scanOptions = ScanOptions.scanOptions().count(100).match(MODEL_EP + "*").build();
List> scans = new ArrayList<>();
- if (connection instanceof RedisClusterConnection) {
- ((RedisClusterConnection) connection).clusterGetNodes().forEach(node -> {
- scans.add(((RedisClusterConnection) connection).scan(node, scanOptions));
- });
+ if (scanConnection instanceof RedisClusterConnection clusterConnection) {
+ clusterConnection.clusterGetNodes().forEach(node ->
+ scans.add(clusterConnection.scan(node, scanOptions)));
} else {
- scans.add(connection.scan(scanOptions));
+ scans.add(scanConnection.scan(scanOptions));
}
scans.forEach(scan -> {
scan.forEachRemaining(key -> {
- byte[] element = connection.get(key);
- configs.add(JacksonUtil.fromBytes(element, LwM2MModelConfig.class));
+ byte[] element = getConnection.get(key);
+ if (element != null) {
+ configs.add(JacksonUtil.fromBytes(element, LwM2MModelConfig.class));
+ }
});
});
return configs;
diff --git a/common/transport/lwm2m/src/test/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStoreTest.java b/common/transport/lwm2m/src/test/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStoreTest.java
new file mode 100644
index 0000000000..9fe29c1188
--- /dev/null
+++ b/common/transport/lwm2m/src/test/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStoreTest.java
@@ -0,0 +1,137 @@
+/**
+ * Copyright © 2016-2026 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.transport.lwm2m.server.store;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+import org.thingsboard.server.transport.lwm2m.server.client.LwM2MClientState;
+import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
+
+import java.util.List;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.thingsboard.server.transport.lwm2m.server.store.util.LwM2MClientSerDes.serialize;
+
+/**
+ * Verifies that {@link TbRedisLwM2MClientStore#getAll()} uses separate connections for
+ * SCAN and GET operations to prevent Jedis 5.x response-ordering corruption that occurs
+ * when both commands share the same connection.
+ */
+@ExtendWith(MockitoExtension.class)
+class TbRedisLwM2MClientStoreTest {
+
+ @Mock
+ RedisConnectionFactory connectionFactory;
+
+ @Mock
+ RedisConnection scanConnection;
+
+ @Mock
+ RedisConnection getConnection;
+
+ TbRedisLwM2MClientStore store;
+
+ @BeforeEach
+ void setUp() {
+ // First getConnection() call → scanConnection, second → getConnection
+ when(connectionFactory.getConnection())
+ .thenReturn(scanConnection)
+ .thenReturn(getConnection);
+ store = new TbRedisLwM2MClientStore(connectionFactory);
+ }
+
+ @Test
+ void getAll_returnsSingleClient() {
+ LwM2mClient client = new LwM2mClient("nodeId", "testEndpoint");
+ client.setState(LwM2MClientState.REGISTERED);
+ byte[] key = "CLIENT#EP#testEndpoint".getBytes();
+ byte[] value = serialize(client);
+
+ // Cursor created before thenReturn to avoid Mockito unfinished-stubbing error
+ Cursor cursor = cursorOf(key);
+ when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor);
+ when(getConnection.get(key)).thenReturn(value);
+
+ Set result = store.getAll();
+
+ assertThat(result).hasSize(1);
+ assertThat(result.iterator().next().getEndpoint()).isEqualTo("testEndpoint");
+ }
+
+ @Test
+ void getAll_getIsNeverCalledOnScanConnection() {
+ Cursor cursor = cursorOf();
+ when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor);
+
+ store.getAll();
+
+ verify(scanConnection, never()).get(any(byte[].class));
+ }
+
+ @Test
+ void getAll_scanIsNeverCalledOnGetConnection() {
+ Cursor cursor = cursorOf();
+ when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor);
+
+ store.getAll();
+
+ verify(getConnection, never()).scan(any(ScanOptions.class));
+ }
+
+ @Test
+ void getAll_skipsKeyWhenValueIsNull() {
+ byte[] key = "CLIENT#EP#gone".getBytes();
+ Cursor cursor = cursorOf(key);
+ when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor);
+ // getConnection.get(key) returns null by default — no stubbing needed
+
+ Set result = store.getAll();
+
+ assertThat(result).isEmpty();
+ }
+
+ /**
+ * Creates a mock {@link Cursor} that iterates over the given keys via {@code forEachRemaining}.
+ * The cursor is created separately (not inside a {@code thenReturn()} argument) to avoid
+ * Mockito's "unfinished stubbing" error caused by nested {@code when()} calls.
+ */
+ @SuppressWarnings("unchecked")
+ private static Cursor cursorOf(byte[]... keys) {
+ Cursor cursor = mock(Cursor.class);
+ List keyList = List.of(keys);
+ doAnswer(inv -> {
+ Consumer action = inv.getArgument(0);
+ keyList.forEach(action);
+ return null;
+ }).when(cursor).forEachRemaining(any(Consumer.class));
+ return cursor;
+ }
+}
diff --git a/common/transport/mqtt/pom.xml b/common/transport/mqtt/pom.xml
index 2714c5c667..f44278a7ee 100644
--- a/common/transport/mqtt/pom.xml
+++ b/common/transport/mqtt/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
transport
org.thingsboard.common.transport
diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
index 3d05e999e0..10245c5a24 100644
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
@@ -88,20 +88,30 @@ public class MqttTransportContext extends TransportContext {
@Value("${transport.mqtt.proxy_enabled:false}")
private boolean proxyEnabled;
- private final AtomicInteger connectionsCounter = new AtomicInteger();
+ private final AtomicInteger connectionsActiveCounterMQTT = new AtomicInteger();
+ private final AtomicInteger connectionsActiveCounterMQTTS = new AtomicInteger();
@PostConstruct
public void init() {
super.init();
- transportService.createGaugeStats("openConnections", connectionsCounter);
+ transportService.createGaugeStats("connections_active", connectionsActiveCounterMQTT, "protocol", "MQTT");
+ transportService.createGaugeStats("connections_active", connectionsActiveCounterMQTTS, "protocol", "MQTTS");
}
- public void channelRegistered() {
- connectionsCounter.incrementAndGet();
+ public void channelRegistered(boolean isSSL) {
+ if (isSSL) {
+ connectionsActiveCounterMQTTS.incrementAndGet();
+ } else {
+ connectionsActiveCounterMQTT.incrementAndGet();
+ }
}
- public void channelUnregistered() {
- connectionsCounter.decrementAndGet();
+ public void channelUnregistered(boolean isSSL) {
+ if (isSSL) {
+ connectionsActiveCounterMQTTS.decrementAndGet();
+ } else {
+ connectionsActiveCounterMQTT.decrementAndGet();
+ }
}
public boolean checkAddress(InetSocketAddress address) {
diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
index f753579292..48fbad3032 100644
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
@@ -180,16 +180,20 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
this.rpcAwaitingAck = new ConcurrentHashMap<>();
}
+ boolean isSSL() {
+ return sslHandler != null;
+ }
+
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
super.channelRegistered(ctx);
- context.channelRegistered();
+ context.channelRegistered(isSSL());
}
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
super.channelUnregistered(ctx);
- context.channelUnregistered();
+ context.channelUnregistered(isSSL());
}
@Override
@@ -254,6 +258,17 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
}
}
+ String getClientAddr(ChannelHandlerContext ctx) {
+ try {
+ InetSocketAddress remote = getAddress(ctx);
+ if (remote == null) return "unknown";
+ String host = remote.getAddress() != null ? remote.getAddress().getHostAddress() : remote.getHostString();
+ return host + ":" + remote.getPort();
+ } catch (Exception ignored) {
+ return "unknown";
+ }
+ }
+
InetSocketAddress getAddress(ChannelHandlerContext ctx) {
var address = ctx.channel().attr(MqttTransportService.ADDRESS).get();
if (address == null) {
@@ -1150,19 +1165,25 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
if (cause instanceof IOException) {
if (log.isDebugEnabled()) {
- log.debug("[{}][{}][{}] IOException: {}", sessionId,
+ String clientAddr = getClientAddr(ctx);
+ log.debug("[{}][{}][{}][{}] {}: {}", sessionId,
Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceId).orElse(null),
Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceName).orElse(""),
+ clientAddr,
+ cause.getClass().getSimpleName(),
cause.getMessage(),
cause);
} else if (log.isInfoEnabled()) {
- log.info("[{}][{}][{}] IOException: {}", sessionId,
+ String clientAddr = getClientAddr(ctx);
+ log.info("[{}][{}][{}][{}] {}: {}", sessionId,
Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceId).orElse(null),
Optional.ofNullable(this.deviceSessionCtx.getDeviceInfo()).map(TransportDeviceInfo::getDeviceName).orElse(""),
+ clientAddr,
+ cause.getClass().getSimpleName(),
cause.getMessage());
}
} else {
- log.error("[{}] Unexpected Exception", sessionId, cause);
+ log.error("[{}][{}] Unexpected Exception", sessionId, getClientAddr(ctx), cause);
}
closeCtx(ctx, MqttReasonCodes.Disconnect.SERVER_SHUTTING_DOWN);
diff --git a/common/transport/pom.xml b/common/transport/pom.xml
index 56ed9bbe58..b212bbf469 100644
--- a/common/transport/pom.xml
+++ b/common/transport/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/transport/snmp/pom.xml b/common/transport/snmp/pom.xml
index 79ab1d8f68..6d32b45a9c 100644
--- a/common/transport/snmp/pom.xml
+++ b/common/transport/snmp/pom.xml
@@ -21,7 +21,7 @@
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
transport
diff --git a/common/transport/transport-api/pom.xml b/common/transport/transport-api/pom.xml
index 7785e82b3d..68fcf31a5f 100644
--- a/common/transport/transport-api/pom.xml
+++ b/common/transport/transport-api/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard.common
- 4.3.1
+ 4.3.1.1-SNAPSHOT
transport
org.thingsboard.common.transport
diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
index 5c7d552855..a8f8a8b863 100644
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
@@ -161,5 +161,5 @@ public interface TransportService {
boolean hasSession(SessionInfoProto sessionInfo);
- void createGaugeStats(String openConnections, AtomicInteger connectionsCounter);
+ void createGaugeStats(String statsName, AtomicInteger number, String... tags);
}
diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/config/ssl/SslCredentialsWebServerCustomizer.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/config/ssl/SslCredentialsWebServerCustomizer.java
index e5f81dafc4..34cc1151c4 100644
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/config/ssl/SslCredentialsWebServerCustomizer.java
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/config/ssl/SslCredentialsWebServerCustomizer.java
@@ -30,6 +30,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import java.util.List;
+import java.util.function.BiConsumer;
import java.util.function.Consumer;
@Component
@@ -88,6 +89,11 @@ public class SslCredentialsWebServerCustomizer implements WebServerFactoryCustom
public void addBundleUpdateHandler(String name, Consumer handler) {
// no-op
}
+
+ @Override
+ public void addBundleRegisterHandler(BiConsumer handler) {
+ // no-op
+ }
};
}
diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
index 260957c990..e8ade498db 100644
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
@@ -1257,9 +1257,21 @@ public class DefaultTransportService extends TransportActivityManager implements
}
@Override
- public void createGaugeStats(String statsName, AtomicInteger number) {
- statsFactory.createGauge(StatsType.TRANSPORT + "." + statsName, number);
- statsMap.put(statsName, number);
+ public void createGaugeStats(String statsName, AtomicInteger number, String... tags) {
+ String key = "thingsboard" + "." + StatsType.TRANSPORT.getName() + "." + statsName;
+ statsFactory.createGauge(key, number, tags);
+ statsMap.put(statsName + tagsKey(tags), number);
+ }
+
+ private String tagsKey(String... tags) {
+ if (tags == null || tags.length < 2) return "";
+ StringBuilder sb = new StringBuilder("[");
+ for (int i = 0; i < tags.length; i += 2) {
+ if (i > 0) sb.append(',');
+ sb.append(tags[i]).append('=').append(i + 1 < tags.length ? tags[i + 1] : "");
+ }
+ sb.append(']');
+ return sb.toString();
}
@Scheduled(fixedDelayString = "${transport.stats.print-interval-ms:60000}")
diff --git a/common/util/pom.xml b/common/util/pom.xml
index a7009867b7..ac5955eb6a 100644
--- a/common/util/pom.xml
+++ b/common/util/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/common/util/src/main/java/org/thingsboard/common/util/SsrfProtectionValidator.java b/common/util/src/main/java/org/thingsboard/common/util/SsrfProtectionValidator.java
index 15da77f663..9132d117ae 100644
--- a/common/util/src/main/java/org/thingsboard/common/util/SsrfProtectionValidator.java
+++ b/common/util/src/main/java/org/thingsboard/common/util/SsrfProtectionValidator.java
@@ -38,6 +38,7 @@ public class SsrfProtectionValidator {
private static final Set BLOCKED_HOSTNAME_SUFFIXES = Set.of(".internal", ".local");
private static volatile AdditionalBlockedHosts additionalBlocked = AdditionalBlockedHosts.EMPTY;
+ private static volatile AllowedHosts allowedHosts = AllowedHosts.EMPTY;
// Well-known cloud metadata endpoints not covered by the JDK checks (isLoopback, isSiteLocal, isLinkLocal)
private static final List CLOUD_METADATA_RANGES = List.of(
@@ -66,6 +67,13 @@ public class SsrfProtectionValidator {
}
String hostLower = host.toLowerCase();
+
+ // Allow-listed hostnames bypass all hostname and IP checks
+ AllowedHosts currentAllowed = allowedHosts;
+ if (currentAllowed.hostnames.contains(hostLower)) {
+ return;
+ }
+
if (BLOCKED_HOSTNAMES.contains(hostLower) || additionalBlocked.hostnames.contains(hostLower)) {
throwBlockedHost(host);
}
@@ -98,7 +106,15 @@ public class SsrfProtectionValidator {
}
}
- private static boolean isBlockedAddress(InetAddress address) {
+ public static boolean isBlockedAddress(InetAddress address) {
+ // Check allow-list first: allowed addresses bypass all block checks
+ AllowedHosts currentAllowed = allowedHosts;
+ for (CidrRange cidr : currentAllowed.cidrRanges) {
+ if (cidr.contains(address)) {
+ return false;
+ }
+ }
+
// Covers 127.0.0.0/8 and ::1
if (address.isLoopbackAddress()) {
return true;
@@ -142,14 +158,37 @@ public class SsrfProtectionValidator {
throw new RuntimeException("URI is invalid: host '" + host + "' is not allowed");
}
+ public static boolean isEnabled() {
+ return enabled;
+ }
+
public static void setEnabled(boolean enabled) {
SsrfProtectionValidator.enabled = enabled;
}
public static void setAdditionalBlockedHosts(List entries) {
+ ParsedHostEntries parsed = parseHostEntries(entries);
+ additionalBlocked = new AdditionalBlockedHosts(parsed.cidrRanges, parsed.hostnames);
+ if (!parsed.cidrRanges.isEmpty() || !parsed.hostnames.isEmpty()) {
+ log.info("SSRF additional blocked hosts configured: {} CIDR range(s), {} hostname(s)", parsed.cidrRanges.size(), parsed.hostnames.size());
+ }
+ }
+
+ public static void setAllowedHosts(List entries) {
+ ParsedHostEntries parsed = parseHostEntries(entries);
+ allowedHosts = new AllowedHosts(parsed.cidrRanges, parsed.hostnames);
+ if (!parsed.cidrRanges.isEmpty() || !parsed.hostnames.isEmpty()) {
+ log.info("SSRF allowed hosts configured: {} CIDR range(s), {} hostname(s)", parsed.cidrRanges.size(), parsed.hostnames.size());
+ }
+ }
+
+ public static boolean isHostnameAllowed(String hostname) {
+ return allowedHosts.hostnames.contains(hostname.toLowerCase());
+ }
+
+ private static ParsedHostEntries parseHostEntries(List entries) {
if (entries == null || entries.isEmpty()) {
- additionalBlocked = AdditionalBlockedHosts.EMPTY;
- return;
+ return ParsedHostEntries.EMPTY;
}
List cidrRanges = new ArrayList<>();
Set hostnames = new HashSet<>();
@@ -168,10 +207,9 @@ public class SsrfProtectionValidator {
hostnames.add(trimmed.toLowerCase());
}
}
- additionalBlocked = new AdditionalBlockedHosts(
+ return new ParsedHostEntries(
Collections.unmodifiableList(cidrRanges),
Collections.unmodifiableSet(hostnames));
- log.info("SSRF additional blocked hosts configured: {} CIDR range(s), {} hostname(s)", cidrRanges.size(), hostnames.size());
}
private static boolean isIpLiteral(String entry) {
@@ -179,10 +217,18 @@ public class SsrfProtectionValidator {
return !entry.isEmpty() && (Character.isDigit(entry.charAt(0)) || entry.contains(":"));
}
+ private record ParsedHostEntries(List cidrRanges, Set hostnames) {
+ static final ParsedHostEntries EMPTY = new ParsedHostEntries(Collections.emptyList(), Collections.emptySet());
+ }
+
record AdditionalBlockedHosts(List cidrRanges, Set hostnames) {
static final AdditionalBlockedHosts EMPTY = new AdditionalBlockedHosts(Collections.emptyList(), Collections.emptySet());
}
+ record AllowedHosts(List cidrRanges, Set hostnames) {
+ static final AllowedHosts EMPTY = new AllowedHosts(Collections.emptyList(), Collections.emptySet());
+ }
+
record CidrRange(byte[] network, int prefixLength) {
static CidrRange of(String ip, int prefixLength) {
diff --git a/common/util/src/test/java/org/thingsboard/common/util/SsrfProtectionValidatorTest.java b/common/util/src/test/java/org/thingsboard/common/util/SsrfProtectionValidatorTest.java
index 6cb2d21a9a..77f915c09c 100644
--- a/common/util/src/test/java/org/thingsboard/common/util/SsrfProtectionValidatorTest.java
+++ b/common/util/src/test/java/org/thingsboard/common/util/SsrfProtectionValidatorTest.java
@@ -20,14 +20,16 @@ import org.junit.jupiter.api.parallel.ResourceLock;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.net.InetAddress;
import java.net.URI;
import java.util.Collections;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-@ResourceLock("SsrfProtectionValidatorTest") // some tests mutate static additional-blocked-hosts
+@ResourceLock("SsrfProtectionValidator") // to avoid race conditions when modifying SsrfProtectionValidator's static configuration
public class SsrfProtectionValidatorTest {
@ParameterizedTest
@@ -335,4 +337,244 @@ public class SsrfProtectionValidatorTest {
}
}
+ // --- Allow-list tests ---
+
+ @Test
+ void testAllowListCidrAllowsPrivateAddress() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("192.168.1.0/24"));
+ // 192.168.1.1 is normally blocked (site-local), but allow-listed
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.1.1"), true));
+ // Other private ranges remain blocked
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://10.0.0.1"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testAllowListHostnameBypassesSuffixCheck() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("my-device.local"));
+ // .local suffix is normally blocked, but allow-listed hostname passes
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://my-device.local/api"), true));
+ // Other .local hostnames remain blocked
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://other-device.local/api"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testAllowListPrecedenceOverBlockList() {
+ try {
+ // Block 8.8.8.0/24 via additional-blocked, but allow 8.8.8.8 via allow-list
+ SsrfProtectionValidator.setAdditionalBlockedHosts(List.of("8.8.8.0/24"));
+ SsrfProtectionValidator.setAllowedHosts(List.of("8.8.8.8"));
+ // Allow-list should win
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("https://8.8.8.8"), true));
+ // Adjacent IP still blocked
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("https://8.8.8.9"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ } finally {
+ SsrfProtectionValidator.setAdditionalBlockedHosts(Collections.emptyList());
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testIsBlockedAddressPublicApi() throws Exception {
+ InetAddress loopback = InetAddress.getByName("127.0.0.1");
+ assertThat(SsrfProtectionValidator.isBlockedAddress(loopback)).isTrue();
+
+ InetAddress publicIp = InetAddress.getByName("8.8.8.8");
+ assertThat(SsrfProtectionValidator.isBlockedAddress(publicIp)).isFalse();
+
+ // Allow-listed private address
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("10.0.0.0/8"));
+ InetAddress privateIp = InetAddress.getByName("10.1.2.3");
+ assertThat(SsrfProtectionValidator.isBlockedAddress(privateIp)).isFalse();
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testIsEnabledAccessor() {
+ boolean original = SsrfProtectionValidator.isEnabled();
+ try {
+ SsrfProtectionValidator.setEnabled(true);
+ assertThat(SsrfProtectionValidator.isEnabled()).isTrue();
+ SsrfProtectionValidator.setEnabled(false);
+ assertThat(SsrfProtectionValidator.isEnabled()).isFalse();
+ } finally {
+ SsrfProtectionValidator.setEnabled(original);
+ }
+ }
+
+ @Test
+ void testSetAllowedHostsEmptyAndNull() {
+ // Should not throw
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ SsrfProtectionValidator.setAllowedHosts(null);
+ }
+
+ @Test
+ void testIsHostnameAllowed() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("my-device.local", "Internal-Server.Corp"));
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("my-device.local")).isTrue();
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("MY-DEVICE.LOCAL")).isTrue(); // case-insensitive
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("internal-server.corp")).isTrue();
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("other-device.local")).isFalse();
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("example.com")).isFalse();
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testIsHostnameAllowedEmptyList() {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("anything")).isFalse();
+ }
+
+ @Test
+ void testValidateUriUsesStaticEnabledFlag() {
+ boolean original = SsrfProtectionValidator.isEnabled();
+ try {
+ // When enabled, loopback is blocked via the public one-arg overload
+ SsrfProtectionValidator.setEnabled(true);
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://127.0.0.1")))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+
+ // When disabled, loopback passes
+ SsrfProtectionValidator.setEnabled(false);
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://127.0.0.1")));
+ } finally {
+ SsrfProtectionValidator.setEnabled(original);
+ }
+ }
+
+ @Test
+ void testAllowListHostnameCaseInsensitive() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("My-Device.LOCAL"));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://my-device.local/api"), true));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://MY-DEVICE.LOCAL/api"), true));
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testAllowListOverridesCloudMetadataRange() {
+ try {
+ // 169.254.169.254 is link-local (blocked by default), allow-list should override
+ SsrfProtectionValidator.setAllowedHosts(List.of("169.254.169.254"));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://169.254.169.254/latest/meta-data/"), true));
+ // Other link-local still blocked
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://169.254.1.1"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testAllowListOverridesLoopback() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("127.0.0.0/8"));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://127.0.0.1"), true));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://127.1.2.3"), true));
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testAllowListCidrBoundary() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("192.168.1.0/24"));
+ // Last address in range
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.1.255"), true));
+ // First address outside range
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.2.0"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ // Different subnet entirely
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.0.1"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testBlockedIpv6UniqueLocal() throws Exception {
+ // fc00::/7 covers fc00:: through fdff::
+ InetAddress fc00 = InetAddress.getByName("fc00::1");
+ assertThat(SsrfProtectionValidator.isBlockedAddress(fc00)).isTrue();
+
+ InetAddress fdAddr = InetAddress.getByName("fd12:3456:789a::1");
+ assertThat(SsrfProtectionValidator.isBlockedAddress(fdAddr)).isTrue();
+
+ // fe00:: is NOT in fc00::/7 (it's in fe80::/10 link-local, but fe00:: without the 80 bits is different)
+ // 2001:db8:: is a public documentation prefix, not blocked
+ InetAddress publicV6 = InetAddress.getByName("2001:db8::1");
+ assertThat(SsrfProtectionValidator.isBlockedAddress(publicV6)).isFalse();
+ }
+
+ @Test
+ void testParseHostEntriesWithWhitespaceAndBlanks() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of(" 192.168.1.0/24 ", "", " ", "my-host.corp"));
+ // Trimmed CIDR works
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.1.1"), true));
+ // Trimmed hostname works
+ assertThat(SsrfProtectionValidator.isHostnameAllowed("my-host.corp")).isTrue();
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testSetAllowedHostsReplacePrevious() {
+ try {
+ SsrfProtectionValidator.setAllowedHosts(List.of("192.168.1.0/24"));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.1.1"), true));
+
+ // Replace with different range
+ SsrfProtectionValidator.setAllowedHosts(List.of("10.0.0.0/8"));
+ // Old range no longer allowed
+ assertThatThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://192.168.1.1"), true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("URI is invalid");
+ // New range allowed
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://10.1.2.3"), true));
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
+ @Test
+ void testAllowListHostnameBypassesBlockedHostname() {
+ try {
+ // "localhost" is in BLOCKED_HOSTNAMES; allow-listing it should let it through
+ SsrfProtectionValidator.setAllowedHosts(List.of("localhost"));
+ assertThatNoException().isThrownBy(() -> SsrfProtectionValidator.validateUri(URI.create("http://localhost/path"), true));
+ } finally {
+ SsrfProtectionValidator.setAllowedHosts(Collections.emptyList());
+ }
+ }
+
}
diff --git a/common/version-control/pom.xml b/common/version-control/pom.xml
index 5205057d39..90f6973b8d 100644
--- a/common/version-control/pom.xml
+++ b/common/version-control/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
common
org.thingsboard.common
diff --git a/dao/pom.xml b/dao/pom.xml
index 03030748d1..facac3465a 100644
--- a/dao/pom.xml
+++ b/dao/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
thingsboard
dao
@@ -280,7 +280,6 @@
org.apache.maven.plugins
maven-jar-plugin
- ${jar-plugin.version}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java b/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java
index a19ba6c0e7..4b469851cd 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java
@@ -15,6 +15,7 @@
*/
package org.thingsboard.server.dao;
+import org.hibernate.exception.ConstraintViolationException;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
@@ -202,4 +203,13 @@ public final class DaoUtil {
.collect(Collectors.toList());
}
+ public static ConstraintViolationException extractConstraintViolation(Throwable t) {
+ if (t instanceof ConstraintViolationException cve) {
+ return cve;
+ } else if (t != null && t.getCause() instanceof ConstraintViolationException cve) {
+ return cve;
+ }
+ return null;
+ }
+
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java
index 96a86073d4..8e1408f4fc 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java
@@ -50,7 +50,7 @@ public interface NotificationRequestDao extends Dao {
boolean existsByTenantIdAndStatusAndTemplateId(TenantId tenantId, NotificationRequestStatus status, NotificationTemplateId templateId);
- int removeAllByCreatedTimeBefore(long ts);
+ int removeByTenantIdAndCreatedTimeBeforeBatch(TenantId tenantId, long ts, int batchSize);
NotificationRequestInfo findInfoById(TenantId tenantId, NotificationRequestId id);
diff --git a/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java b/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java
index 484b3c49b3..2e30c0d598 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java
@@ -196,7 +196,9 @@ public class BaseOtaPackageService extends AbstractCachedEntityService INCORRECT_OTA_PACKAGE_ID + id);
try {
+ Long oid = getDataOidById(tenantId, otaPackageId);
otaPackageDao.removeById(tenantId, otaPackageId.getId());
+ unlinkDataIfPresent(tenantId, otaPackageId, oid);
publishEvictEvent(new OtaPackageCacheEvictEvent(otaPackageId));
eventPublisher.publishEvent(DeleteEntityEvent.builder().tenantId(tenantId).entityId(otaPackageId).build());
} catch (Exception t) {
@@ -215,6 +217,30 @@ public class BaseOtaPackageService extends AbstractCachedEntityService tenantOtaPackageRemover =
- new PaginatedRemover<>() {
-
- @Override
- protected PageData findEntities(TenantId tenantId, TenantId id, PageLink pageLink) {
- return otaPackageInfoDao.findOtaPackageInfoByTenantId(id, pageLink);
- }
+ private final PaginatedRemover tenantOtaPackageRemover = new PaginatedRemover<>() {
+ @Override
+ protected PageData findEntities(TenantId tenantId, TenantId id, PageLink pageLink) {
+ return otaPackageInfoDao.findOtaPackageInfoByTenantId(id, pageLink);
+ }
- @Override
- protected void removeEntity(TenantId tenantId, OtaPackageInfo entity) {
- deleteOtaPackage(tenantId, entity.getId());
- }
- };
+ @Override
+ protected void removeEntity(TenantId tenantId, OtaPackageInfo entity) {
+ deleteOtaPackage(tenantId, entity.getId());
+ }
+ };
@Override
public Optional> findEntity(TenantId tenantId, EntityId entityId) {
diff --git a/dao/src/main/java/org/thingsboard/server/dao/ota/OtaPackageDao.java b/dao/src/main/java/org/thingsboard/server/dao/ota/OtaPackageDao.java
index c11a13cbe1..875aaea4ed 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/ota/OtaPackageDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/ota/OtaPackageDao.java
@@ -18,15 +18,20 @@ package org.thingsboard.server.dao.ota;
import org.thingsboard.server.common.data.OtaPackage;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.data.ota.OtaPackageType;
import org.thingsboard.server.dao.Dao;
import org.thingsboard.server.dao.ExportableEntityDao;
import org.thingsboard.server.dao.TenantEntityWithDataDao;
+import java.util.UUID;
+
public interface OtaPackageDao extends Dao, TenantEntityWithDataDao, ExportableEntityDao {
Long sumDataSizeByTenantId(TenantId tenantId);
OtaPackage findOtaPackageByTenantIdAndTitleAndVersion(TenantId tenantId, String title, String version);
+ Long getDataOidById(UUID id);
+
+ Integer unlinkLargeObject(Long dataOid);
+
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java
index f88b672a34..37fe2950b4 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java
@@ -24,12 +24,13 @@ import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.dao.Dao;
public interface RpcDao extends Dao {
+
PageData findAllByDeviceId(TenantId tenantId, DeviceId deviceId, PageLink pageLink);
PageData findAllByDeviceIdAndStatus(TenantId tenantId, DeviceId deviceId, RpcStatus rpcStatus, PageLink pageLink);
PageData findAllRpcByTenantId(TenantId tenantId, PageLink pageLink);
- int deleteOutdatedRpcByTenantId(TenantId tenantId, Long expirationTime);
+ int deleteOutdatedRpcByTenantIdBatch(TenantId tenantId, Long expirationTime, int batchSize);
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/validator/Oauth2ClientDataValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/validator/Oauth2ClientDataValidator.java
index 42e872a87c..289220710c 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/service/validator/Oauth2ClientDataValidator.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/service/validator/Oauth2ClientDataValidator.java
@@ -17,6 +17,7 @@ package org.thingsboard.server.dao.service.validator;
import lombok.AllArgsConstructor;
import org.springframework.stereotype.Component;
+import org.thingsboard.common.util.SsrfProtectionValidator;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.oauth2.MapperType;
@@ -28,6 +29,8 @@ import org.thingsboard.server.common.data.oauth2.TenantNameStrategyType;
import org.thingsboard.server.exception.DataValidationException;
import org.thingsboard.server.dao.service.DataValidator;
+import java.net.URI;
+
@Component
@AllArgsConstructor
public class Oauth2ClientDataValidator extends DataValidator {
@@ -64,6 +67,11 @@ public class Oauth2ClientDataValidator extends DataValidator {
if (StringUtils.isEmpty(customConfig.getUrl())) {
throw new DataValidationException("Custom mapper URL should be specified!");
}
+ try {
+ SsrfProtectionValidator.validateUri(new URI(customConfig.getUrl()));
+ } catch (Exception e) {
+ throw new DataValidationException("Custom mapper URL is not allowed: " + e.getMessage());
+ }
}
}
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java
index 9d32e91ca2..f4cc406a75 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java
@@ -98,8 +98,8 @@ public class JpaNotificationRequestDao extends JpaAbstractDao findIdsByTenantId(@Param("tenantId") UUID tenantId, Pageable pageable);
+ // The 'data' column is of type OID (PostgreSQL large object reference), so it returns the OID as Long
+ @Query(value = "SELECT data FROM ota_package WHERE id = :id AND data IS NOT NULL", nativeQuery = true)
+ Long getDataOidById(@Param("id") UUID id);
+
+ @Transactional
+ @Query(value = "SELECT lo_unlink(:oid)", nativeQuery = true)
+ Integer unlinkLargeObject(@Param("oid") Long oid);
+
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java
index 9daae69d2a..a929b6d1e4 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java
@@ -71,8 +71,8 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao,
@Transactional
@Override
- public int deleteOutdatedRpcByTenantId(TenantId tenantId, Long expirationTime) {
- return rpcRepository.deleteOutdatedRpcByTenantId(tenantId.getId(), expirationTime);
+ public int deleteOutdatedRpcByTenantIdBatch(TenantId tenantId, Long expirationTime, int batchSize) {
+ return rpcRepository.deleteOutdatedRpcByTenantIdBatch(tenantId.getId(), expirationTime, batchSize);
}
@Override
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcRepository.java
index 8a9489333f..3f76170c84 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcRepository.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcRepository.java
@@ -21,20 +21,27 @@ import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
+import org.springframework.transaction.annotation.Transactional;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.dao.model.sql.RpcEntity;
import java.util.UUID;
public interface RpcRepository extends JpaRepository {
+
Page findAllByTenantIdAndDeviceId(UUID tenantId, UUID deviceId, Pageable pageable);
Page findAllByTenantIdAndDeviceIdAndStatus(UUID tenantId, UUID deviceId, RpcStatus status, Pageable pageable);
Page findAllByTenantId(UUID tenantId, Pageable pageable);
+ @Transactional
@Modifying
- @Query(value = "DELETE FROM rpc WHERE tenant_id = :tenantId AND created_time < :expirationTime",
+ @Query(value = "DELETE FROM rpc WHERE id IN " +
+ "(SELECT id FROM rpc WHERE tenant_id = :tenantId AND created_time < :expirationTime LIMIT :batchSize)",
nativeQuery = true)
- int deleteOutdatedRpcByTenantId(@Param("tenantId") UUID tenantId, @Param("expirationTime") Long expirationTime);
+ int deleteOutdatedRpcByTenantIdBatch(@Param("tenantId") UUID tenantId,
+ @Param("expirationTime") Long expirationTime,
+ @Param("batchSize") int batchSize);
+
}
diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/OtaPackageServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/OtaPackageServiceTest.java
index 002a5a7845..69fa3fbc5d 100644
--- a/dao/src/test/java/org/thingsboard/server/dao/service/OtaPackageServiceTest.java
+++ b/dao/src/test/java/org/thingsboard/server/dao/service/OtaPackageServiceTest.java
@@ -36,14 +36,14 @@ import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.dao.device.DeviceProfileService;
import org.thingsboard.server.dao.device.DeviceService;
-import org.thingsboard.server.exception.DataValidationException;
+import org.thingsboard.server.dao.ota.OtaPackageDao;
import org.thingsboard.server.dao.ota.OtaPackageService;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.dao.tenant.TenantProfileService;
+import org.thingsboard.server.exception.DataValidationException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
@@ -77,6 +77,8 @@ public class OtaPackageServiceTest extends AbstractServiceTest {
@Autowired
TenantProfileService tenantProfileService;
@Autowired
+ OtaPackageDao otaPackageDao;
+ @Autowired
TbTenantProfileCache tenantProfileCache;
@Before
@@ -118,10 +120,8 @@ public class OtaPackageServiceTest extends AbstractServiceTest {
Assert.assertEquals(1, otaPackageService.sumDataSizeByTenantId(tenantId));
int maxSumDataSize = 8;
- List packages = new ArrayList<>(maxSumDataSize);
-
for (int i = 2; i <= maxSumDataSize; i++) {
- packages.add(createAndSaveFirmware(tenantId, "0." + i));
+ createAndSaveFirmware(tenantId, "0." + i);
Assert.assertEquals(i, otaPackageService.sumDataSizeByTenantId(tenantId));
}
@@ -533,6 +533,39 @@ public class OtaPackageServiceTest extends AbstractServiceTest {
Assert.assertNull(foundFirmware);
}
+ @Test
+ public void testDeleteOtaPackageWithoutData() {
+ OtaPackageInfo firmwareInfo = new OtaPackageInfo();
+ firmwareInfo.setTenantId(tenantId);
+ firmwareInfo.setDeviceProfileId(deviceProfileId);
+ firmwareInfo.setType(FIRMWARE);
+ firmwareInfo.setTitle(TITLE);
+ firmwareInfo.setVersion(VERSION);
+ OtaPackageInfo savedFirmwareInfo = otaPackageService.saveOtaPackageInfo(firmwareInfo, false);
+
+ Assert.assertNotNull(savedFirmwareInfo);
+ Assert.assertNotNull(savedFirmwareInfo.getId());
+
+ // Should not throw NPE when deleting package without data (OID is null)
+ otaPackageService.deleteOtaPackage(tenantId, savedFirmwareInfo.getId());
+
+ OtaPackageInfo foundFirmware = otaPackageService.findOtaPackageInfoById(tenantId, savedFirmwareInfo.getId());
+ Assert.assertNull(foundFirmware);
+ }
+
+ @Test
+ public void testDeleteOtaPackageUnlinksLargeObject() {
+ OtaPackage savedFirmware = createAndSaveFirmware(tenantId, VERSION);
+
+ Long oid = otaPackageDao.getDataOidById(savedFirmware.getId().getId());
+ Assert.assertNotNull(oid);
+
+ otaPackageService.deleteOtaPackage(tenantId, savedFirmware.getId());
+
+ // Verify the large object was unlinked - PostgreSQL throws an exception when the object doesn't exist
+ assertThatThrownBy(() -> otaPackageDao.unlinkLargeObject(oid)).hasMessageContaining("large object " + oid + " does not exist");
+ }
+
@Test
public void testFindTenantFirmwaresByTenantId() {
List firmwares = new ArrayList<>();
@@ -567,8 +600,8 @@ public class OtaPackageServiceTest extends AbstractServiceTest {
}
} while (pageData.hasNext());
- Collections.sort(firmwares, idComparator);
- Collections.sort(loadedFirmwares, idComparator);
+ firmwares.sort(idComparator);
+ loadedFirmwares.sort(idComparator);
assertThat(firmwares).isEqualTo(loadedFirmwares);
@@ -622,8 +655,8 @@ public class OtaPackageServiceTest extends AbstractServiceTest {
}
} while (pageData.hasNext());
- Collections.sort(firmwares, idComparator);
- Collections.sort(loadedFirmwares, idComparator);
+ firmwares.sort(idComparator);
+ loadedFirmwares.sort(idComparator);
assertThat(firmwares).isEqualTo(loadedFirmwares);
@@ -726,4 +759,5 @@ public class OtaPackageServiceTest extends AbstractServiceTest {
firmware.setDataSize(DATA_SIZE);
return firmware;
}
+
}
diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java
index aa43826734..b66cdb4363 100644
--- a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java
+++ b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java
@@ -65,7 +65,7 @@ public class TimeseriesServiceNoSqlTest extends BaseTimeseriesServiceTest {
new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(5), new JsonDataEntry("test", "{\"test\":\"testValue\"}")));
DeviceId deviceId = new DeviceId(Uuids.timeBased());
- tsService.save(tenantId, deviceId, timeseries, ttlInSec);
+ tsService.save(tenantId, deviceId, timeseries, ttlInSec).get(MAX_TIMEOUT, TimeUnit.SECONDS);
List fullList = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("test", 0L,
TimeUnit.MINUTES.toMillis(6), 1000, 10, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS);
diff --git a/dao/src/test/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDaoTest.java b/dao/src/test/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDaoTest.java
new file mode 100644
index 0000000000..5e2fe2308b
--- /dev/null
+++ b/dao/src/test/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDaoTest.java
@@ -0,0 +1,133 @@
+/**
+ * Copyright © 2016-2026 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.sql.notification;
+
+import org.junit.After;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.thingsboard.server.common.data.id.NotificationRequestId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.notification.NotificationRequest;
+import org.thingsboard.server.common.data.notification.NotificationRequestStatus;
+import org.thingsboard.server.dao.AbstractJpaDaoTest;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class JpaNotificationRequestDaoTest extends AbstractJpaDaoTest {
+
+ @Autowired
+ JpaNotificationRequestDao notificationRequestDao;
+
+ private final List createdRequests = new ArrayList<>();
+
+ @After
+ public void tearDown() {
+ for (NotificationRequest request : createdRequests) {
+ notificationRequestDao.removeById(request.getTenantId(), request.getId().getId());
+ }
+ createdRequests.clear();
+ }
+
+ @Test
+ public void testBatchDeletion() {
+ TenantId sysTenantId = TenantId.SYS_TENANT_ID;
+ long now = System.currentTimeMillis();
+ long oldTimestamp = now - TimeUnit.DAYS.toMillis(30);
+
+ NotificationRequest oldRequest1 = createNotificationRequest(sysTenantId, oldTimestamp);
+ notificationRequestDao.save(sysTenantId, oldRequest1);
+
+ NotificationRequest oldRequest2 = createNotificationRequest(sysTenantId, oldTimestamp);
+ notificationRequestDao.save(sysTenantId, oldRequest2);
+
+ NotificationRequest freshRequest = createNotificationRequest(sysTenantId, now);
+ notificationRequestDao.save(sysTenantId, freshRequest);
+
+ TenantId tenant2Id = TenantId.fromUUID(UUID.fromString("3d193a7a-774b-4c05-84d5-f7fdcf7a37cf"));
+ NotificationRequest tenant2Request = createNotificationRequest(tenant2Id, oldTimestamp);
+ notificationRequestDao.save(tenant2Id, tenant2Request);
+
+ int batchSize = 10_000;
+
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(sysTenantId, oldTimestamp - 1, batchSize)).isEqualTo(0);
+
+ long expirationTime = now - TimeUnit.DAYS.toMillis(15);
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(sysTenantId, expirationTime, batchSize)).isEqualTo(2);
+
+ assertThat(notificationRequestDao.findById(sysTenantId, freshRequest.getId().getId())).isNotNull();
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenant2Id, now + 1, batchSize)).isEqualTo(1);
+ }
+
+ @Test
+ public void testBatchDeletionWithSmallBatchSize() {
+ TenantId tenantId = TenantId.SYS_TENANT_ID;
+ long oldTimestamp = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30);
+
+ for (int i = 0; i < 10; i++) {
+ NotificationRequest request = createNotificationRequest(tenantId, oldTimestamp);
+ notificationRequestDao.save(tenantId, request);
+ }
+
+ int batchSize = 3;
+ long expirationTime = System.currentTimeMillis();
+
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(3);
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(3);
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(3);
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(1);
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(0);
+ }
+
+ @Test
+ public void testBatchDeletionIsolationBetweenTenants() {
+ TenantId tenant1 = TenantId.SYS_TENANT_ID;
+ TenantId tenant2 = TenantId.fromUUID(UUID.fromString("3d193a7a-774b-4c05-84d5-f7fdcf7a37cf"));
+ long oldTimestamp = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30);
+
+ for (int i = 0; i < 5; i++) {
+ NotificationRequest request = createNotificationRequest(tenant1, oldTimestamp);
+ notificationRequestDao.save(tenant1, request);
+ }
+
+ for (int i = 0; i < 3; i++) {
+ NotificationRequest request = createNotificationRequest(tenant2, oldTimestamp);
+ notificationRequestDao.save(tenant2, request);
+ }
+
+ int batchSize = 10_000;
+ long expirationTime = System.currentTimeMillis();
+
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenant1, expirationTime, batchSize)).isEqualTo(5);
+ assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenant2, expirationTime, batchSize)).isEqualTo(3);
+ }
+
+ private NotificationRequest createNotificationRequest(TenantId tenantId, long createdTime) {
+ NotificationRequest request = new NotificationRequest();
+ request.setId(new NotificationRequestId(UUID.randomUUID()));
+ request.setTenantId(tenantId);
+ request.setCreatedTime(createdTime);
+ request.setTargets(List.of(UUID.randomUUID()));
+ request.setStatus(NotificationRequestStatus.SENT);
+ createdRequests.add(request);
+ return request;
+ }
+
+}
diff --git a/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java b/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java
index 1629922685..921339a92b 100644
--- a/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java
+++ b/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java
@@ -51,9 +51,10 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest {
rpc.setDeviceId(new DeviceId(UUID.randomUUID()));
rpcDao.saveAndFlush(rpc.getTenantId(), rpc);
- assertThat(rpcDao.deleteOutdatedRpcByTenantId(TenantId.SYS_TENANT_ID, 0L)).isEqualTo(0);
- assertThat(rpcDao.deleteOutdatedRpcByTenantId(TenantId.SYS_TENANT_ID, Long.MAX_VALUE)).isEqualTo(2);
- assertThat(rpcDao.deleteOutdatedRpcByTenantId(tenantId, System.currentTimeMillis() + 1)).isEqualTo(1);
+ int batchSize = 10_000;
+ assertThat(rpcDao.deleteOutdatedRpcByTenantIdBatch(TenantId.SYS_TENANT_ID, 0L, batchSize)).isEqualTo(0);
+ assertThat(rpcDao.deleteOutdatedRpcByTenantIdBatch(TenantId.SYS_TENANT_ID, Long.MAX_VALUE, batchSize)).isEqualTo(2);
+ assertThat(rpcDao.deleteOutdatedRpcByTenantIdBatch(tenantId, System.currentTimeMillis() + 1, batchSize)).isEqualTo(1);
}
}
diff --git a/docker/monitoring/grafana/provisioning/dashboards/transport_connections.json b/docker/monitoring/grafana/provisioning/dashboards/transport_connections.json
new file mode 100644
index 0000000000..f092315e94
--- /dev/null
+++ b/docker/monitoring/grafana/provisioning/dashboards/transport_connections.json
@@ -0,0 +1,458 @@
+{
+ "annotations": {
+ "list": [
+ {
+ "builtIn": 1,
+ "datasource": {
+ "type": "datasource",
+ "uid": "grafana"
+ },
+ "enable": true,
+ "hide": true,
+ "iconColor": "rgba(0, 211, 255, 1)",
+ "name": "Annotations & Alerts",
+ "type": "dashboard"
+ }
+ ]
+ },
+ "editable": true,
+ "fiscalYearStartMonth": 0,
+ "graphTooltip": 0,
+ "id": null,
+ "links": [],
+ "panels": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "description": "",
+ "fieldConfig": {
+ "defaults": {
+ "color": {
+ "mode": "palette-classic"
+ },
+ "custom": {
+ "axisBorderShow": false,
+ "axisCenteredZero": false,
+ "axisColorMode": "text",
+ "axisLabel": "",
+ "axisPlacement": "auto",
+ "barAlignment": 0,
+ "drawStyle": "line",
+ "fillOpacity": 10,
+ "gradientMode": "none",
+ "hideFrom": {
+ "legend": false,
+ "tooltip": false,
+ "viz": false
+ },
+ "insertNulls": false,
+ "lineInterpolation": "linear",
+ "lineWidth": 1,
+ "pointSize": 5,
+ "scaleDistribution": {
+ "type": "linear"
+ },
+ "showPoints": "never",
+ "spanNulls": true,
+ "stacking": {
+ "group": "A",
+ "mode": "none"
+ },
+ "thresholdsStyle": {
+ "mode": "off"
+ }
+ },
+ "mappings": [],
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "dark-orange",
+ "value": null
+ }
+ ]
+ },
+ "unit": "short"
+ },
+ "overrides": [
+ {
+ "matcher": {
+ "id": "byName",
+ "options": "Total connections"
+ },
+ "properties": [
+ {
+ "id": "color",
+ "value": {
+ "fixedColor": "light-orange",
+ "mode": "fixed"
+ }
+ }
+ ]
+ }
+ ]
+ },
+ "gridPos": {
+ "h": 15,
+ "w": 19,
+ "x": 0,
+ "y": 0
+ },
+ "id": 19,
+ "options": {
+ "legend": {
+ "calcs": [],
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "editorMode": "code",
+ "exemplar": false,
+ "expr": "sum by (protocol) (thingsboard_transport_connections_active{})",
+ "format": "time_series",
+ "instant": false,
+ "interval": "",
+ "intervalFactor": 1,
+ "legendFormat": "{{protocol}}",
+ "refId": "A"
+ }
+ ],
+ "title": "Active Connections by Protocol",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "fieldConfig": {
+ "defaults": {
+ "color": {
+ "mode": "thresholds"
+ },
+ "mappings": [],
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "rgba(209, 138, 78, 1)",
+ "value": null
+ }
+ ]
+ }
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 15,
+ "w": 5,
+ "x": 19,
+ "y": 0
+ },
+ "id": 21,
+ "options": {
+ "colorMode": "value",
+ "graphMode": "none",
+ "justifyMode": "auto",
+ "orientation": "auto",
+ "percentChangeColorMode": "standard",
+ "reduceOptions": {
+ "calcs": [
+ "lastNotNull"
+ ],
+ "fields": "",
+ "values": false
+ },
+ "showPercentChange": false,
+ "text": {},
+ "textMode": "auto",
+ "wideLayout": true
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "editorMode": "code",
+ "exemplar": true,
+ "expr": "sum (thingsboard_transport_connections_active)",
+ "instant": true,
+ "interval": "",
+ "legendFormat": "",
+ "refId": "A"
+ }
+ ],
+ "title": "Total Connections",
+ "type": "stat"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "description": "",
+ "fieldConfig": {
+ "defaults": {
+ "color": {
+ "mode": "palette-classic"
+ },
+ "custom": {
+ "axisBorderShow": false,
+ "axisCenteredZero": false,
+ "axisColorMode": "text",
+ "axisLabel": "",
+ "axisPlacement": "auto",
+ "barAlignment": 0,
+ "drawStyle": "line",
+ "fillOpacity": 10,
+ "gradientMode": "none",
+ "hideFrom": {
+ "legend": false,
+ "tooltip": false,
+ "viz": false
+ },
+ "insertNulls": false,
+ "lineInterpolation": "linear",
+ "lineWidth": 1,
+ "pointSize": 5,
+ "scaleDistribution": {
+ "type": "linear"
+ },
+ "showPoints": "never",
+ "spanNulls": true,
+ "stacking": {
+ "group": "A",
+ "mode": "none"
+ },
+ "thresholdsStyle": {
+ "mode": "off"
+ }
+ },
+ "mappings": [],
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "green",
+ "value": null
+ },
+ {
+ "color": "red",
+ "value": 80
+ }
+ ]
+ },
+ "unit": "short"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 15,
+ "w": 19,
+ "x": 0,
+ "y": 15
+ },
+ "id": 22,
+ "options": {
+ "legend": {
+ "calcs": [],
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "editorMode": "code",
+ "exemplar": true,
+ "expr": "thingsboard_transport_connections_active",
+ "format": "time_series",
+ "interval": "",
+ "intervalFactor": 1,
+ "legendFormat": "{{protocol}} {{pod_name}}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "Active Connections by Pod",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "fieldConfig": {
+ "defaults": {
+ "color": {
+ "mode": "thresholds"
+ },
+ "custom": {
+ "align": "auto",
+ "cellOptions": {
+ "type": "auto"
+ },
+ "filterable": false,
+ "inspect": false
+ },
+ "mappings": [],
+ "min": 0,
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "rgba(209, 138, 78, 1)",
+ "value": null
+ }
+ ]
+ }
+ },
+ "overrides": [
+ {
+ "matcher": {
+ "id": "byName",
+ "options": "Connections"
+ },
+ "properties": [
+ {
+ "id": "custom.cellOptions",
+ "value": {
+ "mode": "gradient",
+ "type": "gauge"
+ }
+ }
+ ]
+ }
+ ]
+ },
+ "gridPos": {
+ "h": 15,
+ "w": 5,
+ "x": 19,
+ "y": 15
+ },
+ "id": 23,
+ "options": {
+ "cellHeight": "sm",
+ "footer": {
+ "countRows": false,
+ "fields": "",
+ "reducer": [
+ "sum"
+ ],
+ "show": false
+ },
+ "showHeader": true,
+ "sortBy": []
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "prometheus"
+ },
+ "editorMode": "code",
+ "exemplar": true,
+ "expr": "sum by (pod_name) (thingsboard_transport_connections_active)",
+ "format": "table",
+ "instant": true,
+ "interval": "",
+ "legendFormat": "__auto",
+ "queryType": "randomWalk",
+ "refId": "A"
+ }
+ ],
+ "title": "Connection Count by Pod",
+ "transformations": [
+ {
+ "id": "organize",
+ "options": {
+ "excludeByName": {
+ "Time": true,
+ "__name__": true,
+ "app": true,
+ "controller_revision_hash": true,
+ "instance": true,
+ "job": true,
+ "kubernetes_name": true,
+ "kubernetes_namespace": true,
+ "node_name": true,
+ "pod_name": false,
+ "statefulset_kubernetes_io_pod_name": true
+ },
+ "indexByName": {},
+ "renameByName": {
+ "Time": "",
+ "Value": "Connections",
+ "pod_name": "Pod"
+ }
+ }
+ },
+ {
+ "id": "sortBy",
+ "options": {
+ "fields": {},
+ "sort": [
+ {
+ "field": "Pod"
+ }
+ ]
+ }
+ }
+ ],
+ "type": "table"
+ }
+ ],
+ "refresh": "30s",
+ "schemaVersion": 39,
+ "tags": [
+ "thingsboard"
+ ],
+ "templating": {
+ "list": []
+ },
+ "time": {
+ "from": "now-1h",
+ "to": "now"
+ },
+ "timepicker": {
+ "refresh_intervals": [
+ "10s",
+ "30s",
+ "1m",
+ "5m",
+ "15m",
+ "30m",
+ "1h",
+ "2h",
+ "1d"
+ ]
+ },
+ "timezone": "",
+ "title": "Transport Connections",
+ "uid": "lewbrlwjerwkj3",
+ "version": 1,
+ "weekStart": ""
+}
diff --git a/edqs/pom.xml b/edqs/pom.xml
index ee12918b91..79cfe1fd32 100644
--- a/edqs/pom.xml
+++ b/edqs/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
thingsboard
edqs
@@ -33,7 +33,6 @@
UTF-8
${basedir}/..
java
- false
process-resources
package
tb-edqs
@@ -179,10 +178,6 @@
org.apache.maven.plugins
maven-assembly-plugin
-
- org.apache.maven.plugins
- maven-install-plugin
-
org.xolstice.maven.plugins
protobuf-maven-plugin
diff --git a/monitoring/pom.xml b/monitoring/pom.xml
index e5a102ea1f..40acb51649 100644
--- a/monitoring/pom.xml
+++ b/monitoring/pom.xml
@@ -21,7 +21,7 @@
4.0.0
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
thingsboard
@@ -33,7 +33,6 @@
UTF-8
${basedir}/..
java
- false
process-resources
package
tb-monitoring
@@ -154,8 +153,8 @@
maven-assembly-plugin
- org.apache.maven.plugins
- maven-install-plugin
+ org.codehaus.mojo
+ build-helper-maven-plugin
diff --git a/msa/black-box-tests/pom.xml b/msa/black-box-tests/pom.xml
index b96d11dc5f..4718b2222c 100644
--- a/msa/black-box-tests/pom.xml
+++ b/msa/black-box-tests/pom.xml
@@ -21,7 +21,7 @@
org.thingsboard
- 4.3.1
+ 4.3.1.1-SNAPSHOT
msa
org.thingsboard.msa
@@ -128,42 +128,50 @@
org.thingsboard.msa
js-executor
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa
web-ui
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa
tb-node
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa.transport
coap
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa.transport
http
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa.transport
mqtt
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa.transport
lwm2m
- docker-info
+ ${project.version}
+ pom
org.thingsboard.msa.transport
snmp
- docker-info
+ ${project.version}
+ pom
org.thingsboard.common
@@ -206,7 +214,7 @@
org.apache.maven.surefire
surefire-testng
- ${surefire.version}
+ ${maven-surefire-plugin.version}