@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;
  }
@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);

    // flush once, write 4 int, and 2 byte array
    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);
  }