Merge pull request #16071 from thingsboard/fix_bug_sparkplug_alias

Sparkplug: fixed empty telemetry metric name when using alias

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

@@ -288,6 +288,33 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         return listKeys;
     }
 
+    protected List<String> connectionWithNBirthMetricNameAndAlias(MetricDataType metricDataType, String metricKey, Object metricValue, Long alias) throws Exception {
+        List<String> listKeys = new ArrayList<>();
+        SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(calendar.getTimeInMillis());
+        long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS;
+        long valueBdSec = getBdSeqNum();
+        payloadBirthNode.addMetrics(createMetric(valueBdSec, ts, keysBdSeq, Int64, -1L));
+        listKeys.add(SparkplugMessageType.NBIRTH.name() + " " + keysBdSeq);
+        payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean, -1L));
+        listKeys.add(keyNodeRebirth);
+
+        payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType, alias));
+
+        listKeys.add(metricKey);
+
+        if (client.isConnected()) {
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+                    payloadBirthNode.build().toByteArray(), 0, false);
+        }
+        return listKeys;
+    }
+
+    protected void createdAddMetricValueWithAliasTsKv(SparkplugBProto.Payload.Builder dataPayload, Object value, MetricDataType metricDataType,
+                                                      long ts) throws ThingsboardException {
+        dataPayload.addMetrics(createMetric(value, ts, null, metricDataType, 4L));
+    }
+
     protected void createdAddMetricValuePrimitiveTsKv(List<TsKvEntry> listTsKvEntry, List<String> listKeys,
                                                       SparkplugBProto.Payload.Builder dataPayload, long ts) throws ThingsboardException {
 

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

@@ -18,9 +18,12 @@ package org.thingsboard.server.transport.mqtt.sparkplug.timeseries;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.Assert;
+import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
+import org.thingsboard.server.common.data.kv.LongDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto;
 import org.thingsboard.server.transport.mqtt.sparkplug.AbstractMqttV5ClientSparkplugTest;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType;
 import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType;
 
 import java.util.ArrayList;
@@ -51,6 +54,52 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac
         Assert.assertEquals(listKeys.size(), finalFuture.get().get().size());
     }
 
+    protected void processClientWithCorrectAccessTokenPublishNBIRTH_NDATA_AliasWithoutMetricName() throws Exception {
+        clientWithCorrectNodeAccessTokenWithNDEATH();
+        String metricBirthName_Alias = "Node Metric int32 Only Alias";
+        MetricDataType metricBirthDataType = metricBirthDataType_Int32;
+        List<String> listKeys = connectionWithNBirthMetricNameAndAlias(metricBirthDataType, metricBirthName_Alias, nextInt32(), 4L);
+        Assert.assertTrue("Connection node is failed", client.isConnected());
+        AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFuture = new AtomicReference<>();
+        await(alias + SparkplugMessageType.NBIRTH.name())
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    finalFuture.set(tsService.findLatest(tenantId, savedGateway.getId(), listKeys));
+                    return !finalFuture.get().get().isEmpty();
+                });
+        Assert.assertEquals(listKeys.size(), finalFuture.get().get().size());
+
+        String messageTypeName = SparkplugMessageType.NDATA.name();
+
+        List<TsKvEntry> listTsKvEntry = new ArrayList<>();
+
+        SparkplugBProto.Payload.Builder ndataPayload = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(calendar.getTimeInMillis())
+                .setSeq(getSeqNum());
+        long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS;
+        int valueKey = nextInt32();
+
+        TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(metricBirthName_Alias, Long.valueOf(String.valueOf(valueKey))));
+        listTsKvEntry.add(tsKvEntry);
+        createdAddMetricValueWithAliasTsKv(ndataPayload, valueKey, metricBirthDataType, ts);
+
+        if (client.isConnected()) {
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode,
+                    ndataPayload.build().toByteArray(), 0, false);
+        }
+
+        AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFutureAlias = new AtomicReference<>();
+        await(alias + SparkplugMessageType.NDATA.name())
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    finalFutureAlias.set(tsService.findAllLatest(tenantId, savedGateway.getId()));
+                    return finalFutureAlias.get().get().size() == (listKeys.size() + listTsKvEntry.size() + 1);
+                });
+        Assert.assertTrue("Actual tsKvEntries is not containsAll Expected tsKvEntries", containsIgnoreVersion(finalFutureAlias .get().get(), listTsKvEntry));
+
+
+    }
+
     protected void processClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple() throws Exception {
         List<String> listKeys = new ArrayList<>();
         clientWithCorrectNodeAccessTokenWithNDEATH();

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

@@ -43,6 +43,11 @@ public class MqttV5ClientSparkplugBTelemetryTest extends AbstractMqttV5ClientSpa
         processClientWithCorrectAccessTokenPublishNBIRTH();
     }
 
+    @Test
+    public void testClientWithCorrectAccessTokenPublishNBIRTH_NDATA_AliasWithoutMetricName() throws Exception {
+        processClientWithCorrectAccessTokenPublishNBIRTH_NDATA_AliasWithoutMetricName();
+    }
+
     @Test
     public void testClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple() throws Exception {
         processClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple();

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

@@ -257,8 +257,8 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
                 } else {
                     if (attributesMetricNames == null || !matches(attributesMetricNames, metricName)) {
                         long ts = protoMetric.getTimestamp();
-                        String key = SPARKPLUG_BD_SEQUENCE_NUMBER_KEY.equals(protoMetric.getName()) ?
-                                topicTypeName + " " + protoMetric.getName() : protoMetric.getName();
+                        String key = SPARKPLUG_BD_SEQUENCE_NUMBER_KEY.equals(metricName) ?
+                                topicTypeName + " " + metricName : metricName;
                         Optional<TransportProtos.KeyValueProto> keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric);
                         keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts)));
                     }