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 testJdbcLogicalVarCharRowDataToAvroSchema() {
    String expectedAvroSchemaJson =
        "{ "
            + " \"name\": \"topLevelRecord\", "
            + " \"type\": \"record\", "
            + " \"fields\": [{ "
            + "   \"name\": \"my_varchar_field\", "
            + "   \"type\": {\"type\": \"string\", \"logicalType\": \"varchar\", \"maxLength\": 10}"
            + "  }, "
            + "  { "
            + "   \"name\": \"my_longvarchar_field\", "
            + "   \"type\": {\"type\": \"string\", \"logicalType\": \"varchar\", \"maxLength\": 50}"
            + "  }, "
            + "  { "
            + "   \"name\": \"my_nvarchar_field\", "
            + "   \"type\": {\"type\": \"string\", \"logicalType\": \"varchar\", \"maxLength\": 10}"
            + "  }, "
            + "  { "
            + "   \"name\": \"my_longnvarchar_field\", "
            + "   \"type\": {\"type\": \"string\", \"logicalType\": \"varchar\", \"maxLength\": 50}"
            + "  }, "
            + "  { "
            + "   \"name\": \"fixed_length_char_field\", "
            + "   \"type\": {\"type\": \"string\", \"logicalType\": \"char\", \"maxLength\": 25}"
            + "  } "
            + " ] "
            + "}";

    Schema beamSchema =
        Schema.builder()
            .addField(
                Field.of(
                    "my_varchar_field", FieldType.logicalType(JdbcType.StringType.varchar(10))))
            .addField(
                Field.of(
                    "my_longvarchar_field",
                    FieldType.logicalType(JdbcType.StringType.longvarchar(50))))
            .addField(
                Field.of(
                    "my_nvarchar_field", FieldType.logicalType(JdbcType.StringType.nvarchar(10))))
            .addField(
                Field.of(
                    "my_longnvarchar_field",
                    FieldType.logicalType(JdbcType.StringType.longnvarchar(50))))
            .addField(
                Field.of(
                    "fixed_length_char_field",
                    FieldType.logicalType(JdbcType.StringType.fixedLengthChar(25))))
            .build();

    assertEquals(
        new org.apache.avro.Schema.Parser().parse(expectedAvroSchemaJson),
        AvroUtils.toAvroSchema(beamSchema));
  }