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 testJdbcLogicalDateAndTimeRowDataToAvroSchema() {
    String expectedAvroSchemaJson =
        "{ "
            + " \"name\": \"topLevelRecord\", "
            + " \"type\": \"record\", "
            + " \"fields\": [{ "
            + "   \"name\": \"my_date_field\", "
            + "   \"type\": { \"type\": \"int\", \"logicalType\": \"date\" }"
            + "  }, "
            + "  { "
            + "   \"name\": \"my_time_field\", "
            + "   \"type\": { \"type\": \"int\", \"logicalType\": \"time-millis\" }"
            + "  }"
            + " ] "
            + "}";

    Schema beamSchema =
        Schema.builder()
            .addField(Field.of("my_date_field", FieldType.logicalType(JdbcType.DATE)))
            .addField(Field.of("my_time_field", FieldType.logicalType(JdbcType.TIME)))
            .build();

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