/**
 * A user facing public util class to help converting {@link org.apache.avro.Schema Avro Schemas}
 * to Flink {@link DataType} and {@link Schema Table Schema}.
 */
@PublicEvolving
public class AvroSchemaUtils {

    /** private constructor for util class to prevent instantiation. */
    private AvroSchemaUtils() {}

    /**
     * Converts the given Avro schema string to a Flink {@link DataType}. This is useful
     * when user wants to convert a Table to a DataStream.
     *
     * @param avroSchemaString Avro schema string
     * @return the DataType converted from Avro schema string.
     * @see org.apache.flink.table.api.bridge.java.StreamTableEnvironment#toDataStream(Table, AbstractDataType)
     */
    public static DataType convertToDataType(String avroSchemaString) {
        return AvroSchemaConverter.convertToDataType(avroSchemaString);
    }

    /**
     * Converts the given Avro schema to a Flink {@link DataType}. This is useful
     * when user wants to convert a Table to a DataStream.
     *
     * @param schema Avro schema
     * @return the DataType converted from Avro schema string.
     * @see org.apache.flink.table.api.bridge.java.StreamTableEnvironment#toDataStream(Table, AbstractDataType)
     */
    public static DataType convertToDataType(org.apache.avro.Schema schema) {
        return AvroSchemaConverter.convertToDataType(schema);
    }

    /**
     * Converts the given Avro schema string to a Flink
     * {@link org.apache.flink.table.api.Schema Table Schema}.This is useful when user wants
     * to convert a DataStream into a Table.
     *
     * @param avroSchemaString Avro schema string.
     * @see org.apache.flink.table.api.bridge.java.StreamTableEnvironment#fromDataStream(DataStream, Schema)
     * @return the Table Schema converted from Avro schema string.
     */
    public static Schema convertToTableSchema(String avroSchemaString) {
        return AvroSchemaConverter.convertToTableSchema(avroSchemaString);
    }

    /**
     * Converts the given Avro schema to a Flink
     * {@link org.apache.flink.table.api.Schema Table Schema}.This is useful when user wants
     * to convert a DataStream into a Table.
     *
     * @param schema Avro schema.
     * @see org.apache.flink.table.api.bridge.java.StreamTableEnvironment#fromDataStream(DataStream, Schema)
     * @return the Table Schema converted from Avro schema string.
     */
    public static Schema convertToTableSchema(org.apache.avro.Schema schema) {
        return AvroSchemaConverter.convertToTableSchema(schema);
    }
}

    public static final ConfigOption<String> AVRO_SCHEMA =
            ConfigOptions.key("schema")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("An optional Avro reader schema string. By default, Flink table"
                            + " will generate an Avro reader schema for a table according to the defined"
                            + " Flink table schema. However, the generated schema has some limitations."
                            + " For example, it may not have the wanted default value for a field."
                            + " In that case, users can provide the intended Avro reader schema using "
                            + " this config to help generate a more accurate reader schema.");

@PublicEvolving
public final class DataTypes {
	...

    /**
     * Data type of a variable-length character string representing a {@link Enum} class.
     *
     * @param clazz the {@link Enum} class.
     */

    public static DataType ENUM_STRING(Class<? extends Enum<?>> clazz) {
        return new AtomicDataType(new VarCharType(Integer.MAX_VALUE), clazz);
    }
	...
}

	{
		...
		final String avroSchemaString = "...";
        StreamExecutionEnvironment env = new StreamExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // Get table schema.
        Schema tableSchema = AvroSchemaUtils.convertToTableSchema(avroSchemaString);
        // Set connector options.
        Map<String, String> connectorOptions = new HashMap<>();
        connectorOptions.put(FactoryUtil.CONNECTOR.key(), "SOME_CONNECTOR");
        connectorOptions.put("format", "avro");
        connectorOptions.put("avro.schema", avroSchemaString);
        // Set additional options for the connector
        connectorOptions.put("additional.options", "...");

        // Create catalog table.
        CatalogTable table = CatalogTable.of(
                Schema.newBuilder().fromSchema(tableSchema).build(),
                "Products Catalog Table",
                Collections.emptyList(),
                connectorOptions);

        // register the catalog table
        Catalog catalog = tEnv
                .getCatalog(tEnv.getCurrentCatalog())
                .orElseThrow(() -> new RuntimeException("Catalog does not exist."));
        catalog.createTable(
                new ObjectPath(tEnv.getCurrentDatabase(), "MyTable"),
                table,
                true);

        // Read the records into t1.
        Table t1 = tEnv.sqlQuery("SELECT * FROM MyTable");

        // Convert t1 to DataStream of MyRecord.
        DataStream<MyRecord> dataStream = tEnv.toDataStream(t1, AvroSchemaUtils.convertToDataType(avroSchemaString));

        // Convert DataStream of MyRecord to t2.
        Table t2 = tEnv.fromDataStream(dataStream, AvroSchemaUtils.convertToTableSchema(avroSchemaString));
		...
	}