@Test
public void testSend() throws Exception {

  Mutation<?> firstMutation = mock(Mutation.class);
  Mutation<?> secondMutation = mock(Mutation.class);
  Mutation<?> thirdMutation = mock(Mutation.class);
  Mutation<?> fourthMutation = mock(Mutation.class);
  Mutation<?> fifthMutation = mock(Mutation.class);
  Mutation.Metadata firstMetadata = mock(Mutation.Metadata.class);
  Mutation.Metadata secondMetadata = mock(Mutation.Metadata.class);
  Mutation.Metadata thirdMetadata = mock(Mutation.Metadata.class);
  when(firstMetadata.getId()).thenReturn(3L);
  when(secondMetadata.getId()).thenReturn(2L);
  when(thirdMetadata.getId()).thenReturn(4L);
  when(firstMutation.getMetadata()).thenReturn(firstMetadata);
  when(secondMutation.getMetadata()).thenReturn(secondMetadata);
  when(thirdMutation.getMetadata()).thenReturn(thirdMetadata);
  when(keyProvider.get(firstMutation)).thenReturn("4");
  when(keyProvider.get(secondMutation)).thenReturn("6");
  when(keyProvider.get(thirdMutation)).thenReturn("9");
  when(keyProvider.get(fourthMutation)).thenReturn("2");
  when(keyProvider.get(fifthMutation)).thenReturn("1");
  assertEquals(null, firstDestination.getLastPublishedMutation());
  destinationPool.send(
      Arrays.asList(firstMutation, secondMutation, thirdMutation, fourthMutation, fifthMutation));
  verify(firstDestination, times(1)).send(Collections.singletonList(firstMutation));
  verify(secondDestination, times(1)).send(ImmutableList.of(thirdMutation, fifthMutation));
  verify(thirdDestination, times(1)).send(ImmutableList.of(secondMutation, fourthMutation));
  when(firstDestination.getLastPublishedMutation()).thenReturn((Mutation) firstMutation);
  when(secondDestination.getLastPublishedMutation()).thenReturn((Mutation) thirdMutation);
  when(thirdDestination.getLastPublishedMutation()).thenReturn((Mutation) secondMutation);
  when(fourthDestination.getLastPublishedMutation()).thenReturn(null);
  assertEquals(secondMutation, destinationPool.getLastPublishedMutation());
  when(firstDestination.getLastPublishedMutation()).thenReturn(null);
  assertEquals(null, destinationPool.getLastPublishedMutation());
}