Schema Registry formats

You can avoid defining the Avro and JSON schema for Kafka table sources and sinks, when the schema is stored in Cloudera Schema Registry.

Such topics are accessible through automatically generated tables from the read-only registry catalog.

Maven dependency

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-cloudera-registry</artifactId>
  <version>1.0-csa1.13.1.0</version>
</dependency>

If you need to define a table outside the registry catalog, the following example can be used:

CREATE TABLE source_table (
   id BIGINT,
   name STRING,
   description STRING
) WITH (
   'connector.type' = 'kafka',
   'connector.version' = 'universal',
   'connector.topic' = 'message',
   'connector.startup-mode' = 'latest-offset',
   'connector.properties.bootstrap.servers' = '<hostname>:<port>',
   'connector.properties.group.id' = 'test',
   'format.type' = 'registry',
   'format.registry.properties.schema.registry.url' = 'http(s)://<hostname>:<port>/api/v1'
)

Cloudera Schema Registry connector for Flink stores the schema version info in the Kafka messages by default. This means that the format.registry.properties.store.schema.version.id.in.header property is set to false by default.

The schema name in the registry is usually the same as the Kafka topic name, but can be overridden by the format.registry.schema-name property.