GH-49433: [C++] Buffer ARROW_LOG output to prevent thread interleaving

When using the ARROW_LOG macros from multiple threads, messages were getting
mingled together in stderr because the operator<< was writing directly to
the global stream piece-by-piece.

This commit introduces an internal std::ostringstream buffer to CerrLog so
that messages are accumulated locally and flushed atomically upon destruction.

Author: Shockp

cpp/src/arrow/util/logging.cc

@@ -24,6 +24,8 @@
 #endif
 #include <cstdlib>
 #include <iostream>
+#include <mutex>
+#include <sstream>
 
 #ifdef ARROW_USE_GLOG
 
@@ -67,7 +69,9 @@ class CerrLog {
 
   virtual ~CerrLog() {
     if (has_logged_) {
-      std::cerr << std::endl;
+      static std::mutex cerr_mutex;
+      std::lock_guard<std::mutex> lock(cerr_mutex);
+      std::cerr << buffer_.str() << std::endl;
     }
     if (severity_ == ArrowLogLevel::ARROW_FATAL) {
       PrintBackTrace();
@@ -77,21 +81,22 @@ class CerrLog {
 
   std::ostream& Stream() {
     has_logged_ = true;
-    return std::cerr;
+    return buffer_;
   }
 
   template <class T>
   CerrLog& operator<<(const T& t) {
     if (severity_ != ArrowLogLevel::ARROW_DEBUG) {
       has_logged_ = true;
-      std::cerr << t;
+      buffer_ << t;
     }
     return *this;
   }
 
  protected:
   const ArrowLogLevel severity_;
   bool has_logged_;
+  std::ostringstream buffer_;
 
   void PrintBackTrace() {
 #ifdef ARROW_WITH_BACKTRACE

cpp/src/arrow/util/logging_test.cc

@@ -18,6 +18,8 @@
 #include <chrono>
 #include <cstdint>
 #include <iostream>
+#include <thread>
+#include <vector>
 
 #include <gtest/gtest.h>
 
@@ -95,6 +97,27 @@ TEST(ArrowCheck, PayloadEvaluatedOnFailure) {
   ASSERT_TRUE(tracer.was_printed);
 }
 
+TEST(ArrowLog, MultiThreadedLogging) {
+  constexpr int kNumThreads = 10;
+  constexpr int kNumMessges = 10;
+  std::vector<std::thread> threads;
+
+  threads.reserve(kNumThreads);
+
+  for (int i = 0; i < kNumThreads; ++i) {
+    threads.emplace_back([i]() {
+      for (int j = 0; j < kNumMessges; ++j) {
+        ARROW_LOG(INFO) << "Thread" << i << " message " << j
+                        << " - testing thread safety.";
+      }
+    });
+  }
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+}
+
 }  // namespace util
 
 TEST(DcheckMacros, DoNotEvaluateReleaseMode) {