@Override
  public void onPublish(InterceptPublishMessage msg) {
    try {
      final String clientId = msg.getClientID();
      if (!clientIdToSessionMap.containsKey(clientId)) {
        return;
      }
      final MqttClientSession session = clientIdToSessionMap.get(msg.getClientID());
      final ByteBuf payload = msg.getPayload();
      final String topic = msg.getTopicName();
      final String username = msg.getUsername();
      final MqttQoS qos = msg.getQos();

      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(
            "Receive publish message. clientId: {}, username: {}, qos: {}, topic: {}, payload: {}",
            clientId,
            username,
            qos,
            topic,
            payload);
      }

      final List<Message> messages = payloadFormat.format(topic, payload);
      if (messages == null) {
        return;
      }

      for (Message message : messages) {
        if (message == null) {
          continue;
        }
        if (useTableInsert) {
          extractTable((TableMessage) message, session);
        } else {
          extractTree((TreeMessage) message, session);
        }
      }
    } catch (Throwable t) {
      LOGGER.warn("onPublish execution exception, msg is {}, error is ", msg, t);
    } finally {
      // release the payload of the message
      super.onPublish(msg);
    }
  }
PublishHandler handle the messages from MQTT clients.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java