@Override
  public int deserializeKV(DataInputStream in, SizedWritable<?> key,
      SizedWritable<?> value) throws IOException {

    if (!in.hasUnReadData()) {
      return 0;
    }

    key.length = in.readInt();
    value.length = in.readInt();

    keySerializer.deserialize(in, key.length, key.v);
    valueSerializer.deserialize(in, value.length, value.v);

    return key.length + value.length + KV_HEAD_LENGTH;
  }
@Test
  public void testDeserializerNoData() throws IOException {
    final DataInputStream in = Mockito.mock(DataInputStream.class);
    Mockito.when(in.hasUnReadData()).thenReturn(false);
    Assert.assertEquals(0, serializer.deserializeKV(in, key, value));
  }