diff --git a/docs/zh/06-advanced/05-data-in/07-mqtt.md b/docs/zh/06-advanced/05-data-in/07-mqtt.md
index af99cd3621..f54086b61b 100644
--- a/docs/zh/06-advanced/05-data-in/07-mqtt.md
+++ b/docs/zh/06-advanced/05-data-in/07-mqtt.md
@@ -33,13 +33,14 @@ TDengine 可以通过 MQTT 连接器从 MQTT 代理订阅数据并将其写入 T
 
 ### 3. 配置连接和认证信息
 
-在 **MQTT地址** 中填写 MQTT 代理的地址，例如：`192.168.1.42:1883`
+在 **MQTT 地址** 中填写 MQTT 代理的地址，例如：`192.168.1.42`
+
+在 **MQTT 端口** 中填写 MQTT 代理的端口，例如：`1883`
 
 在 **用户** 中填写 MQTT 代理的用户名。
 
 在 **密码** 中填写 MQTT 代理的密码。
 
-点击 **连通性检查** 按钮，检查数据源是否可用。
 
 ![mqtt-03.png](./mqtt-03.png)
 
@@ -64,6 +65,8 @@ TDengine 可以通过 MQTT 连接器从 MQTT 代理订阅数据并将其写入 T
 
 在 **订阅主题及 QoS 配置** 中填写要消费的 Topic 名称。使用如下格式设置： `topic1::0,topic2::1`。
 
+点击 **检查连通性** 按钮，检查数据源是否可用。
+
 ![mqtt-05.png](./mqtt-05.png)
 
 ### 6. 配置 MQTT Payload 解析
diff --git a/docs/zh/06-advanced/05-data-in/mqtt-03.png b/docs/zh/06-advanced/05-data-in/mqtt-03.png
index c53da8f14d..e671d32675 100644
Binary files a/docs/zh/06-advanced/05-data-in/mqtt-03.png and b/docs/zh/06-advanced/05-data-in/mqtt-03.png differ
diff --git a/docs/zh/06-advanced/05-data-in/mqtt-05.png b/docs/zh/06-advanced/05-data-in/mqtt-05.png
index d362e8be86..2b4cfcabb3 100644
Binary files a/docs/zh/06-advanced/05-data-in/mqtt-05.png and b/docs/zh/06-advanced/05-data-in/mqtt-05.png differ
diff --git a/docs/zh/14-reference/01-components/04-taosx.md b/docs/zh/14-reference/01-components/04-taosx.md
index ce372a8007..e78fdba3d0 100644
--- a/docs/zh/14-reference/01-components/04-taosx.md
+++ b/docs/zh/14-reference/01-components/04-taosx.md
@@ -451,6 +451,16 @@ taosX 会将监控指标上报给 taosKeeper，这些监控指标会被 taosKeep
 | written_blocks          | 本次运行此任务写人成功的 raw block 数                       |
 | failed_blocks           | 本次运行此任务写入失败的 raw block 数                       |
 
+### Kafka 数据源相关指标
+
+| 字段                          | 描述                         |
+| ----------------------------- | ---------------------------- |
+| kafka_consumers               | 本次运行任务 Kafka 消费者数  |
+| kafka_total_partitions        | Kafka 主题总分区数           |
+| kafka_consuming_partitions    | 本次运行任务正在消费的分区数 |
+| kafka_consumed_messages       | 本次运行任务已经消费的消息数 |
+| total_kafka_consumed_messages | 累计消费的消息总数           |
+
 ## taosX 数据解析插件
 
 接入 kafka / mqtt 消息中间件时，需要对原始数据进行解析，如果使用 json/regex 等模式解析器无法满足解析需求，同时 UDT(自定义解析脚本) 也无法满足性能要求时，可以自定义数据解析插件。
diff --git a/source/libs/executor/src/executorInt.c b/source/libs/executor/src/executorInt.c
index 5fc483087a..8c76b1a1c3 100644
--- a/source/libs/executor/src/executorInt.c
+++ b/source/libs/executor/src/executorInt.c
@@ -687,10 +687,10 @@ int32_t copyResultrowToDataBlock(SExprInfo* pExprInfo, int32_t numOfExprs, SResu
       code = blockDataEnsureCapacity(pBlock, pBlock->info.rows + pCtx[j].resultInfo->numOfRes);
       QUERY_CHECK_CODE(code, lino, _end);
 
-      int32_t winCode = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
-      if (TAOS_FAILED(winCode)) {
-        qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(winCode));
-        QUERY_CHECK_CODE(winCode, lino, _end);
+      code = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
+      if (TSDB_CODE_SUCCESS != code) {
+        qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(code));
+        QUERY_CHECK_CODE(code, lino, _end);
       }
     } else if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_select_value") == 0) {
       // do nothing