@Test
public void testSerializeKV() throws IOException {

  final DataOutputStream dataOut = Mockito.mock(DataOutputStream.class);
  Mockito.when(dataOut.hasUnFlushedData()).thenReturn(true);
  Mockito.when(dataOut.shortOfSpace(key.length + value.length +
                                    Constants.SIZEOF_KV_LENGTH)).thenReturn(true);
  final int written = serializer.serializeKV(dataOut, key, value);
  // flush once, write 4 int, and 2 byte array
  Mockito.verify(dataOut, Mockito.times(1)).flush();
  Mockito.verify(dataOut, Mockito.times(4)).writeInt(anyInt());
  Mockito.verify(dataOut, Mockito.times(2)).write(any(byte[].class),
      anyInt(), anyInt());
  Assert.assertEquals(written, key.length + value.length + Constants.SIZEOF_KV_LENGTH);
}