public static AvroGenericCoder of(Schema schema) {
    return AvroGenericCoder.of(schema);
  }
@Test
  public void testKryoSerialization() throws Exception {
    Pojo value = new Pojo("Hello", 42, DATETIME_A);
    AvroCoder<Pojo> coder = AvroCoder.of(Pojo.class);

    // Kryo instantiation
    Kryo kryo = new Kryo();
    kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
    kryo.addDefaultSerializer(AvroCoder.SerializableSchemaSupplier.class, JavaSerializer.class);

    // Serialization of object without any memoization
    ByteArrayOutputStream coderWithoutMemoizationBos = new ByteArrayOutputStream();
    try (Output output = new Output(coderWithoutMemoizationBos)) {
      kryo.writeClassAndObject(output, coder);
    }

    // Force thread local memoization to store values.
    CoderProperties.coderDecodeEncodeEqual(coder, value);

    // Serialization of object with memoized fields
    ByteArrayOutputStream coderWithMemoizationBos = new ByteArrayOutputStream();
    try (Output output = new Output(coderWithMemoizationBos)) {
      kryo.writeClassAndObject(output, coder);
    }

    // Copy empty and memoized variants of the Coder
    ByteArrayInputStream bisWithoutMemoization =
        new ByteArrayInputStream(coderWithoutMemoizationBos.toByteArray());
    AvroCoder<Pojo> copiedWithoutMemoization =
        (AvroCoder<Pojo>) kryo.readClassAndObject(new Input(bisWithoutMemoization));
    ByteArrayInputStream bisWithMemoization =
        new ByteArrayInputStream(coderWithMemoizationBos.toByteArray());
    AvroCoder<Pojo> copiedWithMemoization =
        (AvroCoder<Pojo>) kryo.readClassAndObject(new Input(bisWithMemoization));

    CoderProperties.coderDecodeEncodeEqual(copiedWithoutMemoization, value);
    CoderProperties.coderDecodeEncodeEqual(copiedWithMemoization, value);
  }