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 shouldSignalEarlyTerminationProperly()
      throws Exception {
    // Given:
    List<SendingMailbox> destinations = ImmutableList.of(_mailbox1, _mailbox2);
    BlockExchange exchange = new TestBlockExchange(destinations);
    TransferableBlock block = new TransferableBlock(ImmutableList.of(new Object[]{"val"}),
        new DataSchema(new String[]{"foo"}, new ColumnDataType[]{ColumnDataType.STRING}), DataBlock.Type.ROW);

    // When send normal block and some mailbox has terminated
    when(_mailbox1.isEarlyTerminated()).thenReturn(true);
    boolean isEarlyTerminated = exchange.send(block);

    // Then:
    Assert.assertFalse(isEarlyTerminated);

    // When send normal block and both terminated
    when(_mailbox2.isTerminated()).thenReturn(true);
    isEarlyTerminated = exchange.send(block);

    // Then:
    Assert.assertFalse(isEarlyTerminated);

    // When send metadata block
    when(_mailbox2.isEarlyTerminated()).thenReturn(true);
    isEarlyTerminated = exchange.send(block);

    // Then:
    Assert.assertTrue(isEarlyTerminated);
  }