GH-47642: [C++] Catch exceptions from initial_task in AsyncTaskScheduler (#49860)

### Rationale for this change

If the `initial_task` passed to `AsyncTaskScheduler::Make` throws a C++ exception (rather than returning a failed `Status`), `OnTaskFinished` is never called. This leaves `running_tasks_` permanently at 1, causing a `DCHECK` failure in debug builds and an indefinite hang in release builds because the scheduler's `finished` future is never completed. In Acero, this manifests as `DeclarationToTable` (and similar APIs) hanging forever when a `SourceNode` generator throws during `StartProducing`.

### What changes are included in this PR?

Add exception handling logic.

### Are these changes tested?

Yes. 

### Are there any user-facing changes?

No API changes. 
* GitHub Issue: #47642

Lead-authored-by: egolearner <lijiliang@outlook.com>
Co-authored-by: Antoine Pitrou <pitrou@free.fr>
Signed-off-by: Antoine Pitrou <antoine@python.org>

Author: egolearner

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();