Status

Current state: Under Discussion  Accepted

Discussion thread:  here

Vote thread: here

JIRA: 16913 

...

schemasschema.file.locationcontent

If a value for this is not provided, the current behaviour (reading schemas from the message payload) is followed.

If a value is provided, it is will be used as the location on disk of a text file containing a schema.

For example, you could use the config:

schemas.enable: =true
schemas.file.location: /some/filelocation/schema.json

where /some/filelocation/schema.json contains:

{
    schema.content={"type": "struct",
    "fields": [
      {
        "field": "id",
        "type": "string",
        "optional": false
      },
      {
        "field": "name",
        "type": "string",
        "optional": false
      }
    ]
  }

and the Kafka message payload contains:

...

This option will only be used for toConnectData calls. It will be ignored for fromConnectData , this means that this convertor will be explicitly used for sink connector not source connector

Users would be able use ConfigProviders to provide the contents of a file as a schema.

For Example, 

# Config for Kafka Connect
config.providers=directory
config.providers.directory.class=org.apache.kafka.common.config.provider.DirectoryConfigProvider

# Config for Kafka Convertor
schema.content=${directory:/schema:schema.json}

KIP-993: Allow restricting files accessed by File and Directory ConfigProviders will allow users to restrict what files to be used.

While using Connect REST API to create Connector with Schema Configuration, we will need include escape sequences in the schema.content field since REST API payload needs to be a JSON.

For Example

{
  "name": "api-file-source-json",
  "config": {
    "value.converter.schemas.enable": true,
    "value.converter.schema.content": "{\"type\": \"struct\", \"fields\": [{ \"field\": \"id\", \"type\": \"string\", \"optional\": false },{ \"field\": \"name\", \"type\": \"string\", \"optional\": false }]}"
  }
}

Proposed Changes

Functional changes will be made to org.apache.kafka.connect.json.JsonConverter  configure

The configure method will be extended.

If the schemasschema.file.locationcontent  option is not set, the current behaviour will remain as-is.

If the schemasschema.file.locationcontent  option is set, the schema file will be read and an instance variable will store the Connect schema to use.

...

We are not proposing to implement changes in fromConnectData, such as to write the schema found in the message to a file.

We are not proposing to create a custom JsonConverter, since we would have to duplicate a lot of code from this existing Json Converter (private methods etc.) for the new convertor to work.