Merge pull request #14987 from thingsboard/sparkplug-any-name-node-unique-device-names

sparkplug-unique-device-names

application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java

@@ -22,6 +22,7 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ByteString;
+import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.exception.EntitiesLimitExceededException;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
@@ -114,6 +115,7 @@ import java.util.stream.Collectors;
 
 import static org.thingsboard.server.service.transport.BasicCredentialsValidationResult.PASSWORD_MISMATCH;
 import static org.thingsboard.server.service.transport.BasicCredentialsValidationResult.VALID;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_SEPARATOR;
 
 /**
  * Created by ashvayka on 05.10.18.
@@ -330,84 +332,203 @@ public class DefaultTransportApiService implements TransportApiService {
     }
 
     private TransportApiResponseMsg handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
-        DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
+        DeviceId gatewayId = toDeviceId(requestMsg);
         Device gateway = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, gatewayId);
-        Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock());
+        String deviceName = requestMsg.getDeviceName();
-        deviceCreationLock.lock();
+        Lock lock = deviceCreationLocks.computeIfAbsent(deviceName, k -> new ReentrantLock());
+        lock.lock();
         try {
-            Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName());
+            Device device = findOrCreateDevice(requestMsg, gateway, gatewayId);
-            if (device == null) {
+            updateLastConnectedGateway(device, gatewayId);
-                TenantId tenantId = gateway.getTenantId();
+            return buildResponse(device);
-                device = new Device();
-                device.setTenantId(tenantId);
-                device.setName(requestMsg.getDeviceName());
-                device.setType(requestMsg.getDeviceType());
-                device.setCustomerId(gateway.getCustomerId());
-                DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
-
-                device.setDeviceProfileId(deviceProfile.getId());
-                ObjectNode additionalInfo = JacksonUtil.newObjectNode();
-                additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
-                device.setAdditionalInfo(additionalInfo);
-                device = deviceService.saveDevice(device);
-
-                relationService.saveRelation(tenantId, new EntityRelation(gateway.getId(), device.getId(), "Created"));
-
-                TbMsgMetaData metaData = new TbMsgMetaData();
-                CustomerId customerId = gateway.getCustomerId();
-                if (customerId != null && !customerId.isNullUid()) {
-                    metaData.putValue("customerId", customerId.toString());
-                }
-                metaData.putValue("gatewayId", gatewayId.toString());
-
-                DeviceId deviceId = device.getId();
-                JsonNode entityNode = JacksonUtil.valueToTree(device);
-                TbMsg tbMsg = TbMsg.newMsg()
-                        .type(TbMsgType.ENTITY_CREATED)
-                        .originator(deviceId)
-                        .customerId(customerId)
-                        .copyMetaData(metaData)
-                        .dataType(TbMsgDataType.JSON)
-                        .data(JacksonUtil.toString(entityNode))
-                        .build();
-                tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null);
-            } else {
-                JsonNode deviceAdditionalInfo = device.getAdditionalInfo();
-                if (deviceAdditionalInfo == null) {
-                    deviceAdditionalInfo = JacksonUtil.newObjectNode();
-                }
-                if (deviceAdditionalInfo.isObject() &&
-                        (!deviceAdditionalInfo.has(DataConstants.LAST_CONNECTED_GATEWAY)
-                                || !gatewayId.toString().equals(deviceAdditionalInfo.get(DataConstants.LAST_CONNECTED_GATEWAY).asText()))) {
-                    ObjectNode newDeviceAdditionalInfo = (ObjectNode) deviceAdditionalInfo;
-                    newDeviceAdditionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
-                    deviceService.saveDevice(device);
-                }
-            }
-            GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
-                    .setDeviceInfo(ProtoUtils.toDeviceInfoProto(device));
-            DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
-            if (deviceProfile != null) {
-                builder.setDeviceProfile(ProtoUtils.toProto(deviceProfile));
-            } else {
-                log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
-            }
-            return TransportApiResponseMsg.newBuilder()
-                    .setGetOrCreateDeviceResponseMsg(builder.build())
-                    .build();
         } catch (JsonProcessingException e) {
-            log.warn("[{}] Failed to lookup device by gateway id and name: [{}]", gatewayId, requestMsg.getDeviceName(), e);
+            log.warn("[{}] Failed to process device [{}]", gatewayId, deviceName, e);
             throw new RuntimeException(e);
         } catch (EntitiesLimitExceededException e) {
-            log.warn("[{}][{}] API limit exception: [{}]", e.getTenantId(), gatewayId, e.getMessage());
+            return buildLimitErrorResponse(e, gatewayId);
-            return TransportApiResponseMsg.newBuilder()
-                    .setGetOrCreateDeviceResponseMsg(
-                            GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
-                                    .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT))
-                    .build();
         } finally {
-            deviceCreationLock.unlock();
+            lock.unlock();
+        }
+    }
+
+    private DeviceId toDeviceId(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
+        return new DeviceId(new UUID(
+                requestMsg.getGatewayIdMSB(),
+                requestMsg.getGatewayIdLSB()
+        ));
+    }
+
+    private Device findOrCreateDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg,
+                                      Device gateway,
+                                      DeviceId gatewayId) throws JsonProcessingException {
+        TenantId tenantId = gateway.getTenantId();
+        String deviceName = requestMsg.getDeviceName();
+        Device device = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName);
+        if (device != null) {
+            return device;
+        }
+        String[] topicPath = requestMsg.getDeviceName().split(DEVICE_NAME_SPLIT_SEPARATOR);
+        device = tryRenameSparkplugDevice(requestMsg, gateway, topicPath);
+        if (device != null) {
+            return device;
         }
+        device = createNewDevice(requestMsg, gateway, gatewayId, topicPath);
+        pushCreatedEvent(device, gateway);
+        return device;
+    }
+
+    private Device tryRenameSparkplugDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg, Device gateway, String[] topicPath) {
+        if (!requestMsg.getIsSparkplug()) {
+            return null;
+        }
+
+        if (topicPath.length != 3) {
+            return null;
+        }
+
+        String deviceId = topicPath[2];
+        Device existingDevice = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), deviceId);
+
+        if (existingDevice == null) {
+            return null;
+        }
+
+        // Security check: verify that the device was created by this gateway
+        boolean isRelated = false;
+        try {
+            // Security check: verify that the device was originally created by this gateway
+            isRelated = relationService.checkRelation(
+                    gateway.getTenantId(),
+                    gateway.getId(),
+                    existingDevice.getId(),
+                    "Created",
+                    RelationTypeGroup.COMMON
+            );
+        } catch (Exception e) {
+            // Log the error from the relation service but return null to allow potential recovery
+            log.error("[{}] Error checking relation for device {}", gateway.getId(), existingDevice.getId(), e);
+            return null;
+        }
+
+        // If the device is found but not related to this gateway, it's a security breach
+        if (!isRelated) {
+            log.error("[{}] Security breach attempt! Gateway tried to rename device [{}] without 'Created' relation.",
+                    gateway.getId(), existingDevice.getId());
+            // Throwing exception to halt the entire connection process
+            throw new RuntimeException("Security breach attempt! Unauthorized device rename.");
+        }
+
+        // Logic for renaming the device if it's related and no naming conflicts exist
+        boolean changed = false;
+        String newName = requestMsg.getDeviceName();
+
+        if (!newName.equals(existingDevice.getName())) {
+            // Check if the new name is already taken by another device
+            Device conflictDevice = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), newName);
+
+            if (conflictDevice != null) {
+                log.warn("[{}] Cannot rename device [{}] to [{}]: name already exists!",
+                        gateway.getId(), existingDevice.getId(), newName);
+                return existingDevice;
+            }
+
+            existingDevice.setName(newName);
+
+            // Update label only if it's empty to avoid overwriting user changes
+            if (existingDevice.getLabel() == null || existingDevice.getLabel().isEmpty()) {
+                existingDevice.setLabel(deviceId);
+            }
+
+            changed = true;
+        }
+
+        return changed ? deviceService.saveDevice(existingDevice) : existingDevice;
+    }
+
+    private Device createNewDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg,
+                                   Device gateway,
+                                   DeviceId gatewayId,  String[] topicPath) {
+        TenantId tenantId = gateway.getTenantId();
+        Device device = new Device();
+        device.setTenantId(tenantId);
+        device.setName(requestMsg.getDeviceName());
+        if (requestMsg.getIsSparkplug()) {
+            if (topicPath.length == 3) device.setLabel(topicPath[2]);
+        }
+        device.setType(requestMsg.getDeviceType());
+        device.setCustomerId(gateway.getCustomerId());
+        DeviceProfile profile =
+                deviceProfileCache.findOrCreateDeviceProfile(tenantId, requestMsg.getDeviceType());
+        device.setDeviceProfileId(profile.getId());
+        ObjectNode additionalInfo = JacksonUtil.newObjectNode();
+        additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
+        device.setAdditionalInfo(additionalInfo);
+        device = deviceService.saveDevice(device);
+        relationService.saveRelation(
+                tenantId,
+                new EntityRelation(gateway.getId(), device.getId(), "Created")
+        );
+        return device;
+    }
+
+    private void updateLastConnectedGateway(Device device, DeviceId gatewayId) {
+        String gatewayIdStr = gatewayId.toString();
+        JsonNode info = device.getAdditionalInfo();
+        ObjectNode objectNode = (info instanceof ObjectNode)
+                ? (ObjectNode) info
+                : JacksonUtil.newObjectNode();
+        if (!objectNode.has(DataConstants.LAST_CONNECTED_GATEWAY)
+                || !gatewayIdStr.equals(objectNode.get(DataConstants.LAST_CONNECTED_GATEWAY).asText())) {
+            objectNode.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayIdStr);
+            device.setAdditionalInfo(objectNode);
+            deviceService.saveDevice(device);
+        }
+    }
+
+    private void pushCreatedEvent(Device device, Device gateway) {
+        TenantId tenantId = gateway.getTenantId();
+        CustomerId customerId = gateway.getCustomerId();
+        TbMsgMetaData metaData = new TbMsgMetaData();
+        metaData.putValue("gatewayId", gateway.getId().toString());
+        if (customerId != null && !customerId.isNullUid()) {
+            metaData.putValue("customerId", customerId.toString());
+        }
+        JsonNode entityNode = JacksonUtil.valueToTree(device);
+        TbMsg msg = TbMsg.newMsg()
+                .type(TbMsgType.ENTITY_CREATED)
+                .originator(device.getId())
+                .customerId(customerId)
+                .copyMetaData(metaData)
+                .dataType(TbMsgDataType.JSON)
+                .data(JacksonUtil.toString(entityNode))
+                .build();
+        tbClusterService.pushMsgToRuleEngine(tenantId, device.getId(), msg, null);
+    }
+
+    private TransportApiResponseMsg buildResponse(Device device) throws JsonProcessingException {
+        GetOrCreateDeviceFromGatewayResponseMsg.Builder builder =
+                GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
+                        .setDeviceInfo(ProtoUtils.toDeviceInfoProto(device));
+        DeviceProfile profile =
+                deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
+        if (profile != null) {
+            builder.setDeviceProfile(ProtoUtils.toProto(profile));
+        }
+        return TransportApiResponseMsg.newBuilder()
+                .setGetOrCreateDeviceResponseMsg(builder.build())
+                .build();
+    }
+
+    private TransportApiResponseMsg buildLimitErrorResponse(EntitiesLimitExceededException e,
+                                                            DeviceId gatewayId) {
+        log.warn("[{}][{}] API limit exception: [{}]",
+                e.getTenantId(), gatewayId, e.getMessage());
+        return TransportApiResponseMsg.newBuilder()
+                .setGetOrCreateDeviceResponseMsg(
+                        GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
+                                .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT)
+                )
+                .build();
     }
 
     private TransportApiResponseMsg handle(ProvisionDeviceRequestMsg requestMsg) {

application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java

@@ -88,7 +88,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
             assertNotNull(accessToken);
         }
         if (config.getGatewayName() != null) {
-            savedGateway = createDevice(config.getGatewayName(), deviceProfile.getName(), true);
+            savedGateway = createDevice(config.getGatewayName(), deviceProfile.getName(), !config.isSparkplug);
             DeviceCredentials gatewayCredentials =
                     doGet("/api/device/" + savedGateway.getId().getId().toString() + "/credentials", DeviceCredentials.class);
             assertNotNull(gatewayCredentials);

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java

@@ -31,7 +31,9 @@ import org.junit.Assert;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.TransportPayloadType;
+import org.thingsboard.server.common.data.asset.AssetInfo;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.BooleanDataEntry;
 import org.thingsboard.server.common.data.kv.DoubleDataEntry;
@@ -39,6 +41,7 @@ import org.thingsboard.server.common.data.kv.JsonDataEntry;
 import org.thingsboard.server.common.data.kv.LongDataEntry;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto;
 import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest;
 import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties;
@@ -57,6 +60,7 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_CONNACK;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 import static org.thingsboard.common.util.JacksonUtil.newArrayNode;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Bytes;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int16;
@@ -67,8 +71,13 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataTyp
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt32;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt64;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt8;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_SEPARATOR;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR;
 
 /**
  * Created by nickAS21 on 12.01.23
@@ -82,6 +91,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
     protected ThreadLocalRandom random = ThreadLocalRandom.current();
 
     protected static final String groupId = "SparkplugBGroupId";
+    protected static final String edgeNodeDeviceName = "Test Connect Sparkplug client node";
     protected static final String edgeNode = "SparkpluBNode";
     protected static final String keysBdSeq = "bdSeq";
     protected static final String alias = "Failed Telemetry/Attribute proto sparkplug payload. SparkplugMessageType ";
@@ -98,14 +108,32 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
     protected static final String metricBirthName_Int32 = "Device Metric int32";
     protected Set<String> sparkplugAttributesMetricNames;
 
-    public void beforeSparkplugTest() throws Exception {
+    public void beforeSparkplugTest(boolean isCreateDevices) throws Exception {
         MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder()
-                .gatewayName("Test Connect Sparkplug client node")
+                .gatewayName(edgeNodeDeviceName)
                 .isSparkplug(true)
                 .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames)
                 .transportPayloadType(TransportPayloadType.PROTOBUF)
                 .build();
         processBeforeTest(configProperties);
+        if (isCreateDevices) {
+            // 1. Create the first device with a short name (legacy style)
+            String deviceName1 = deviceId + "_1";
+            Device device1 = createDevice(deviceName1, deviceProfile.getName(), false);
+
+            // 2. Establish 'Created' relation so the transport identifies this gateway as the owner
+            String relationType = "Created";
+            EntityRelation relation1 = createFromRelation(savedGateway, device1, relationType);
+            doPost("/api/relation", relation1).andExpect(status().isOk());
+
+            // 3. Create the second device with a full-path name
+            String deviceName2 = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceId + "_2";
+            Device device2 = createDevice(deviceName2, deviceProfile.getName(), false);
+
+            // 4. Establish 'Created' relation for the second device as well
+            EntityRelation relation2 = createFromRelation(savedGateway, device2, relationType);
+            doPost("/api/relation", relation2).andExpect(status().isOk());
+        }
     }
 
     public void clientWithCorrectNodeAccessTokenWithNDEATH() throws Exception {
@@ -146,7 +174,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         options.setSessionExpiryInterval(0L);
         options.setUserName(gatewayAccessToken);
         String nameSpace = nameSpaceBad.length == 0 ? TOPIC_ROOT_SPB_V_1_0 : nameSpaceBad[0];
-        String topic = nameSpace + "/" + groupId + "/" + SparkplugMessageType.NDEATH.name() + "/" + edgeNode;
+        String topic = nameSpace + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NDEATH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode;
         // The NDEATH message MUST set the MQTT Will QoS to 1 and Retained flag to false
         MqttMessage msg = new MqttMessage();
         msg.setId(0);
@@ -159,38 +187,35 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
     protected List<Device> connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(int cntDevices, long ts) throws Exception {
         List<Device> devices = new ArrayList<>();
         clientWithCorrectNodeAccessTokenWithNDEATH();
-        String keyInt = "Node Metric int32";
+        MetricDataType metricDataType = Int32;
+        String key = "Node Metric int32";
         int valueDeviceInt32 = 1024;
-        SparkplugBProto.Payload.Metric metricInt = createMetric(valueDeviceInt32, ts, keyInt,  Int32, -1L);
+        SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, -1L);
-        String keyStringEmpty = "Node Metric String Empty";
-        String valueDeviceStringEmpty = "";
-        SparkplugBProto.Payload.Metric metricStringEmpty = createMetric(valueDeviceStringEmpty, ts, keyStringEmpty, MetricDataType.String, -1L);
         SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder()
                 .setTimestamp(ts)
                 .setSeq(getBdSeqNum());
-        payloadBirthNode.addMetrics(metricInt);
+        payloadBirthNode.addMetrics(metric);
-        payloadBirthNode.addMetrics(metricStringEmpty);
         payloadBirthNode.setTimestamp(ts);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
 
         valueDeviceInt32 = 4024;
-        metricInt = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L);
+        metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L);
         for (int i = 0; i < cntDevices; i++) {
             SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder()
                     .setTimestamp(ts)
                     .setSeq(getSeqNum());
-            String deviceName = deviceId + "_" + i;
+            String deviceIdName = deviceId + "_" + i;
-
+            String deviceName = groupId +  ":" + edgeNode +  ":" + deviceIdName;
-            payloadBirthDevice.addMetrics(metricInt);
+            payloadBirthDevice.addMetrics(metric);
             if (client.isConnected()) {
-                client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName,
+                client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdName,
                         payloadBirthDevice.build().toByteArray(), 0, false);
                 AtomicReference<Device> device = new AtomicReference<>();
-                await(alias + "find device [" + deviceName + "] after created")
+                await(alias + "find device [" + deviceIdName + "] after created")
-                        .atMost(200, TimeUnit.SECONDS)
+                        .atMost(40, TimeUnit.SECONDS)
                         .ignoreExceptions()
                         .until(() -> {
                             device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class));
@@ -198,13 +223,265 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
                         });
                 devices.add(device.get());
             }
-
         }
 
         Assert.assertEquals(cntDevices, devices.size());
         return devices;
     }
 
+    protected void connectClientWithCorrectAccessTokenWithNDEATHDevicesCreatingBefore_Test(int cntDevices) throws Exception {
+        long ts = calendar.getTimeInMillis();
+        List<Device> devices = new ArrayList<>();
+        clientWithCorrectNodeAccessTokenWithNDEATH();
+        MetricDataType metricDataType = Int32;
+        String key = "Node Metric int32";
+        int valueDeviceInt32 = 1024;
+        SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, -1L);
+        SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(ts)
+                .setSeq(getBdSeqNum());
+        payloadBirthNode.addMetrics(metric);
+        payloadBirthNode.setTimestamp(ts);
+        if (client.isConnected()) {
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode,
+                    payloadBirthNode.build().toByteArray(), 0, false);
+        }
+
+        valueDeviceInt32 = 4024;
+        metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L);
+        // as old device name -> deviceId
+        String deviceIdNameLabel1 = deviceId + "_1";
+
+        if (client.isConnected()) {
+            SparkplugBProto.Payload.Builder payloadBirthDevice1 = SparkplugBProto.Payload.newBuilder()
+                    .setTimestamp(ts)
+                    .setSeq(getSeqNum());
+            payloadBirthDevice1.addMetrics(metric);
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdNameLabel1,
+                    payloadBirthDevice1.build().toByteArray(), 0, false);
+
+        }
+        String deviceName1 = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceIdNameLabel1;
+        AtomicReference<Device> device1 = new AtomicReference<>();
+        await(alias + "find device [" + deviceName1 + "] before connecting")
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    device1.set(doGet("/api/tenant/devices?deviceName=" + deviceName1, Device.class));
+                    return device1.get() != null;
+                });
+        devices.add(device1.get());
+
+        // as new device name ->  groupId +  ":" + edgeNode +  ":" + deviceId;
+        String deviceIdName2 = deviceId + "_2";
+        if (client.isConnected()) {
+            SparkplugBProto.Payload.Builder payloadBirthDevice2 = SparkplugBProto.Payload.newBuilder()
+                    .setTimestamp(ts)
+                    .setSeq(getSeqNum());
+            payloadBirthDevice2.addMetrics(metric);
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdName2,
+                    payloadBirthDevice2.build().toByteArray(), 0, false);
+        }
+        String deviceName2 = groupId + DEVICE_NAME_SPLIT_SEPARATOR + edgeNode + DEVICE_NAME_SPLIT_SEPARATOR + deviceIdName2;
+        AtomicReference<Device> device2 = new AtomicReference<>();
+        await(alias + "find device [" + deviceName2 + "] before connecting")
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    device2.set(doGet("/api/tenant/devices?deviceName=" + deviceName2, Device.class));
+                    return device2.get() != null;
+                });
+        devices.add(device2.get());
+        Assert.assertEquals(cntDevices, devices.size());
+        state_ONLINE_ALL (devices, calendar.getTimeInMillis());
+        // Without full topic: as it was in the old version. When deviceId is updated to full theme, Label is also updated to old deviceId
+        Assert.assertEquals(deviceIdNameLabel1, device1.get().getLabel());
+        // // With a full topic: if new. When creating a device by a client to a full topic, if the Label was not filled in - we do not touch it.
+        Assert.assertNull(device2.get().getLabel());
+    }
+
+    /**
+     * Coverage: Rename when a device with the target full-path name already exists (collision).
+     */
+    protected void renameCollisionWhenTargetNameAlreadyExists_Test() throws Exception {
+        long ts = calendar.getTimeInMillis();
+        String shortName = deviceId + "_1"; // Created in beforeTest
+        String fullPathName = groupId + ":" + edgeNode + ":" + shortName;
+
+        // Manually create a device that already has the "new" full-path name to trigger a collision
+        createDevice(fullPathName, deviceProfile.getName(), false);
+
+        clientWithCorrectNodeAccessTokenWithNDEATH();
+
+        SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(ts)
+                .setSeq(getSeqNum());
+        payload.addMetrics(createMetric(123, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L));
+
+        // Gateway sends DBIRTH for the short name.
+        // Transport will try to rename it but should find a conflict and handle it gracefully.
+        client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + shortName,
+                payload.build().toByteArray(), 0, false);
+
+        await("Checking stability after collision")
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    Device oldDevice = doGet("/api/tenant/devices?deviceName=" + shortName, Device.class);
+                    Device conflictDevice = doGet("/api/tenant/devices?deviceName=" + fullPathName, Device.class);
+                    // Both devices must still exist, proving no exception crashed the process
+                    return oldDevice != null && conflictDevice != null;
+                });
+    }
+
+    /**
+     * Coverage: The privilege concern — attempt to rename a device not owned by the gateway.
+     * This test verifies that the original device's ID remains unchanged, meaning it was not hijacked.
+     */
+    protected void unauthorizedRenameAttemptBad_Test() throws Exception {
+        long ts = calendar.getTimeInMillis();
+        String strangerName = "unauthorized_device_rename";
+
+        // 1. Create a "stranger" device via API (it has no 'Created' relation to the gateway)
+        Device stranger = new Device();
+        stranger.setName(strangerName);
+        stranger.setType("default");
+        doPost("/api/device", stranger);
+        final DeviceId originalStrangerId = stranger.getId();
+
+        clientWithCorrectNodeAccessTokenWithNDEATH();
+
+        SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(ts).setSeq(getSeqNum());
+
+        // 2. Unauthorized gateway attempts to rename this device via Sparkplug topic path
+        client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + strangerName,
+                payload.build().toByteArray(), 0, false);
+
+        String expectedFullPath = groupId + ":" + edgeNode + ":" + strangerName;
+
+        // 3. Verify security: the original device must still be linked to its short name with the same ID
+        await("Verify original device was not hijacked")
+                .atMost(40, TimeUnit.SECONDS)
+                .pollDelay(2, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    // Check if the original device still exists with its original ID
+                    Device currentStranger = doGet("/api/tenant/devices?deviceName=" + strangerName, Device.class);
+                    Assert.assertNotNull("Original device disappeared!", currentStranger);
+                    Assert.assertEquals("Security breach: Original device ID changed!", originalStrangerId, currentStranger.getId());
+
+                    // Even if the gateway created a NEW device with a full path, it must have a different ID
+                    Device newDevice = doGet("/api/tenant/devices?deviceName=" + expectedFullPath, Device.class);
+                    if (newDevice != null) {
+                        Assert.assertNotEquals("Stranger device was successfully hijacked (IDs match)!", originalStrangerId, newDevice.getId());
+                    }
+                });
+    }
+
+    /**
+     * Coverage: The privilege concern — attempt to rename a device not owned by the gateway.
+     */
+    protected void unauthorizedRenameAttempt_Test() throws Exception {
+        long ts = calendar.getTimeInMillis();
+        String strangerName = "unauthorized_device_rename";
+
+        // Create a device without a "Created" relation to the gateway
+        Device stranger = new Device();
+        stranger.setName(strangerName);
+        stranger.setType("default");
+        doPost("/api/device", stranger);
+
+        clientWithCorrectNodeAccessTokenWithNDEATH();
+
+        SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(ts).setSeq(getSeqNum());
+
+        // Unauthorized gateway attempts to rename the device via Sparkplug topic
+        client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + strangerName,
+                payload.build().toByteArray(), 0, false);
+
+        String expectedFullPath = groupId + ":" + edgeNode + ":" + strangerName;
+        await().atMost(30, TimeUnit.SECONDS).untilAsserted(() ->
+                doGet("/api/tenant/devices?deviceName=" + expectedFullPath, Device.class, status().isNotFound())
+        );
+    }
+
+    protected void state_ONLINE_ALL (List<Device> devices, long ts) {
+        TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), ONLINE.name()));
+        await(alias + messageName(STATE) + ", device: " + savedGateway.getName())
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    var foundEntry = tsService.findAllLatest(tenantId, savedGateway.getId()).get().stream()
+                            .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey()))
+                            .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue()))
+                            .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs())
+                            .findFirst();
+                    return foundEntry.isPresent();
+                });
+
+        for (Device device : devices) {
+            await(alias + messageName(STATE) + ", device: " + device.getName())
+                    .atMost(40, TimeUnit.SECONDS)
+                    .until(() -> {
+                        var foundEntry = tsService.findAllLatest(tenantId, device.getId()).get().stream()
+                                .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey()))
+                                .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue()))
+                                .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs())
+                                .findFirst();
+                        return foundEntry.isPresent();
+                    });
+        }
+    }
+
+    /**
+     * Coverage: Concurrent first-message registration with the lock mechanism.
+     */
+    protected void concurrentFirstMessageRegistration_Test() throws Exception {
+        int threadCount = 5;
+        String concurrentDeviceName = "concurrent_device";
+        clientWithCorrectNodeAccessTokenWithNDEATH();
+
+        java.util.concurrent.ExecutorService executor = java.util.concurrent.Executors.newFixedThreadPool(threadCount);
+        long ts = calendar.getTimeInMillis();
+
+        for (int i = 0; i < threadCount; i++) {
+            executor.submit(() -> {
+                try {
+                    SparkplugBProto.Payload.Builder payload = SparkplugBProto.Payload.newBuilder()
+                            .setTimestamp(ts).setSeq(0);
+                    client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/DBIRTH/" + edgeNode + "/" + concurrentDeviceName,
+                            payload.build().toByteArray(), 0, false);
+                } catch (Exception e) {
+                    log.error("Concurrent publish failed", e);
+                }
+            });
+        }
+
+        String expectedName = groupId + ":" + edgeNode + ":" + concurrentDeviceName;
+        await("Wait for concurrent registration result")
+                .atMost(40, TimeUnit.SECONDS) // Restored to 40s as requested
+                .until(() -> doGet("/api/tenant/devices?deviceName=" + expectedName, Device.class) != null);
+
+        executor.shutdown();
+    }
+
+    /**
+     * Coverage: Sparkplug-message handling when msgId <= 0 (#7).
+     * Verifies that the transport does not close the session for Sparkplug clients using msgId 0.
+     */
+    protected void sparkplugSessionStaysAliveWithZeroMsgId_Test() throws Exception {
+        // clientMqttV5ConnectWithNDEATH internally sets msgId = 0 for the Will message.
+        // This validates that the connection is accepted despite msgId being 0.
+        IMqttToken connectionResult = clientMqttV5ConnectWithNDEATH(calendar.getTimeInMillis(), 0, -1L);
+        Assert.assertTrue("Sparkplug connection should be successful with msgId=0", client.isConnected());
+
+        // Publish NBIRTH message which usually goes through the aggregate callback.
+        // This verifies that msgId=0 in the callback does not trigger closeDeviceSession.
+        connectionWithNBirth(Int32, "test_metric_msgId_0", 555);
+
+        // Awaitility to ensure the session remains open after processing.
+        await("Verify Sparkplug session remains open after receiving msgId=0")
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> client.isConnected());
+    }
+
     protected List<Device> connectClientWithCorrectAccessTokenWithNDEATHWithAliasCreatedDevices(long ts) throws Exception {
         List<Device> devices = new ArrayList<>();
         Long alias = 0L;
@@ -219,7 +496,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         payloadBirthNode.addMetrics(metric);
         payloadBirthNode.setTimestamp(ts);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
 
@@ -228,15 +505,16 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder()
                 .setTimestamp(ts)
                 .setSeq(getSeqNum());
-        String deviceName = deviceId + "_" + 1;
+        String deviceIdName = deviceId + "_1";
+        String deviceName = groupId +  ":" + edgeNode +  ":" + deviceIdName;
 
         payloadBirthDevice.addMetrics(metric);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + deviceIdName,
                     payloadBirthDevice.build().toByteArray(), 0, false);
             AtomicReference<Device> device = new AtomicReference<>();
             await(alias + "find device [" + deviceName + "] after created")
-                    .atMost(200, TimeUnit.SECONDS)
+                    .atMost(40, TimeUnit.SECONDS)
                     .ignoreExceptions()
                     .until(() -> {
                         device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class));
@@ -282,7 +560,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         listKeys.add(metricKey);
 
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
         return listKeys;
@@ -525,4 +803,9 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         }
     }
 
+    private EntityRelation createFromRelation(Device mainDevice, Device device, String relationType) {
+        return new EntityRelation(mainDevice.getId(), device.getId(), relationType);
+    }
+
+
 }

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesInProfileTest.java

@@ -33,7 +33,7 @@ public class MqttV5ClientSparkplugBAttributesInProfileTest extends AbstractMqttV
     public void beforeTest() throws Exception {
         sparkplugAttributesMetricNames = new HashSet<>();
         sparkplugAttributesMetricNames.add(metricBirthName_Int32);
-        beforeSparkplugTest();
+        beforeSparkplugTest(false);
     }
 
     @After

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/attributes/MqttV5ClientSparkplugBAttributesTest.java

@@ -29,7 +29,7 @@ public class MqttV5ClientSparkplugBAttributesTest extends AbstractMqttV5ClientSp
 
     @Before
     public void beforeTest() throws Exception {
-        beforeSparkplugTest();
+        beforeSparkplugTest(false);
     }
 
     @After

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java

@@ -37,10 +37,10 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.OFFLINE;
-import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR;
 
 /**
  * Created by nickAS21 on 12.01.23
@@ -95,31 +95,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
     protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_ALL(int cntDevices) throws Exception {
         long ts = calendar.getTimeInMillis();
         List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts);
-
+        state_ONLINE_ALL (devices, ts);
-        TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), ONLINE.name()));
-        await(alias + messageName(STATE) + ", device: " + savedGateway.getName())
-                .atMost(40, TimeUnit.SECONDS)
-                .until(() -> {
-                    var foundEntry = tsService.findAllLatest(tenantId, savedGateway.getId()).get().stream()
-                            .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey()))
-                            .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue()))
-                            .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs())
-                            .findFirst();
-                    return foundEntry.isPresent();
-                });
-
-        for (Device device : devices) {
-            await(alias + messageName(STATE) + ", device: " + device.getName())
-                    .atMost(40, TimeUnit.SECONDS)
-                    .until(() -> {
-                        var foundEntry = tsService.findAllLatest(tenantId, device.getId()).get().stream()
-                                .filter(tsKv -> tsKv.getKey().equals(tsKvEntry.getKey()))
-                                .filter(tsKv -> tsKv.getValue().equals(tsKvEntry.getValue()))
-                                .filter(tsKv -> tsKv.getTs() == tsKvEntry.getTs())
-                                .findFirst();
-                        return foundEntry.isPresent();
-                    });
-        }
     }
 
     protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OneDeviceOFFLINE(int cntDevices, int indexDeviceDisconnect) throws Exception {
@@ -135,7 +111,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
         if (client.isConnected()) {
             List<Device> devicesList = new ArrayList<>(devices);
             Device device =  devicesList.get(indexDeviceDisconnect);
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DDEATH.name() + "/" + edgeNode + "/" + device.getName(),
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + SparkplugMessageType.DDEATH.name() + TOPIC_SPLIT_SEPARATOR + edgeNode + TOPIC_SPLIT_SEPARATOR + device.getName(),
                     payloadDeathDevice.build().toByteArray(), 0, false);
             await(alias + messageName(STATE) + ", device: " + device.getName())
                     .atMost(40, TimeUnit.SECONDS)

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest.java

@@ -0,0 +1,71 @@
+/**
+ * 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.mqtt.sparkplug.connection;
+
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.thingsboard.server.dao.service.DaoSqlTest;
+
+/**
+ * Created by nickAS21 on 16.02.26
+ */
+@DaoSqlTest
+public class MqttV5ClientSparkplugBConnectionDevicesCreatingBeforeTest extends AbstractMqttV5ClientSparkplugConnectionTest {
+
+    /**
+     * String deviceName_1 = deviceId + "_1"; Only name device. Without a complete topic: how it was in the old version.
+     * String deviceName_2 = groupId + DEVICE_NAME_SPLIT_REGEXP + edgeNode + DEVICE_NAME_SPLIT_REGEXP + deviceId + "_2"; With complete topic: how it was in the new version.
+     * @throws Exception
+     */
+    @Before
+    public void beforeTest() throws Exception {
+        beforeSparkplugTest(true);
+    }
+
+    @After
+    public void afterTest() throws MqttException {
+        if (client.isConnected()) {
+            client.disconnect();
+        }
+    }
+
+    @Test
+    public void testClientWithCorrectAccessTokenWithNDEATHTwoDevicesCreatingBeforeFirstNameDeviceIdSecondNameFull() throws Exception {
+        connectClientWithCorrectAccessTokenWithNDEATHDevicesCreatingBefore_Test(2);
+    }
+
+    @Test
+    public void testRenameWhenDeviceFullPathAlreadyExists_Collision() throws Exception {
+        renameCollisionWhenTargetNameAlreadyExists_Test();
+    }
+
+    @Test
+    public void testUnauthorizedRenameAttempt() throws Exception {
+        unauthorizedRenameAttempt_Test();
+    }
+
+    @Test
+    public void testConcurrentFirstMessageRegistration() throws Exception {
+        concurrentFirstMessageRegistration_Test();
+    }
+
+    @Test
+    public void testSparkplugSessionStaysAliveWithZeroMsgId() throws Exception {
+        sparkplugSessionStaysAliveWithZeroMsgId_Test();
+    }
+}

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java

@@ -29,7 +29,7 @@ public class MqttV5ClientSparkplugBConnectionTest extends AbstractMqttV5ClientSp
 
     @Before
     public void beforeTest() throws Exception {
-        beforeSparkplugTest();
+        beforeSparkplugTest(false);
     }
 
     @After

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java

@@ -28,7 +28,7 @@ public class MqttV5RpcSparkplugTest  extends AbstractMqttV5RpcSparkplugTest {
 
     @Before
     public void beforeTest() throws Exception {
-        beforeSparkplugTest();
+        beforeSparkplugTest(false);
     }
 
     @After
@@ -47,6 +47,7 @@ public class MqttV5RpcSparkplugTest  extends AbstractMqttV5RpcSparkplugTest {
     public void testClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception {
         processClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success();
     }
+
     @Test
     public void testClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success() throws Exception {
         processClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success();

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java

@@ -30,6 +30,7 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR;
 
 /**
  * Created by nickAS21 on 12.01.23
@@ -67,7 +68,7 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac
         createdAddMetricValuePrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts);
 
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + messageTypeName + TOPIC_SPLIT_SEPARATOR + edgeNode,
                     ndataPayload.build().toByteArray(), 0, false);
         }
 
@@ -96,7 +97,7 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac
         createdAddMetricValueArraysPrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts);
 
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + groupId + TOPIC_SPLIT_SEPARATOR + messageTypeName + TOPIC_SPLIT_SEPARATOR + edgeNode,
                     ndataPayload.build().toByteArray(), 0, false);
         }
 

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java

@@ -29,7 +29,7 @@ public class MqttV5ClientSparkplugBTelemetryTest extends AbstractMqttV5ClientSpa
 
     @Before
     public void beforeTest() throws Exception {
-        beforeSparkplugTest();
+        beforeSparkplugTest(false);
     }
 
     @After

common/proto/src/main/proto/queue.proto

@@ -470,6 +470,7 @@ message GetOrCreateDeviceFromGatewayRequestMsg {
   int64 gatewayIdLSB = 2;
   string deviceName = 3;
   string deviceType = 4;
+  bool isSparkplug = 5;
 }
 
 message GetOrCreateDeviceFromGatewayResponseMsg {

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java

@@ -68,6 +68,7 @@ import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor;
 import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService;
 import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopic;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -260,7 +261,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
         log.trace("[{}][{}][{}] onDeviceConnect: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName);
         int msgId = getMsgId(msg);
         AtomicBoolean ackSent = new AtomicBoolean(false);
-        process(onDeviceConnect(deviceName, deviceType),
+        process(onDeviceConnect(deviceName, deviceType, false),
                 result -> {
                     ack(msg, MqttReasonCodes.PubAck.SUCCESS);
                     log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName);
@@ -276,7 +277,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
         }
     }
 
-    ListenableFuture<T> onDeviceConnect(String deviceName, String deviceType) {
+    ListenableFuture<T> onDeviceConnect(String deviceName, String deviceType, boolean isSparkplug) {
         T result = devices.get(deviceName);
         if (result == null) {
             Lock deviceCreationLock = deviceCreationLockMap.computeIfAbsent(deviceName, s -> new ReentrantLock());
@@ -284,7 +285,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
             try {
                 result = devices.get(deviceName);
                 if (result == null) {
-                    return getDeviceCreationFuture(deviceName, deviceType);
+                    return getDeviceCreationFuture(deviceName, deviceType, isSparkplug);
                 } else {
                     return Futures.immediateFuture(result);
                 }
@@ -296,7 +297,27 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
         }
     }
 
-    private ListenableFuture<T> getDeviceCreationFuture(String deviceName, String deviceType) {
+    ListenableFuture<T> onDeviceConnectSparkplug(SparkplugTopic topic, String deviceType) {
+        String fullPath = topic.getNodeDeviceNameAllPath();
+        // Primary lookup: try to find the device by its full-path name (standard for new devices)
+        T result = devices.get(fullPath);
+
+        if (result == null) {
+            // Secondary lookup (Legacy Fallback): check for the short name if full path is not found.
+            // This supports devices migrated from older versions.
+            String shortName = topic.getNodeDeviceName();
+            result = devices.get(shortName);
+        }
+
+        if (result != null) {
+            return Futures.immediateFuture(result);
+        } else {
+            // If not found in cache at all, proceed with connection/creation using full path
+            return onDeviceConnect(fullPath, deviceType, true);
+        }
+    }
+
+    private ListenableFuture<T> getDeviceCreationFuture(String deviceName, String deviceType, boolean isSparkplug) {
         final SettableFuture<T> futureToSet = SettableFuture.create();
         ListenableFuture<T> future = deviceFutures.putIfAbsent(deviceName, futureToSet);
         if (future != null) {
@@ -309,6 +330,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
                             .setDeviceType(deviceType)
                             .setGatewayIdMSB(gateway.getDeviceId().getId().getMostSignificantBits())
                             .setGatewayIdLSB(gateway.getDeviceId().getId().getLeastSignificantBits())
+                            .setIsSparkplug(isSparkplug)
                             .build(),
                     new TransportServiceCallback<>() {
                         @Override
@@ -844,8 +866,16 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
                                 deviceSessionCtx, msgId, MqttReasonCodes.PubAck.UNSPECIFIED_ERROR.byteValue()));
                     }
                 }
+                // Check for malformed packets (msgId <= 0)
                 if (msgId <= 0) {
-                    closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MALFORMED_PACKET);
+                    if (deviceSessionCtx.isSparkplug()) {
+                        // Sparkplug devices may use msgId = 0 during the initial connection or for QoS 0 messages.
+                        // We bypass the MALFORMED_PACKET check to allow these sessions to proceed.
+                        log.trace("[{}] Sparkplug session: allowed msgId [{}] for device: [{}]", sessionId, msgId, deviceName);
+                    } else {
+                        // Standard MQTT defense: close session for invalid message IDs
+                        closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MALFORMED_PACKET);
+                    }
                 }
             }
 
@@ -872,7 +902,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
     }
 
     protected void process(String deviceName, Consumer<T> onSuccess, Consumer<Throwable> onFailure) {
-        ListenableFuture<T> deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE);
+        ListenableFuture<T> deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE, false);
         process(deviceCtxFuture, onSuccess, onFailure);
     }
 

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java

@@ -128,5 +128,5 @@ public class SparkplugDeviceSessionContext extends AbstractGatewayDeviceSessionC
                             rpcRequest.getMethodName() + ". " + e.getMessage());
         }
     }
-
+    
 }

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java

@@ -21,7 +21,6 @@ import com.google.common.util.concurrent.MoreExecutors;
 import com.google.gson.JsonSyntaxException;
 import io.netty.handler.codec.mqtt.MqttMessage;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import io.netty.handler.codec.mqtt.MqttReasonCodes;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -61,8 +60,8 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetr
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.fromSparkplugBMetricToKeyValueProto;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.validatedValueByTypeMetric;
-import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR;
-import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_STATE_REGEXP;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_STATE_SEPARATOR;
 
 @Slf4j
 @SpecVersion(spec = "sparkplug", version = "3.0.0")
@@ -106,9 +105,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
     }
 
     public void onAttributesTelemetryProto(int msgId, SparkplugBProto.Payload sparkplugBProto, SparkplugTopic topic) throws AdaptorException, ThingsboardException {
-        String deviceName = topic.getNodeDeviceName();
+        String deviceName = checkDeviceName(this.deviceSessionCtx.getDeviceInfo().getDeviceName());
-        checkDeviceName(deviceName);
-
         ListenableFuture<MqttDeviceAwareSessionContext> contextListenableFuture;
         if (topic.isNode()) {
             if (topic.isType(NBIRTH)) {
@@ -118,19 +115,26 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
             }
             contextListenableFuture = Futures.immediateFuture(this.deviceSessionCtx);
         } else {
-            ListenableFuture<SparkplugDeviceSessionContext> deviceCtx = onDeviceConnectProto(topic);
+            try {
-            contextListenableFuture = Futures.transform(deviceCtx, ctx -> {
+                deviceName = checkDeviceName(topic.getNodeDeviceNameAllPath());
-                if (topic.isType(DBIRTH)) {
+                ListenableFuture<SparkplugDeviceSessionContext> deviceCtx = this.onDeviceConnectProto(topic);
-                    sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), deviceName, ONLINE,
+                String finalDeviceName = deviceName;
-                            sparkplugBProto.getTimestamp());
+                contextListenableFuture = Futures.transform(deviceCtx, ctx -> {
-                    try {
+                    if (topic.isType(DBIRTH)) {
-                        ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
+                        sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), finalDeviceName, ONLINE,
-                    } catch (IllegalArgumentException | DuplicateKeyException e) {
+                                sparkplugBProto.getTimestamp());
+                        try {
+                            ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
+                        } catch (IllegalArgumentException | DuplicateKeyException e) {
+                            log.error("[{}] Failed to set birth metrics", finalDeviceName, e);
                             throw new RuntimeException(e);
+                        }
                     }
-                }
+                    return ctx;
-                return ctx;
+                }, MoreExecutors.directExecutor());
-            }, MoreExecutors.directExecutor());
+            }  catch (IllegalArgumentException | DuplicateKeyException e) {
+                throw new RuntimeException(e);
+            }
         }
         Set<String> attributesMetricNames = ((MqttDeviceProfileTransportConfiguration) deviceSessionCtx
                 .getDeviceProfile().getProfileData().getTransportConfiguration()).getSparkplugAttributesMetricNames();
@@ -200,7 +204,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
      */
     public void handleSparkplugSubscribeMsg(MqttTopicSubscription subscription) throws ThingsboardException {
         String topic = subscription.topicFilter();
-        if (topic != null && topic.startsWith(TOPIC_STATE_REGEXP)) {
+        if (topic != null && topic.startsWith(TOPIC_STATE_SEPARATOR)) {
             log.trace("Subscribing on it’s own spBv1.0/STATE/[the Sparkplug Host Application] - Implemented as status via checkSparkplugNodeSession");
         } else if (this.validateTopicDataSubscribe(topic)) {
             // TODO if need subscription DATA
@@ -222,7 +226,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
             ThingsboardException {
         try {
             String deviceType = this.gateway.getDeviceType() + " device";
-            return onDeviceConnect(topic.getNodeDeviceName(), deviceType);
+            return onDeviceConnectSparkplug(topic, deviceType);
         } catch (RuntimeException e) {
             log.error("Failed Sparkplug Device connect proto!", e);
             throw new ThingsboardException(e, ThingsboardErrorCode.BAD_REQUEST_PARAMS);
@@ -384,7 +388,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
      * @throws ThingsboardException if an error occurs while parsing
      */
     public boolean validateTopicDataSubscribe(String topic) throws ThingsboardException {
-        String[] splitTopic = topic.split(TOPIC_SPLIT_REGEXP);
+        String[] splitTopic = topic.split(TOPIC_SPLIT_SEPARATOR);
         if (splitTopic.length >= 4 && splitTopic.length <= 5 &&
                 splitTopic[0].equals(this.sparkplugTopicNode.getNamespace()) &&
                 splitTopic[1].equals(this.sparkplugTopicNode.getGroupId()) &&

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java

@@ -21,8 +21,9 @@ import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.parseMessageType;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_SEPARATOR;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
-import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_SEPARATOR;
 
 /**
  * Created by nickAS21 on 12.12.22
@@ -196,7 +197,7 @@ public class SparkplugTopic {
         try {
             if (isValidIdElementToUTF8(topicString)) {
                 SparkplugMessageType messageType;
-                String[] splitTopic = topicString.split(TOPIC_SPLIT_REGEXP);
+                String[] splitTopic = topicString.split(TOPIC_SPLIT_SEPARATOR);
                 if (TOPIC_ROOT_SPB_V_1_0.equals(splitTopic[0])) {
                     if (splitTopic.length == 3) {
                         messageType = parseMessageType(splitTopic[1]);
@@ -329,6 +330,17 @@ public class SparkplugTopic {
         return isNode() ? edgeNodeId : deviceId;
     }
 
+    public String getNodeDeviceNameAllPath() {
+        StringBuilder sb = new StringBuilder();
+        if (hostApplicationId == null) {
+            sb.append(getGroupId()).append(DEVICE_NAME_SPLIT_SEPARATOR).append(getEdgeNodeId());
+            if (getDeviceId() != null) {
+                sb.append(DEVICE_NAME_SPLIT_SEPARATOR).append(getDeviceId());
+            }
+        }
+        return sb.toString();
+    }
+
     public static boolean isValidIdElementToUTF8(String deviceIdElement) {
         if (deviceIdElement == null) {
             return false;

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java

@@ -34,8 +34,9 @@ public class SparkplugTopicService {
     private static final Map<String, SparkplugTopic> SPLIT_TOPIC_CACHE = new HashMap<>();
     public static final String TOPIC_ROOT_SPB_V_1_0 = "spBv1.0";
     public static final String TOPIC_ROOT_CERT_SP = "$sparkplug/certificates/";
-    public static final String TOPIC_SPLIT_REGEXP = "/";
+    public static final String TOPIC_SPLIT_SEPARATOR = "/";
-    public static final String TOPIC_STATE_REGEXP = TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + STATE.name() + TOPIC_SPLIT_REGEXP;
+    public static final String DEVICE_NAME_SPLIT_SEPARATOR = ":";
+    public static final String TOPIC_STATE_SEPARATOR = TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_SEPARATOR + STATE.name() + TOPIC_SPLIT_SEPARATOR;
 
     public static SparkplugTopic getSplitTopic(String topic) throws ThingsboardException {
         SparkplugTopic sparkplugTopic = SPLIT_TOPIC_CACHE.get(topic);