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

Discussion 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).

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.

This increases the size of messages significantly.

It is also inconvenient, as few applications will produce messages that look like this.

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "field": "id",
        "type": "string",
        "optional": false
      },
      {
        "field": "name",
        "type": "string",
        "optional": false
      }
    ]
  },
  "payload": {
    "id": "emp_001",
    "name": "Kevin"
  }
}

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

schemas.file.location

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 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:

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

and the Kafka message payload contains:

{
    "id": "emp_001",
    "name": "Kevin"
}

Existing parsing and interpretation of schemas will be identical to today.

The only difference is the ability to provide schemas externally to individual messages.

This option will only be used for toConnectData calls. It will be ignored for fromConnectData.

Proposed Changes

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

The configure method will be extended.

If the schemas.file.location  option is not set, the current behaviour will remain as-is.

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

byte[] schemadata = config.getSchemaContents();
JsonNode schemaNode = jsonDeserializer.deserialize("", schemadata);
schema = asConnectSchema(schemaNode);

If the file is not found, not readable, or the contents cannot be parsed as a Connect schema, this will throw an exception to alert the user to the configuration problem.

Compatibility, Deprecation, and Migration Plan

There is no compatibility impact from this proposal.

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.

Unit tests will cover error cases, such as missing schema files, or schemas that do not match the message format.

Rejected Alternatives

We are not proposing to automatically infer or generate schemas from message contents, as discussed in KIP-301

We are not proposing support for multiple different schemas, (such as to allow for different schemas for different messages). Where this level of complexity is required, it would be better to use a Converter that integrates with a schema registry, such as io.confluent.connect.json.JsonSchemaConverter  or io.apicurio.registry.utils.converter.ExtJsonConverter 

We are not proposing support for JSON Schema itself, as we prefer to maintain compatibility with the existing schemas used by JsonConverter.

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