[bugfix](core) writer status is read and write concurrently and will core by use after free (#29982)

Co-authored-by: yiguolei <yiguolei@gmail.com>

be/src/vec/sink/writer/async_result_writer.cpp

@@ -51,13 +51,13 @@ Status AsyncResultWriter::sink(Block* block, bool eos) {
         add_block = _get_free_block(block, rows);
     }
 
+    std::lock_guard l(_m);
     // if io task failed, just return error status to
     // end the query
     if (!_writer_status.ok()) {
         return _writer_status;
     }
 
-    std::lock_guard l(_m);
     _eos = eos;
     if (_dependency && _is_finished()) {
         _dependency->set_ready();
@@ -128,6 +128,10 @@ void AsyncResultWriter::process_block(RuntimeState* state, RuntimeProfile* profi
     // if not in transaction or status is in error or force close we can do close in
     // async IO thread
     if (!_writer_status.ok() || !in_transaction()) {
+        std::lock_guard l(_m);
+        // Using lock to make sure the writer status is not modified
+        // There is a unique ptr err_msg in Status, if it is modified, the unique ptr
+        // maybe released. And it will core because use after free.
         _writer_status = close(_writer_status);
         _need_normal_close = false;
     }

be/src/vec/sink/writer/async_result_writer.h

@@ -84,7 +84,10 @@ public:
     // Add the IO thread task process block() to thread pool to dispose the IO
     void start_writer(RuntimeState* state, RuntimeProfile* profile);
 
-    Status get_writer_status() { return _writer_status; }
+    Status get_writer_status() {
+        std::lock_guard l(_m);
+        return _writer_status;
+    }
 
 protected:
     Status _projection_block(Block& input_block, Block* output_block);