@Override
  public Coder<KV<byte[], byte[]>> getOutputCoder() {
    return KvCoder.of(ByteArrayCoder.of(), ByteArrayCoder.of());
  }
@Test
  public void coderShouldBeSane() {
    CoderProperties.coderSerializable(source.getOutputCoder());
  }