private void extractTable(final TableMessage message, final MqttClientSession session) {
    try {
      TimestampPrecisionUtils.checkTimestampPrecision(message.getTimestamp());
      InsertTabletStatement insertTabletStatement = constructInsertTabletStatement(message);
      session.setSqlDialect(IClientSession.SqlDialect.TABLE);
      final EnrichedEvent event =
          new PipeStatementInsertionEvent(
              pipeName,
              creationTime,
              pipeTaskMeta,
              null,
              null,
              session.getUsername(),
              true,
              true,
              message.getDatabase(),
              insertTabletStatement);
      if (!event.increaseReferenceCount(MQTTPublishHandler.class.getName())) {
        LOGGER.warn("The reference count of the event {} cannot be increased, skipping it.", event);
        return;
      }
      pendingQueue.waitedOffer(event);
    } catch (Exception e) {
      LOGGER.warn(
          "meet error when polling mqtt source message database {}, table {}, tags {}, attributes {}, fields {}, at time {}, because {}",
          message.getDatabase(),
          message.getTable(),
          message.getTagKeys(),
          message.getAttributeKeys(),
          message.getFields(),
          message.getTimestamp(),
          e.getMessage(),
          e);
    }
  }
PublishHandler handle the messages from MQTT clients.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java