Table of Contents

Status

Current state: ["Under Discussion"]

Discussion thread: 

JIRA: 

...

Page properties
Discussion thread Vote thread JIRA Release 1.3

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

Code Block
language java linenumbers true
String schemaRegistryURL = ...;         Map<String, String> kafkaProps = ...;         SchemaRegistryCatalog catalog = SchemaRegistryCatalog.builder()                 .schemaRegistryURL(schemaRegistryURL)                 .kafkaOptions(kafkaProps)                 .catalogName("myCatalog")                 .dbName("myDB")                 .build();         tEnv.registerCatalog("catalog1myCatalog", catalog);         // ---------- Consume stream from Kafka -------------------         // Assumes there is a topic named 'transactions'          StringString query = "SELECT\n" +             " id, amount\n" +             "FROM myCatalog.myDB.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 schemaregistry-api for details.

security

• https://docs.confluent.io/current/security/basic-auth.html#schema-registry
• https://docs.confluent.io/current/schema-registry/security/index.html#sr-security-overview
• https://docs.confluent.io/current/schema-registry/serdes-develop/index.html#basic-auth-security-for-producers-and-consumers

Schema Compatibility

From 5.5.0, Schema Registry allows different  compatibility level per-subject or globally, the Schema Registry service would force compatibility check when the schema evolves.

• https://docs.confluent.io/current/schema-registry/avro.html#compatibility-types
• http://avro.apache.org/docs/1.7.7/spec.html#Schema+Resolution
• https://docs.confluent.io/current/schema-registry/serdes-develop/serdes-json.html#json-schema-compatibility-rules
• https://docs.confluent.io/current/schema-registry/develop/api.html#id1

Schema & Format

Schema and Format Binding

From 5.5.0, user can get format for a schema through confluent SchemaRegistryClient#getLatestSchemaMetadata, i.e. Avro, Json or Protobuf.

Format Compatibility

The format also has its own compatibility rules, for example, for Avro: http://avro.apache.org/docs/1.7.7/spec.html#Schema+Resolution

Design Proposal

ConfluentSchemaRegistryCatalog

Note: We only support avro format for this FLIP !!!

SchemaRegistryCatalog

The SchemaRegistryCatalog The ConfluentSchemaRegistryCatalog interacts with the Confluent Schema Registry Service directly through its RESTful API, java SDK client, e.g. the SchemaRegistryClient.

• when open/initialize, it fetches all the topic list of current Kafka cluster and makes a cache in the memory.

...

• when a specific catalog table info is requested from the SQL context, it fetches the Avro schema string and format type of the topic's latest subject schema, then generates a catalog table instance and put it into the local cache.
• We dropped the idea to have table cache because the cache would introduce in-consistency with the latest SR state.

A catalog table instance with the following attributes was cached into the memorygenerated each time #getTable is invoked:

• TableSchema: inferred from the Avro schema string
• Confluent schema registry URL
• The Avro schema string
• The schema registry subject name (actually with format {topic-name}-value, used for write)
• The Kafka topic name
• Format named identifier "confluentavro-registry-avrosr", this helps to find formats for Confluent Schema Registry Se/De
• Common properties for all the topics (as a parameter of the Catalog)

The graph below illustrates how this catalog works with the Confluent Schema Registry Service:

ConfluentRegistryAvroRowFormatFactory

ConfluentRegistryAvroRowFormatFactory is a factory for Confluent Schema Registry Avro Se/De formats. It has following attributes:

...

Parameters

The SchemaRegistryCatalog.Builder can be used to configure the following options:

...

Basic Auth Security for Producers and Consumers

See https://docs.confluent.io/current/schema-registry/serdes-develop/index.html#basic-auth-security-for-producers-and-consumers.

If you want to configure the client to enable SSL, use a custom SchemaRegistryClient when constructing the catalog.

RegistryAvroFormatFactory

RegistryAvroFormatFactory is a factory for Confluent Schema Registry Avro Se/De formats. It has following attributes:

• format: avro-sr: factory ID, required
• schema-registry.url: schema registry URL, required
• schema-string: avro schema string, required
• schema-registry.subject: subject to write to, required only for sink table

Public Interfaces

/**
 * Catalog for
 * <a href="https://docs.confluent.io/current/schema-registry/schema_registry_tutorial.html">Confluent Schema Registry</a>.
 * It allows to access all the topics of current Confluent Schema Registry Service
 * through SQL or TableAPI, there is no need to create any table explicitly.
 *
 * <p>The code snippet below illustrates how to use this catalog:
 * <pre>
 *      String schemaRegistryURL = ...;
 * 		Map<String, String> kafkaProps  ConfluentSchemaRegistryCatalog= ...;
 * 		SchemaRegistryCatalog catalog = new ConfluentSchemaRegistryCatalog(SchemaRegistryCatalog.builder()
 * 			properties,	.schemaRegistryURL(schemaRegistryURL)
 * 			schemaRegistryURL,	.kafkaOptions(kafkaProps)
 * 				.catalogName("catalog1myCatalog",)
 * 				.dbName("db1"myDB")
 * 				.build();
 * 		tEnv.registerCatalog("catalog1myCatalog", catalog);
 *
 * 		// ---------- Consume stream from Kafka -------------------
 *
 * 		// Assumes there is a topic named 'transactions'
 * 		String query = "SELECT\n" +
 * 			"  id, amount\n" +
 * 			"FROM catalog1myCatalog.db1myDB.transactions1transactions";
 * </pre>
 *
 * <p>We only support TopicNameStrategy for subject naming strategy,
 * for which all the records in one topic has the same schema, see
 * <a href="https://docs.confluent.io/current/schema-registry/serializer-formatter.html#how-the-naming-strategies-work">How the Naming Strategies Work</a>
 * for details.
 *
 * <p>You can specify some common options for these topics. All the tables from this catalog
 * would take the same options. If this is not your request, use dynamic table options setsetting up
 * within per-table scope.
 */
public class* ConfluentSchemaRegistryCatalog<p>The extendsbehaviors:
 AbstractCatalog* {
}

/** Format factory for confluent schema registry Avro to Row. */
public class ConfluentRegistryAvroRowFormatFactory extends TableFormatFactoryBase<Row>
		implements SerializationSchemaFactory<Row>, DeserializationSchemaFactory<Row> {
}

/** Confluent schema registry Avro row serialization schema.*/
public class ConfluentRegistryAvroRowSerializationSchema extends AvroRowSerializationSchema {
<ul>
 *     <li>The catalog only supports reading messages with the latest enabled schema for any given
 *     Kafka topic at the time when the SQL query was compiled.</li>
 *     <li>No time-column and watermark support.</li>
 *     <li>The catalog is read-only. It does not support table creations
 *     or deletions or modifications.</li>
 *     <li>The catalog only supports Kafka message values prefixed with schema id,
 *     this is also the default behaviour for the SchemaRegistry Kafka producer format.</li>
 * </ul>
 */
public class SchemaRegistryCatalog extends TableCatalog {}

/**
 * Confluent registry Avro row deserialization schema. Table format factory for providing configured instances of Schema Registry Avro to RowData
 * {@link SerializationSchema} and {@link DeserializationSchema}.
 */
public class ConfluentRegistryAvroRowDeserializationSchemaRegistryAvroFormatFactory implements
		DeserializationFormatFactory,
		extends AvroRowDeserializationSchemaSerializationFormatFactory {
}

The Expected Behaviors

• The catalog only supports reading messages with the latest enabled schema for any given Kafka topic at the time when the SQL query was compiled
• No time-column and watermark support
• The catalog is read-only. It does not support table creations or deletions or modifications
• The catalog only supports Kafka message values prefixed with schema id, this is also the default behavior for the SchemaRegistry Kafka producer format

The Table Schema and Watermark Definition

...