public boolean send(TransferableBlock block)
      throws Exception {
    if (block.isErrorBlock()) {
      // Send error block to all mailboxes to propagate the error
      for (SendingMailbox sendingMailbox : _sendingMailboxes) {
        sendBlock(sendingMailbox, block);
      }
      return false;
    }

    if (block.isSuccessfulEndOfStreamBlock()) {
      // Send metadata to only one randomly picked mailbox, and empty EOS block to other mailboxes
      int numMailboxes = _sendingMailboxes.size();
      int mailboxIdToSendMetadata = ThreadLocalRandom.current().nextInt(numMailboxes);
      assert block.getQueryStats() != null;
      for (int i = 0; i < numMailboxes; i++) {
        SendingMailbox sendingMailbox = _sendingMailboxes.get(i);
        TransferableBlock blockToSend =
            i == mailboxIdToSendMetadata ? block : TransferableBlockUtils.getEndOfStreamTransferableBlock();
        sendBlock(sendingMailbox, blockToSend);
      }
      return false;
    }

    assert block.isDataBlock();
    boolean isEarlyTerminated = true;
    for (SendingMailbox sendingMailbox : _sendingMailboxes) {
      if (!sendingMailbox.isEarlyTerminated()) {
        isEarlyTerminated = false;
        break;
      }
    }
    if (!isEarlyTerminated) {
      route(_sendingMailboxes, block);
    }
    return isEarlyTerminated;
  }
@Test
  public void shouldSplitBlocks()
      throws Exception {
    // Given:
    List<SendingMailbox> destinations = ImmutableList.of(_mailbox1);

    DataSchema schema = new DataSchema(new String[]{"foo"}, new ColumnDataType[]{ColumnDataType.STRING});

    TransferableBlock inBlock =
        new TransferableBlock(ImmutableList.of(new Object[]{"one"}, new Object[]{"two"}), schema, DataBlock.Type.ROW);

    TransferableBlock outBlockOne =
        new TransferableBlock(ImmutableList.of(new Object[]{"one"}), schema, DataBlock.Type.ROW);

    TransferableBlock outBlockTwo =
        new TransferableBlock(ImmutableList.of(new Object[]{"two"}), schema, DataBlock.Type.ROW);

    BlockExchange exchange = new TestBlockExchange(destinations,
        (block, type, maxSize) -> ImmutableList.of(outBlockOne, outBlockTwo).iterator());

    // When:
    exchange.send(inBlock);

    // Then:
    ArgumentCaptor<TransferableBlock> captor = ArgumentCaptor.forClass(TransferableBlock.class);
    Mockito.verify(_mailbox1, Mockito.times(2)).send(captor.capture());

    List<TransferableBlock> sentBlocks = captor.getAllValues();
    Assert.assertEquals(sentBlocks.size(), 2, "expected to send two blocks");
    Assert.assertEquals(sentBlocks.get(0).getContainer(), outBlockOne.getContainer());
    Assert.assertEquals(sentBlocks.get(1).getContainer(), outBlockTwo.getContainer());
  }