KVSerializer implements IKVSerializer { @Override public int serializePartitionKV(DataOutputStream out, int partitionId, SizedWritable<?> key, SizedWritable<?> value) throws IOException { if (key.length == SizedWritable.INVALID_LENGTH || value.length == SizedWritable.INVALID_LENGTH) { updateLength(key, value); } final int keyLength = key.length; final int valueLength = value.length; int bytesWritten = KV_HEAD_LENGTH + keyLength + valueLength; if (partitionId != -1) { bytesWritten += Constants.SIZEOF_PARTITION_LENGTH; } if (out.hasUnFlushedData() && out.shortOfSpace(bytesWritten)) { out.flush(); } if (partitionId != -1) { out.writeInt(partitionId); } out.writeInt(keyLength); out.writeInt(valueLength); keySerializer.serialize(key.v, out); valueSerializer.serialize(value.v, out); return bytesWritten; } KVSerializer(Class<K> kclass, Class<V> vclass); @Override void updateLength(SizedWritable<?> key, SizedWritable<?> value); @Override int serializeKV(DataOutputStream out, SizedWritable<?> key, SizedWritable<?> value); @Override int serializePartitionKV(DataOutputStream out, int partitionId,
      SizedWritable<?> key, SizedWritable<?> value); @Override int deserializeKV(DataInputStream in, SizedWritable<?> key,
      SizedWritable<?> value); static final int KV_HEAD_LENGTH; }
@Test public void testSerializePartitionKV() 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 + Constants.SIZEOF_PARTITION_LENGTH)) .thenReturn(true); final int written = serializer.serializePartitionKV(dataOut, 100, key, value); Mockito.verify(dataOut, Mockito.times(1)).flush(); Mockito.verify(dataOut, Mockito.times(5)).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 + Constants.SIZEOF_PARTITION_LENGTH); }