avro serdes

  private static <T extends SpecificRecord> T getSpecificRecord(
      String json, Schema writersSchema, Class<T> recordClass)
          throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {

    Schema readersSchema = (Schema) recordClass.getMethod("getClassSchema").invoke(null);
    Decoder jsonDecoder = DecoderFactory.get().jsonDecoder(writersSchema, json);
    ResolvingDecoder resolvingDecoder = DecoderFactory.get().resolvingDecoder(writersSchema, readersSchema, jsonDecoder);
    DatumReader<T> reader = new SpecificDatumReader<>(recordClass);
    return reader.read(null, resolvingDecoder);
  }


  public MySpecificRecord deserialize(ByteBuffer byteBuffer) {
    try {
      BinaryMessageDecoder<MySpecificRecord> decoder =
          MySpecificRecord.createDecoder(customSchemaStore);
      return decoder.decode(byteBuffer);
    } catch (IOException e) {
      throw new IllegalArgumentException("Failed to deserialize MySpecificRecord", e);
    }
  }