This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under Discussion  Accepted

Discussion thread:  here

Vote thread: here[Change the link from the KIP proposal email archive to your own email thread]

JIRA:  [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast). 16913 

Motivation

When using a Connector that requires a schema (such as JDBC connectors) with JSON messages, the current JSONConverter requires a copy of the schema to be included in every message.

...

The ideal alternative (using a different JSON converter, paired with a schema registry) is difficult to adopt in some situations, such as where a schema registry is not available, or the Kafka message producers cannot be modified to include a reference or ID for the appropriate schema.

Public Interfaces

This proposal introduces a new optional config option for org.apache.kafka.connect.json.JsonConverter

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.

...

The new config option is optional, and if not provided, the existing behaviour remains unmodified.

Test Plan

Unit tests will cover successful cases, including a variety of schema types.

...

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.