GH-47642: [C++] Catch exceptions from initial_task in AsyncTaskScheduler

cpp/src/arrow/util/async_util.cc

@@ -23,6 +23,7 @@
 #include "arrow/util/tracing_internal.h"
 
 #include <condition_variable>
+#include <exception>
 #include <list>
 #include <memory>
 #include <mutex>
@@ -466,7 +467,19 @@ Future<> AsyncTaskScheduler::Make(FnOnce<Status(AsyncTaskScheduler*)> initial_ta
   auto scope = START_SCOPED_SPAN_SV(span, "AsyncTaskScheduler::InitialTask"sv);
   auto scheduler = std::make_unique<AsyncTaskSchedulerImpl>(std::move(stop_token),
                                                             std::move(abort_callback));
-  Status initial_task_st = std::move(initial_task)(scheduler.get());
+  Status initial_task_st;
+  // GH-47642: We normally don't catch exceptions in Arrow C++ code, as the error
+  // reporting model uses the Status object instead. Usually, an uncaught exception
+  // will simply terminate the process, surfacing the programming error.
+  // However, an exception thrown from the initial task would result in a much
+  // harder to diagnose process hang.
+  try {
+    initial_task_st = std::move(initial_task)(scheduler.get());
+  } catch (const std::exception& e) {
+    initial_task_st = Status::UnknownError("Initial task threw an exception: ", e.what());
+  } catch (...) {
+    initial_task_st = Status::UnknownError("Initial task threw an unknown exception");
+  }
   scheduler->OnTaskFinished(std::move(initial_task_st));
   // Keep scheduler alive until finished
   return scheduler->OnFinished().Then([scheduler = std::move(scheduler)] {});

cpp/src/arrow/util/async_util_test.cc

@@ -23,9 +23,11 @@
 #include <memory>
 #include <mutex>
 #include <queue>
+#include <stdexcept>
 #include <thread>
 #include <unordered_set>
 
+#include <gmock/gmock-matchers.h>
 #include <gtest/gtest.h>
 
 #include "arrow/result.h"
@@ -204,6 +206,31 @@ TEST(AsyncTaskScheduler, InitialTaskFails) {
   ASSERT_FINISHES_AND_RAISES(Invalid, finished);
 }
 
+TEST(AsyncTaskScheduler, InitialTaskThrowsException) {
+  // If the initial task throws a C++ exception (not a Status), the scheduler
+  // should catch it, convert to a failed Status, and not hang indefinitely.
+  // See https://github.com/apache/arrow/issues/47642
+
+  // Case 1: initial task throws with no other tasks
+  Future<> finished =
+      AsyncTaskScheduler::Make([&](AsyncTaskScheduler* scheduler) -> Status {
+        throw std::runtime_error("some exception");
+      });
+  EXPECT_FINISHES_AND_RAISES_WITH_MESSAGE_THAT(
+      UnknownError, ::testing::HasSubstr("some exception"), finished);
+
+  // Case 2: initial task throws while another task is still running
+  Future<> task = Future<>::Make();
+  finished = AsyncTaskScheduler::Make([&](AsyncTaskScheduler* scheduler) -> Status {
+    EXPECT_TRUE(scheduler->AddSimpleTask([&]() { return task; }, kDummyName));
+    throw std::runtime_error("some exception after adding task");
+  });
+  AssertNotFinished(finished);
+  task.MarkFinished();
+  EXPECT_FINISHES_AND_RAISES_WITH_MESSAGE_THAT(
+      UnknownError, ::testing::HasSubstr("some exception after adding task"), finished);
+}
+
 TEST(AsyncTaskScheduler, TaskDestroyedBeforeSchedulerEnds) {
   bool my_task_destroyed = false;
   Future<> task_fut = Future<>::Make();