This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new f36974d6421 [To dev/1.3] Cherry-Pick:"Enable the kernel to support
passing topics to MQTT custom plugins" and "Disable the parameter
'ALLOW_ZERO-BYTE_CITE_SIDED_PROPERTY1 NAME' and fix null pointer exception"
(#15985)
f36974d6421 is described below
commit f36974d64216deca2366c56dc0d1c6c8c68ede0a
Author: wenyanshi-123 <[email protected]>
AuthorDate: Tue Jul 22 09:56:53 2025 +0800
[To dev/1.3] Cherry-Pick:"Enable the kernel to support passing topics to
MQTT custom plugins" and "Disable the parameter
'ALLOW_ZERO-BYTE_CITE_SIDED_PROPERTY1 NAME' and fix null pointer exception"
(#15985)
* cp-Enable the kernel to support passing topics to MQTT custom plugins
(#15523)
* cp-Disable the parameter 'ALLOW_ZERO-BYTE_CITE_SIDED_PROPERTY1 NAME' and
fix null pointer exception (#15866)
Co-authored-by: shiwenyan <[email protected]>
---------
Co-authored-by: shiwenyan <[email protected]>
---
.../iotdb/mqtt/server/CustomizedJsonPayloadFormatter.java | 9 ++++++++-
.../apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java | 9 ++++++++-
.../apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java | 6 +++++-
.../org/apache/iotdb/db/protocol/mqtt/PayloadFormatter.java | 12 ++++++++++++
.../main/java/org/apache/iotdb/db/service/MQTTService.java | 2 +-
.../iotdb/db/protocol/mqtt/JSONPayloadFormatterTest.java | 13 ++++++++-----
6 files changed, 42 insertions(+), 9 deletions(-)
diff --git
a/example/mqtt-customize/src/main/java/org/apache/iotdb/mqtt/server/CustomizedJsonPayloadFormatter.java
b/example/mqtt-customize/src/main/java/org/apache/iotdb/mqtt/server/CustomizedJsonPayloadFormatter.java
index 5b708f1bcd5..b0a7adae9ae 100644
---
a/example/mqtt-customize/src/main/java/org/apache/iotdb/mqtt/server/CustomizedJsonPayloadFormatter.java
+++
b/example/mqtt-customize/src/main/java/org/apache/iotdb/mqtt/server/CustomizedJsonPayloadFormatter.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.protocol.mqtt.Message;
import org.apache.iotdb.db.protocol.mqtt.PayloadFormatter;
import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang3.NotImplementedException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -32,7 +33,7 @@ import java.util.List;
public class CustomizedJsonPayloadFormatter implements PayloadFormatter {
@Override
- public List<Message> format(ByteBuf payload) {
+ public List<Message> format(String topic, ByteBuf payload) {
// Suppose the payload is a json format
if (payload == null) {
return Collections.emptyList();
@@ -53,6 +54,12 @@ public class CustomizedJsonPayloadFormatter implements
PayloadFormatter {
return ret;
}
+ @Override
+ @Deprecated
+ public List<Message> format(ByteBuf payload) {
+ throw new NotImplementedException();
+ }
+
@Override
public String getName() {
// set the value of mqtt_payload_formatter in iotdb-common.properties as
the following string:
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
index 524b0a4718d..2dc63743429 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatter.java
@@ -27,6 +27,7 @@ import com.google.gson.JsonObject;
import com.google.gson.JsonParseException;
import com.google.gson.reflect.TypeToken;
import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang3.NotImplementedException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -48,7 +49,7 @@ public class JSONPayloadFormatter implements PayloadFormatter
{
private static final Gson GSON = new GsonBuilder().create();
@Override
- public List<Message> format(ByteBuf payload) {
+ public List<Message> format(String topic, ByteBuf payload) {
if (payload == null) {
return new ArrayList<>();
}
@@ -79,6 +80,12 @@ public class JSONPayloadFormatter implements
PayloadFormatter {
throw new JsonParseException("payload is invalidate");
}
+ @Override
+ @Deprecated
+ public List<Message> format(ByteBuf payload) {
+ throw new NotImplementedException();
+ }
+
private List<Message> formatJson(JsonObject jsonObject) {
Message message = new Message();
message.setDevice(jsonObject.get(JSON_KEY_DEVICE).getAsString());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
index d678fef937b..c24b816420f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
@@ -79,6 +79,10 @@ public class MPPPublishHandler extends
AbstractInterceptHandler {
@Override
public void onConnect(InterceptConnectMessage msg) {
+ if (msg.getClientID() == null || msg.getClientID().trim().isEmpty()) {
+ LOG.error(
+ "Connection refused: client_id is missing or empty. A valid
client_id is required to establish a connection.");
+ }
if (!clientIdToSessionMap.containsKey(msg.getClientID())) {
MqttClientSession session = new MqttClientSession(msg.getClientID());
sessionManager.login(
@@ -123,7 +127,7 @@ public class MPPPublishHandler extends
AbstractInterceptHandler {
topic,
payload);
- List<Message> events = payloadFormat.format(payload);
+ List<Message> events = payloadFormat.format(topic, payload);
if (events == null) {
return;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/PayloadFormatter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/PayloadFormatter.java
index 5f6527be997..3027c1a6e94 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/PayloadFormatter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/PayloadFormatter.java
@@ -36,8 +36,20 @@ public interface PayloadFormatter {
* @param payload
* @return
*/
+ @Deprecated
List<Message> format(ByteBuf payload);
+ /**
+ * format a payload of a topic to a list of messages
+ *
+ * @param topic
+ * @param payload
+ * @return
+ */
+ default List<Message> format(String topic, ByteBuf payload) {
+ return format(payload);
+ }
+
/**
* get the formatter name
*
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/MQTTService.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/MQTTService.java
index ed2644292f0..908bfedb507 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/MQTTService.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/MQTTService.java
@@ -94,7 +94,7 @@ public class MQTTService implements IService {
String.valueOf(iotDBConfig.getMqttHandlerPoolSize()));
properties.setProperty(BrokerConstants.IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME,
"true");
properties.setProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME,
"false");
-
properties.setProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME,
"true");
+
properties.setProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME,
"false");
properties.setProperty(
BrokerConstants.NETTY_MAX_BYTES_PROPERTY_NAME,
String.valueOf(iotDBConfig.getMqttMaxMessageSize()));
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatterTest.java
index 082225984ca..46f6547a483 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatterTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/mqtt/JSONPayloadFormatterTest.java
@@ -38,10 +38,10 @@ public class JSONPayloadFormatterTest {
+ " }";
ByteBuf buf = Unpooled.copiedBuffer(payload, StandardCharsets.UTF_8);
+ String topic = "";
JSONPayloadFormatter formatter = new JSONPayloadFormatter();
- Message message = formatter.format(buf).get(0);
-
+ Message message = formatter.format(topic, buf).get(0);
assertEquals("root.sg.d1", message.getDevice());
assertEquals(Long.valueOf(1586076045524L), message.getTimestamp());
assertEquals("s1", message.getMeasurements().get(0));
@@ -59,9 +59,10 @@ public class JSONPayloadFormatterTest {
+ " }";
ByteBuf buf = Unpooled.copiedBuffer(payload, StandardCharsets.UTF_8);
+ String topic = "";
JSONPayloadFormatter formatter = new JSONPayloadFormatter();
- Message message = formatter.format(buf).get(1);
+ Message message = formatter.format(topic, buf).get(1);
assertEquals("root.sg.d1", message.getDevice());
assertEquals(Long.valueOf(1586076065526L), message.getTimestamp());
@@ -88,9 +89,10 @@ public class JSONPayloadFormatterTest {
+ "]";
ByteBuf buf = Unpooled.copiedBuffer(payload, StandardCharsets.UTF_8);
+ String topic = "";
JSONPayloadFormatter formatter = new JSONPayloadFormatter();
- Message message = formatter.format(buf).get(1);
+ Message message = formatter.format(topic, buf).get(1);
assertEquals("root.sg.d2", message.getDevice());
assertEquals(Long.valueOf(1586076065526L), message.getTimestamp());
@@ -117,9 +119,10 @@ public class JSONPayloadFormatterTest {
+ "]";
ByteBuf buf = Unpooled.copiedBuffer(payload, StandardCharsets.UTF_8);
+ String topic = "";
JSONPayloadFormatter formatter = new JSONPayloadFormatter();
- Message message = formatter.format(buf).get(3);
+ Message message = formatter.format(topic, buf).get(3);
assertEquals("root.sg.d2", message.getDevice());
assertEquals(Long.valueOf(1586076065526L), message.getTimestamp());
