@Override
  public int serializeKV(DataOutputStream out, SizedWritable<?> key, SizedWritable<?> value)
    throws IOException {
    return serializePartitionKV(out, -1, key, value);
  }
@Test
  public void testSerializeNoFlush() throws IOException {
    final DataOutputStream dataOut = Mockito.mock(DataOutputStream.class);

    // suppose there are enough space
    Mockito.when(dataOut.hasUnFlushedData()).thenReturn(true);
    Mockito.when(dataOut.shortOfSpace(anyInt())).thenReturn(false);
    final int written = serializer.serializeKV(dataOut, key, value);

    // flush 0, write 4 int, and 2 byte array
    Mockito.verify(dataOut, Mockito.times(0)).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);
  }