(iotdb) branch master updated: Fix MQTT measurement validation (#18215)

[haonan]

This is an automated email from the ASF dual-hosted git repository.

HTHou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new dfbe8db321d Fix MQTT measurement validation (#18215)
dfbe8db321d is described below

commit dfbe8db321d52ebe7409ecd88b94633d0c75ca22
Author: Jiang Tian <jt2594...@163.com>
AuthorDate: Wed Jul 15 19:00:58 2026 +0800

    Fix MQTT measurement validation (#18215)
---
 .../org/apache/iotdb/mqtt/MPPPublishHandler.java   |  5 ++-
 .../iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java   | 41 ++++++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)

diff --git 
a/external-service-impl/mqtt/src/main/java/org/apache/iotdb/mqtt/MPPPublishHandler.java
 
b/external-service-impl/mqtt/src/main/java/org/apache/iotdb/mqtt/MPPPublishHandler.java
index 72bf4c12600..a99c875fa23 100644
--- 
a/external-service-impl/mqtt/src/main/java/org/apache/iotdb/mqtt/MPPPublishHandler.java
+++ 
b/external-service-impl/mqtt/src/main/java/org/apache/iotdb/mqtt/MPPPublishHandler.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.queryengine.common.SqlDialect;
 import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils;
 import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.db.auth.AuthorityChecker;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -274,7 +275,9 @@ public class MPPPublishHandler extends 
AbstractInterceptHandler {
           
DataNodeDevicePathCache.getInstance().getPartialPath(message.getDevice()));
       TimestampPrecisionUtils.checkTimestampPrecision(message.getTimestamp());
       statement.setTime(message.getTimestamp());
-      statement.setMeasurements(message.getMeasurements().toArray(new 
String[0]));
+      statement.setMeasurements(
+          
PathUtils.checkIsLegalSingleMeasurementsAndUpdate(message.getMeasurements())
+              .toArray(new String[0]));
       if (message.getDataTypes() == null) {
         statement.setDataTypes(new 
TSDataType[message.getMeasurements().size()]);
         statement.setValues(message.getValues().toArray(new Object[0]));
diff --git 
a/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
 
b/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
index 0b52759ad27..8d372c1d810 100644
--- 
a/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
+++ 
b/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
@@ -51,6 +51,7 @@ import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -481,4 +482,44 @@ public class IoTDBMQTTServiceJsonIT {
       }
     }
   }
+
+  @Test
+  public void testIllegalMeasurementIsRejected() throws Exception {
+    try (final ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      String payload =
+          "["
+              + 
"{\"device\":\"root.sg.invalid\",\"timestamp\":1,\"measurements\":[\"a.b\"],\"values\":[1]},"
+              + 
"{\"device\":\"root.sg.invalid\",\"timestamp\":2,\"measurements\":[\"valid\"],\"values\":[2]}"
+              + "]";
+
+      Awaitility.await()
+          .atMost(3, TimeUnit.MINUTES)
+          .pollInterval(1, TimeUnit.SECONDS)
+          .until(
+              () -> {
+                connection.publish("root.sg.invalid", payload.getBytes(), 
QoS.AT_LEAST_ONCE, false);
+                try (final SessionDataSet dataSet =
+                    session.executeQueryStatement(
+                        "select valid from root.sg.invalid where time = 2")) {
+                  return dataSet.hasNext();
+                } catch (StatementExecutionException e) {
+                  if (e.getMessage() != null && e.getMessage().contains("does 
not exist")) {
+                    return false;
+                  }
+                  throw e;
+                }
+              });
+
+      int timeseriesCount = 0;
+      try (final SessionDataSet dataSet =
+          session.executeQueryStatement("show timeseries root.sg.invalid.**")) 
{
+        while (dataSet.hasNext()) {
+          dataSet.next();
+          timeseriesCount++;
+        }
+      }
+      assertEquals(1, timeseriesCount);
+      assertFalse(session.checkTimeseriesExists("root.sg.invalid.a.b"));
+    }
+  }
 }