@Override
  public BaseCombineOperator run() {
    try (InvocationScope ignored = Tracing.getTracer().createScope(CombinePlanNode.class)) {
      return getCombineOperator();
    }
  }
@Test
  public void testCancelPlanNode() {
    CountDownLatch ready = new CountDownLatch(20);
    List<PlanNode> planNodes = new ArrayList<>();
    for (int i = 0; i < 20; i++) {
      planNodes.add(() -> {
        ready.countDown();
        return null;
      });
    }
    // This planNode will keep the planning running and wait to be cancelled.
    CountDownLatch hold = new CountDownLatch(1);
    planNodes.add(() -> {
      try {
        hold.await();
        return null;
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    });
    _queryContext.setEndTimeMs(System.currentTimeMillis() + Server.DEFAULT_QUERY_EXECUTOR_TIMEOUT_MS);
    CombinePlanNode combinePlanNode = new CombinePlanNode(planNodes, _queryContext, _executorService, null);
    AtomicReference<Exception> exp = new AtomicReference<>();
    // Avoid early finalization by not using Executors.newSingleThreadExecutor (java <= 20, JDK-8145304)
    ExecutorService combineExecutor = Executors.newFixedThreadPool(1);
    try {
      Future<?> future = combineExecutor.submit(() -> {
        try {
          return combinePlanNode.run();
        } catch (Exception e) {
          exp.set(e);
          throw e;
        }
      });
      ready.await();
      // At this point, the combinePlanNode is or will be waiting on future.get() for all sub planNodes, and the
      // waiting can be cancelled as below.
      future.cancel(true);
    } catch (Exception e) {
      Assert.fail();
    } finally {
      combineExecutor.shutdownNow();
    }
    TestUtils.waitForCondition((aVoid) -> exp.get() instanceof QueryCancelledException, 10_000,
        "Should have been cancelled");
  }