Function TestTaskGroupErrors

cpp/src/arrow/util/task_group_test.cc:76–116

Check TaskGroup behaviour with some successful and some failing tasks

// Check TaskGroup behaviour with some successful and some failing tasks
void TestTaskGroupErrors(std::shared_ptr<TaskGroup> task_group) {
  const int NSUCCESSES = 2;
  const int NERRORS = 20;

  std::atomic<int> count(0);

  auto task_group_was_ok = false;
  task_group->Append([&]() -> Status {
    for (int i = 0; i < NSUCCESSES; ++i) {
      task_group->Append([&]() {
        count++;
        return Status::OK();
      });
    }
    task_group_was_ok = task_group->ok();
    for (int i = 0; i < NERRORS; ++i) {
      task_group->Append([&]() {
        SleepFor(1e-2);
        count++;
        return Status::Invalid("some message");
      });
    }

    return Status::OK();
  });

  // Task error is propagated
  ASSERT_RAISES(Invalid, task_group->Finish());
  ASSERT_TRUE(task_group_was_ok);
  ASSERT_FALSE(task_group->ok());
  if (task_group->parallelism() == 1) {
    // Serial: exactly two successes and an error
    ASSERT_EQ(count.load(), 3);
  } else {
    // Parallel: at least two successes and an error
    ASSERT_GE(count.load(), 3);
    ASSERT_LE(count.load(), 2 * task_group->parallelism());
  }
  // Finish() is idempotent
  ASSERT_RAISES(Invalid, task_group->Finish());
}

void TestTaskGroupCancel(std::shared_ptr<TaskGroup> task_group, StopSource* stop_source) {
  const int NSUCCESSES = 2;