@Nullable
  public static PipelineBreakerResult executePipelineBreakers(OpChainSchedulerService scheduler,
      MailboxService mailboxService, WorkerMetadata workerMetadata, StagePlan stagePlan,
      Map<String, String> opChainMetadata, long requestId, long deadlineMs) {
    PipelineBreakerContext pipelineBreakerContext = new PipelineBreakerContext();
    PipelineBreakerVisitor.visitPlanRoot(stagePlan.getRootNode(), pipelineBreakerContext);
    if (!pipelineBreakerContext.getPipelineBreakerMap().isEmpty()) {
      try {
        // TODO: This PlanRequestContext needs to indicate it is a pre-stage opChain and only listens to pre-stage
        //     OpChain receive-mail callbacks.
        // see also: MailboxIdUtils TODOs, de-couple mailbox id from query information
        OpChainExecutionContext opChainExecutionContext =
            new OpChainExecutionContext(mailboxService, requestId, deadlineMs, opChainMetadata,
                stagePlan.getStageMetadata(), workerMetadata, null);
        return execute(scheduler, pipelineBreakerContext, opChainExecutionContext);
      } catch (Exception e) {
        LOGGER.error("Caught exception executing pipeline breaker for request: {}, stage: {}", requestId,
            stagePlan.getStageMetadata().getStageId(), e);
        return new PipelineBreakerResult(pipelineBreakerContext.getNodeIdMap(), Collections.emptyMap(),
            TransferableBlockUtils.getErrorTransferableBlock(e), null);
      }
    } else {
      return null;
    }
  }
@Test
  public void shouldReturnEmptyBlockWhenPBExecuteWithIncorrectMailboxNode() {
    MailboxReceiveNode incorrectlyConfiguredMailboxNode = getPBReceiveNode(3);
    StagePlan stagePlan = new StagePlan(incorrectlyConfiguredMailboxNode, _stageMetadata);

    // when
    PipelineBreakerResult pipelineBreakerResult =
        PipelineBreakerExecutor.executePipelineBreakers(_scheduler, _mailboxService, _workerMetadata, stagePlan,
            ImmutableMap.of(), 0, Long.MAX_VALUE);

    // then
    // should return empty block list
    Assert.assertNotNull(pipelineBreakerResult);
    Assert.assertNull(pipelineBreakerResult.getErrorBlock());
    Assert.assertEquals(pipelineBreakerResult.getResultMap().size(), 1);
    List<TransferableBlock> resultBlocks = pipelineBreakerResult.getResultMap().values().iterator().next();
    Assert.assertEquals(resultBlocks.size(), 0);

    Assert.assertNotNull(pipelineBreakerResult.getStageQueryStats());
  }