...

		StreamTableEnvironment tEnv        String schemaRegistryURL = ...;

		ConfluentSchemaRegistryCatalog        Map<String, String> kafkaProps = ...;
        SchemaRegistryCatalog catalog = new ConfluentSchemaRegistryCatalog(
			properties,
			schemaRegistryURL,
			"catalog1",
			"db1");
		SchemaRegistryCatalog.builder()
                .schemaRegistryURL(schemaRegistryURL)
                .kafkaOptions(kafkaProps)
                .catalogName("myCatalog")
                .dbName("myDB")
                .build();
        tEnv.registerCatalog("catalog1", catalog);

		        // ---------- Consume stream from Kafka -------------------

		String        // Assumes there is a topic named 'transactions'
       String query = "SELECT\n" +
			           "  id, amount\n" +
			           "FROM catalog1myCatalog.db1myDB.transactions1";transactions";

Introduction to Confluent Schema Registry

Terminology

What is a topic versus a schema versus a subject?

• Subject: Schema Registry defines a scope in which schemas can evolve, and that scope is the subject. The name of the subject depends on the configured subject name strategy, which by default is set to derive subject name from topic name
• Topic：A Kafka topic contains messages, and each message is a key-value pair. Either the message key or the message value, or both, can be serialized as Avro, JSON, or Protobuf
• Schema name：Schema name, for Avro it is the record name, for Json, it is the title name

See terminology-review for details.

Subject Naming Strategy

There are 3 kinds of naming strategy for current 5.5.1 version:

• TopicNameStrategy : <topic name>-key | <topic name>-value
• RecordNameStrategy : <fully-qualified record name>-key |  <fully-qualified record name>-value
• TopicRecordNameStrategy : <topic name>-<fully-qualified record name>-key | <topic name>-<fully-qualified record name>-value

The RecordNameStrategy allows different schemas(record type) within one Kafka topic，the schema compatibility check for same record name are among all the topics，while TopicRecordNameStrategy checks the compatibility of same record name schema within one Kafka topic.

See sr-schemas-subject-name-strategy for details.

The Restful API

See schemaregistry-api for details.

Design Proposal

ConfluentSchemaRegistryCatalog

...