Fix MQTT measurement validation (#18215)

Author: jt2594838

external-service-impl/mqtt/src/main/java/org/apache/iotdb/mqtt/MPPPublishHandler.java

@@ -24,6 +24,7 @@
 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 @@ private void insertTree(TreeMessage message, MqttClientSession session) {
           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]));

integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java

@@ -51,6 +51,7 @@
 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 void testBatchJsonWithVariousValues() throws Exception {
       }
     }
   }
+
+  @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"));
+    }
+  }
 }