@Override
  protected void route(List<SendingMailbox> destinations, TransferableBlock block)
      throws Exception {
    int numMailboxes = destinations.size();
    if (numMailboxes == 1 || _keySelector == EmptyKeySelector.INSTANCE) {
      sendBlock(destinations.get(0), block);
      return;
    }

    //noinspection unchecked
    List<Object[]>[] mailboxIdToRowsMap = new List[numMailboxes];
    for (int i = 0; i < numMailboxes; i++) {
      mailboxIdToRowsMap[i] = new ArrayList<>();
    }
    List<Object[]> rows = block.getContainer();
    for (Object[] row : rows) {
      int mailboxId = _keySelector.computeHash(row) % numMailboxes;
      mailboxIdToRowsMap[mailboxId].add(row);
    }
    for (int i = 0; i < numMailboxes; i++) {
      if (!mailboxIdToRowsMap[i].isEmpty()) {
        sendBlock(destinations.get(i),
            new TransferableBlock(mailboxIdToRowsMap[i], block.getDataSchema(), block.getType()));
      }
    }
  }
@Test
  public void shouldSplitAndRouteBlocksBasedOnPartitionKey()
      throws Exception {
    // Given:
    TestSelector selector = new TestSelector(Iterators.forArray(2, 0, 1));
    Mockito.when(_block.getContainer()).thenReturn(ImmutableList.of(new Object[]{0}, new Object[]{1}, new Object[]{2}));
    ImmutableList<SendingMailbox> destinations = ImmutableList.of(_mailbox1, _mailbox2);

    // When:
    new HashExchange(destinations, selector, TransferableBlockUtils::splitBlock).route(destinations, _block);

    // Then:
    ArgumentCaptor<TransferableBlock> captor = ArgumentCaptor.forClass(TransferableBlock.class);

    Mockito.verify(_mailbox1, Mockito.times(1)).send(captor.capture());
    Assert.assertEquals(captor.getValue().getContainer().get(0), new Object[]{0});
    Assert.assertEquals(captor.getValue().getContainer().get(1), new Object[]{1});

    Mockito.verify(_mailbox2, Mockito.times(1)).send(captor.capture());
    Assert.assertEquals(captor.getValue().getContainer().get(0), new Object[]{2});
  }