public static org.apache.avro.Schema toAvroSchema(
      Schema beamSchema, @Nullable String name, @Nullable String namespace) {
    final String schemaName = Strings.isNullOrEmpty(name) ? "topLevelRecord" : name;
    final String schemaNamespace = namespace == null ? "" : namespace;
    String childNamespace =
        !"".equals(schemaNamespace) ? schemaNamespace + "." + schemaName : schemaName;
    List<org.apache.avro.Schema.Field> fields = Lists.newArrayList();
    for (Field field : beamSchema.getFields()) {
      org.apache.avro.Schema.Field recordField = toAvroField(field, childNamespace);
      fields.add(recordField);
    }
    return org.apache.avro.Schema.createRecord(schemaName, null, schemaNamespace, false, fields);
  }
@Test
  public void testAvroSchemaFromBeamSchemaWithFieldCollisionCanBeParsed() {

    // Two similar schemas, the only difference is the "street" field type in the nested record.
    Schema contact =
        new Schema.Builder()
            .addField(Field.of("name", FieldType.STRING))
            .addField(
                Field.of(
                    "address",
                    FieldType.row(
                        new Schema.Builder()
                            .addField(Field.of("street", FieldType.STRING))
                            .addField(Field.of("city", FieldType.STRING))
                            .build())))
            .build();

    Schema contactMultiline =
        new Schema.Builder()
            .addField(Field.of("name", FieldType.STRING))
            .addField(
                Field.of(
                    "address",
                    FieldType.row(
                        new Schema.Builder()
                            .addField(Field.of("street", FieldType.array(FieldType.STRING)))
                            .addField(Field.of("city", FieldType.STRING))
                            .build())))
            .build();

    // Ensure that no collisions happen between two sibling fields with same-named child fields
    // (with different schemas, between a parent field and a sub-record field with the same name,
    // and artificially with the generated field name.
    Schema beamSchema =
        new Schema.Builder()
            .addField(Field.of("home", FieldType.row(contact)))
            .addField(Field.of("work", FieldType.row(contactMultiline)))
            .addField(Field.of("address", FieldType.row(contact)))
            .addField(Field.of("topLevelRecord", FieldType.row(contactMultiline)))
            .build();

    org.apache.avro.Schema convertedSchema = AvroUtils.toAvroSchema(beamSchema);
    org.apache.avro.Schema validatedSchema =
        new org.apache.avro.Schema.Parser().parse(convertedSchema.toString());
    assertEquals(convertedSchema, validatedSchema);
  }