@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 shouldReturnBlocksUponNormalOperation() {
    MailboxReceiveNode mailboxReceiveNode = getPBReceiveNode(1);
    StagePlan stagePlan = new StagePlan(mailboxReceiveNode, _stageMetadata);

    // when
    when(_mailboxService.getReceivingMailbox(MAILBOX_ID_1)).thenReturn(_mailbox1);
    Object[] row1 = new Object[]{1, 1};
    Object[] row2 = new Object[]{2, 3};
    when(_mailbox1.poll()).thenReturn(OperatorTestUtil.block(DATA_SCHEMA, row1),
        OperatorTestUtil.block(DATA_SCHEMA, row2),
        TransferableBlockUtils.getEndOfStreamTransferableBlock(OperatorTestUtil.getDummyStats(1)));

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

    // then
    // should have single PB result, receive 2 data blocks, EOS block shouldn't be included
    Assert.assertNotNull(pipelineBreakerResult);
    Assert.assertNull(pipelineBreakerResult.getErrorBlock());
    Assert.assertEquals(pipelineBreakerResult.getResultMap().size(), 1);
    Assert.assertEquals(pipelineBreakerResult.getResultMap().values().iterator().next().size(), 2);

    // should collect stats from previous stage here
    Assert.assertNotNull(pipelineBreakerResult.getStageQueryStats());
    Assert.assertNotNull(pipelineBreakerResult.getStageQueryStats().getUpstreamStageStats(1),
        "Stats for stage 1 should be sent");
  }