refine error message in mpptask

dbms/src/Flash/EstablishCall.cpp

@@ -143,7 +143,7 @@ void EstablishCallData::finishTunnelAndResponder()
     state = FINISH;
     if (mpp_tunnel)
     {
-        mpp_tunnel->consumerFinish("grpc writes failed.", true); //trigger mpp tunnel finish work
+        mpp_tunnel->consumerFinish(fmt::format("{}: finishTunnelAndResponder called.", mpp_tunnel->id()), true); //trigger mpp tunnel finish work
     }
     grpc::Status status(static_cast<grpc::StatusCode>(GRPC_STATUS_UNKNOWN), "Consumer exits unexpected, grpc writes failed.");
     responder.Finish(status, this);

dbms/src/Flash/Mpp/MPPTask.cpp

@@ -379,7 +379,7 @@ void MPPTask::runImpl()
     }
     catch (...)
     {
-        err_msg = getCurrentExceptionMessage(true);
+        err_msg = getCurrentExceptionMessage(true, true);
     }
 
     if (err_msg.empty())
@@ -405,6 +405,8 @@ void MPPTask::runImpl()
         if (status == RUNNING)
         {
             LOG_FMT_ERROR(log, "task running meets error: {}", err_msg);
+            /// trim the stack trace to avoid too many useless information in log
+            trimStackTrace(err_msg);
             try
             {
                 handleError(err_msg);

dbms/src/Flash/Mpp/MPPTunnel.cpp

@@ -220,7 +220,11 @@ void MPPTunnelBase<Writer>::sendJob(bool need_lock)
         err_msg = "fatal error in sendJob()";
     }
     if (!err_msg.empty())
+    {
+        /// append tunnel id to error message
+        err_msg = fmt::format("{} meet error: {}", tunnel_id, err_msg);
         LOG_ERROR(log, err_msg);
+    }
     consumerFinish(err_msg, need_lock);
     if (is_async)
         writer->writeDone(grpc::Status::OK);

dbms/src/Flash/Mpp/Utils.cpp

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include <Flash/Mpp/Utils.h>
+#include <Poco/String.h>
 
 #include <memory>
 
@@ -27,4 +28,14 @@ mpp::MPPDataPacket getPacketWithError(String reason)
     return data;
 }
 
+void trimStackTrace(String & message)
+{
+    auto stack_trace_pos = message.find("Stack trace");
+    if (stack_trace_pos != String::npos)
+    {
+        message.resize(stack_trace_pos);
+        Poco::trimRightInPlace(message);
+    }
+}
+
 } // namespace DB

dbms/src/Flash/Mpp/Utils.h

@@ -23,5 +23,6 @@
 namespace DB
 {
 mpp::MPPDataPacket getPacketWithError(String reason);
+void trimStackTrace(String & message);
 
 } // namespace DB