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 DiscussionAccepted

Discussion thread: here

JIRA: here [Change the link from KAFKA-1 to your own ticket]13510

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

Motivation

When starting a connector, users must provide the connector configuration. The configuration often also includes configurations for other plugins such as SMTs or converters. Today, Connect does not provide a way to see what plugins are available installed apart from connectors (via the /connector-plugins endpoint). This make it difficult for users building data pipeline to know which plugins are available and what is possible. Basically they have to know how the Connect runtime is set up. Even once they know the plugins that are available, they then have to go look at the plugins documentation or, in the worst case, look directly at the source code to find their configuration definitions.
All
Connector plugins that are directly usable should be discoverable via the REST API. Their configuration definitions should also be easily retrieved. This would significantly ease the process of building pipelines and enable building tools and UIs that can manage Connect data pipelines.

Public Interfaces

The proposal is to add 2 new endpoints to address these limitations:

• /GET /connector-plugins: This endpoint will return all plugins that are be updated to allow listing all plugins. The response structure of the objects in the array remain unchanged. A new query parameter "connectorsOnly" will be added and it will default to true so it's fully compatible with the current behavior. Users will be able to list all Connectors, Transformations, Converters, HeaderConverters and Predicates . Plugins will be grouped by plugin.path. This will make it clear to users what's available to use as it's not possible to use a Connector from one path with Transformations from another.plugins by setting it to false. Classes that implement multiple plugin types will appear once for each type. For example SimpleHeaderConverter will be listed as a converter and as a header_converter. Possible values for the "type" field are "sink", "source", "converter", "header_converter", "transformation" and "predicate".

For example GET /connector-plugins?connectorsOnly=false will returnExample output with the runtime plugin.path set to plugin.path=/Users/mickael/tmp/path1/,/Users/mickael/tmp/path2/:

{
  "classpath": [
    {
      "class": "org.apache.kafka.connect.transformsfile.DropHeadersFileStreamSinkConnector",
      "type": "transformationsink",
      "locationversion": "classpath3.2.0"
    },
     {
      "class": "org.apache.kafka.connect.transformsfile.predicates.HasHeaderKeyFileStreamSourceConnector",
      "type": "predicatesource",
      "locationversion": "classpath3.2.0"
    },
     {
      "class": "org.apache.kafka.connect.converters.ByteArrayConverter",
      "type": "converter",
      "location": "classpath"
    },
    {
      "class": "org.apache.kafka.connect.storagetransforms.SimpleHeaderConverterCast$Value",
      "type": "header_convertertransformation",
      "location": "classpath"
    }
  ],
  "/Users/mickael/tmp/path1/": [
    {
      "class": "org.apache.kafka.connect.transforms.filepredicates.FileStreamSinkConnectorHasHeaderKey",
      "type": "sinkpredicate"
  },
  {
     "locationclass": "file:/Users/mickael/tmp/path1/connect-file-3.0.0.jar"
    }
  ],
  "/Users/mickael/tmp/path2/": [
    {
      org.apache.kafka.connect.storage.SimpleHeaderConverter",
    "type": "header_converter"
  },
  {
    "class": "comorg.githubapache.jcustenborder.kafka.connect.transformstorage.common.BytesToString$KeySimpleHeaderConverter",
      "type": "transformationconverter",
  },   
  "location": "file:/Users/mickael/tmp/path2/kafka-connect-transform-common-0.1.0.14.jar"
    }
  ]
}

...

]

Currently only Connector plugins are versioned, so we won't include the version field for other plugins.

• GET /connector-plugins/<plugin>/config: This new endpoint will return the configuration definitions of the specified plugin. It will work with all plugins returned by /connector_plugins.

The plugin can be specified via its fully qualified class name or its Connect alias like in the existing /connector-plugins/<plugin>/config/validate endpoint. If a plugin does not override the config() method, the response is an empty array.

For example, accessing http://localhost:8083/connector-plugins/transformation/org.apache.kafka.connect.transforms.Cast$Value/configdefconfig will return:

[
  {
    "name": "spec",
    "type": "LIST",
    "required": true,
    "default_value": null,
    "importance": "HIGH",
    "documentation": "List of fields and the type to cast them to of the form field1:type,field2:type to cast fields of Maps or Structs. A single type to cast the entire value. Valid types are int8, int16, int32, int64, float32, float64, boolean, and string. Note that binary fields can only be cast to string.",
    "group": null,
    "width": "NONE",
    "display_name": "spec",
    "dependents": [],
    "order": -1
  }
]

This will reuse the ConfigKeyInfo entity which is already exposed via PUT /connector-plugins/{connector-type}/config/validate.

Proposed Changes

REST API:

• A new

...

• path will be added to ConnectorPluginsResource to retrieve the plugin configuration definitions

@GET
@Path("/{plugin}/config")
public Map<String, List<PluginInfo>> listPluginsList<ConfigKeyInfo> getPluginConfig() {}

• Listing connector plugin will accept an optional query parameter "connectorsOnly"  that defaults to true

    return doListPlugins();
}

@GET
@Path("/{type}/{plugin}/configdef")
public List<ConfigKeyInfo>List<ConnectorPluginInfo> getConnectorConfigDeflistConnectorPlugins(final @PathParam@DefaultValue("typetrue") String type, final @PathParam@QueryParam("pluginconnectorsOnly") Stringboolean pluginconnectorsOnly) {
    return doGetConfigDef(type, plugin);
}

Converter interface:

Add a config() method to Converter. Also make it Configurable and Closeable so it's uniform with the other plugins: with a default implementation.

public interface Converter extends Configurable, Closeable {

[...]

    /**
     * Configuration specification for this set of converters.
     * @return the configuration specification; may not be null
     */
    default ConfigDef config() {
        return new ConfigDef();
    }
}

It's common for custom converters to implement both Converter and HeaderConverter. As the 2 methods to retrieve the ConfigDef will have exactly the same signature, it will still be possible to implement both interfaces.

Compatibility, Deprecation, and Migration Plan

...

This KIP is proposing new endpoints, existing endpoints are not changed.

Rejected Alternatives

...

• /connector-plugins keeps its current behavior and will only expose the new behavior when a new query parameter is set.
• When accessing /connector-plugins/<plugin>/config on existing converters that don't implement the config() method, an empty array will be returned. If a converter is also implementing HeaderConverter, and hence already have a config() method, it will be automatically used and the config will be returned.
• /connector-plugins/<plugin>/config is a new endpoint that doesn't cause compatibility issues.

I propose to flip the query parameter value to list all plugins by default in the next major release.

Rejected Alternatives

• Add a new endpoint /plugins for listing all plugins: It would be confusing to list both worker and connector plugins together. We'd then end up with 3 endpoints, /plugins, /worker-plugins and /connector-plugins which is as confusing!
• Group connectors by type when listing them: This would break compatibility with the existing /connector-plugins behavior. As it's a very commonly used endpoint, it's preferred to keep compatibility.
• Add a new endpoint /worker-plugins to list worker plugins (Rest Extensions and Config Providers): The use case is to allow administrators to check the plugins installed in each worker. Connect shouldn't expose worker internal details to all users and it's not clear what information would be useful for admins. Also Connect already has a /admin endpoint which should be reused for admin tasks.
• Make all plugins implement Versioned. Initially we wanted to make all plugins consistent, but this either force having a default implementation for version() which would allow Connectors to not implement it, or force introducing another interface (PossiblyVersioned) to version other plugins which did not make a lot of sense since version does not have any contract today.