public static GenericRecord convertToAvro(Schema schema, Message message) {
    return AvroSupport.convert(schema, message);
  }
@Test
  public void recursiveSchema_withOverflow() throws Exception {
    Schema.Parser parser = new Schema.Parser();
    Schema convertedSchema = parser.parse(getClass().getClassLoader().getResourceAsStream("schema-provider/proto/parent_schema_recursive_depth_2.avsc"));
    Pair<Parent, GenericRecord> inputAndOutput = createInputOutputForRecursiveSchemaWithOverflow(convertedSchema);
    Parent input = inputAndOutput.getLeft();
    GenericRecord actual = serializeAndDeserializeAvro(ProtoConversionUtil.convertToAvro(convertedSchema, inputAndOutput.getLeft()), convertedSchema);
    Assertions.assertEquals(inputAndOutput.getRight(), actual);
    // assert that overflow data can be read back into proto class
    Child parsedSingleChildOverflow = Child.parseFrom(getOverflowBytesFromChildRecord((GenericRecord) actual.get("child")));
    Assertions.assertEquals(input.getChild().getRecurseField().getRecurseField(), parsedSingleChildOverflow);
    // Get children list
    GenericData.Array<GenericRecord> array = (GenericData.Array<GenericRecord>) actual.get("children");
    Child parsedChildren1Overflow = Child.parseFrom(getOverflowBytesFromChildRecord(array.get(0)));
    Assertions.assertEquals(input.getChildren(0).getRecurseField().getRecurseField(), parsedChildren1Overflow);
    Child parsedChildren2Overflow = Child.parseFrom(getOverflowBytesFromChildRecord(array.get(1)));
    Assertions.assertEquals(input.getChildren(1).getRecurseField().getRecurseField(), parsedChildren2Overflow);
  }