10.3. Managing connectors
The Kafka Connect REST API provides endpoints for creating, updating, and deleting connectors directly. You can also use the API to check the status of connectors or change logging levels. When you create a connector through the API, you provide the configuration details for the connector as part of the API call.
You can also add and manage connectors as plugins. Plugins are packaged as JAR files that contain the classes to implement the connectors through the Kafka Connect API. You just need to specify the plugin in the classpath or add it to a plugin path for Kafka Connect to run the connector plugin on startup.
In addition to using the Kafka Connect REST API or plugins to manage connectors, you can also add connector configuration using properties files when running Kafka Connect in standalone mode. To do this, you simply specify the location of the properties file when starting the Kafka Connect worker process. The properties file should contain the configuration details for the connector, including the connector class, source and destination topics, and any required authentication or serialization settings.
10.3.1. Limiting access to the Kafka Connect API
The Kafka Connect REST API can be accessed by anyone who has authenticated access and knows the endpoint URL, which includes the hostname/IP address and port number. It is crucial to restrict access to the Kafka Connect API only to trusted users to prevent unauthorized actions and potential security issues.
For improved security, we recommend configuring the following properties for the Kafka Connect API:
-
(Kafka 3.4 or later)
org.apache.kafka.disallowed.login.modulesto specifically exclude insecure login modules -
connector.client.config.override.policyset toNONEto prevent connector configurations from overriding the Kafka Connect configuration and the consumers and producers it uses
10.3.2. Configuring connectors
Use the Kafka Connect REST API or properties files to create, manage, and monitor connector instances. You can use the REST API when using Kafka Connect in standalone or distributed mode. You can use properties files when using Kafka Connect in standalone mode.
When using the Kafka Connect REST API, you can create connectors dynamically by sending PUT or POST HTTP requests to the Kafka Connect REST API, specifying the connector configuration details in the request body.
When you use the PUT command, it’s the same command for starting and updating connectors.
The REST interface listens on port 8083 by default and supports the following endpoints:
GET /connectors- Return a list of existing connectors.
POST /connectors- Create a connector. The request body has to be a JSON object with the connector configuration.
GET /connectors/<connector_name>- Get information about a specific connector.
GET /connectors/<connector_name>/config- Get configuration of a specific connector.
PUT /connectors/<connector_name>/config- Update the configuration of a specific connector.
GET /connectors/<connector_name>/status- Get the status of a specific connector.
GET /connectors/<connector_name>/tasks- Get a list of tasks for a specific connector
GET /connectors/<connector_name>/tasks/<task_id>/status- Get the status of a task for a specific connector
PUT /connectors/<connector_name>/pause- Pause the connector and all its tasks. The connector will stop processing any messages.
PUT /connectors/<connector_name>/stop- Stop the connector and all its tasks. The connector will stop processing any messages. Stopping a connector from running may be more suitable for longer durations than just pausing.
PUT /connectors/<connector_name>/resume- Resume a paused connector.
POST /connectors/<connector_name>/restart- Restart a connector in case it has failed.
POST /connectors/<connector_name>/tasks/<task_id>/restart- Restart a specific task.
DELETE /connectors/<connector_name>- Delete a connector.
GET /connectors/<connector_name>/topics- Get the topics for a specific connector.
PUT /connectors/<connector_name>/topics/reset- Empty the set of active topics for a specific connector.
GET /connectors/<connector_name>/offsets- Get the current offsets for a connector.
DELETE /connectors/<connector_name>/offsets- Reset the offsets for a connector, which must be in a stopped state.
PATCH /connectors/<connector_name>/offsets-
Adjust the offsets (using an
offsetproperty in the request) for a connector, which must be in a stopped state. GET /connector-plugins- Get a list of all supported connector plugins.
GET /connector-plugins/<connector_plugin_type>/config- Get the configuration for a connector plugin.
PUT /connector-plugins/<connector_type>/config/validate- Validate connector configuration.
10.3.2.2. Specifying connector configuration properties
To configure a Kafka Connect connector, you need to specify the configuration details for source or sink connectors. There are two ways to do this: through the Kafka Connect REST API, using JSON to provide the configuration, or by using properties files to define the configuration properties. The specific configuration options available for each type of connector may differ, but both methods provide a flexible way to specify the necessary settings.
The following options apply to all connectors:
name- The name of the connector, which must be unique within the current Kafka Connect instance.
connector.class-
The class of the connector plug-in. For example,
org.apache.kafka.connect.file.FileStreamSinkConnector. tasks.max- The maximum number of tasks that the specified connector can use. Tasks enable the connector to perform work in parallel. The connector might create fewer tasks than specified.
key.converter-
The class used to convert message keys to and from Kafka format. This overrides the default value set by the Kafka Connect configuration. For example,
org.apache.kafka.connect.json.JsonConverter. value.converter-
The class used to convert message payloads to and from Kafka format. This overrides the default value set by the Kafka Connect configuration. For example,
org.apache.kafka.connect.json.JsonConverter.
You must set at least one of the following options for sink connectors:
topics- A comma-separated list of topics used as input.
topics.regex- A Java regular expression of topics used as input.
For all other options, see the connector properties in the Apache Kafka documentation.
Streams for Apache Kafka includes the example connector configuration files config/connect-file-sink.properties and config/connect-file-source.properties in the Streams for Apache Kafka installation directory.
10.3.3. Creating connectors using the Kafka Connect API
Use the Kafka Connect REST API to create a connector to use with Kafka Connect.
Prerequisites
- A Kafka Connect installation.
Procedure
Prepare a JSON payload with the connector configuration. For example:
{ "name": "my-connector", "config": { "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector", "tasks.max": "1", "topics": "my-topic-1,my-topic-2", "file": "/tmp/output-file.txt" } }Send a POST request to
<KafkaConnectAddress>:8083/connectorsto create the connector. The following example usescurl:curl -X POST -H "Content-Type: application/json" --data @sink-connector.json http://connect0.my-domain.com:8083/connectorsVerify that the connector was deployed by sending a GET request to
<KafkaConnectAddress>:8083/connectors. The following example usescurl:curl http://connect0.my-domain.com:8083/connectors
10.3.4. Deleting connectors using the Kafka Connect API
Use the Kafka Connect REST API to delete a connector from Kafka Connect.
Prerequisites
- A Kafka Connect installation.
Deleting connectors
Verify that the connector exists by sending a
GETrequest to<KafkaConnectAddress>:8083/connectors/<ConnectorName>. The following example usescurl:curl http://connect0.my-domain.com:8083/connectorsTo delete the connector, send a
DELETErequest to<KafkaConnectAddress>:8083/connectors. The following example usescurl:curl -X DELETE http://connect0.my-domain.com:8083/connectors/my-connectorVerify that the connector was deleted by sending a GET request to
<KafkaConnectAddress>:8083/connectors. The following example usescurl:curl http://connect0.my-domain.com:8083/connectors
10.3.5. Adding connector plugins
Kafka provides example connectors to use as a starting point for developing connectors. The following example connectors are included with Streams for Apache Kafka:
- FileStreamSink
- Reads data from Kafka topics and writes the data to a file.
- FileStreamSource
- Reads data from a file and sends the data to Kafka topics.
Both connectors are contained in the libs/connect-file-<kafka_version>.redhat-<build>.jar plugin.
To use the connector plugins in Kafka Connect, you can add them to the classpath or specify a plugin path in the Kafka Connect properties file and copy the plugins to the location.
Specifying the example connectors in the classpath
CLASSPATH=/opt/kafka/libs/connect-file-<kafka_version>.redhat-<build>.jar opt/kafka/bin/connect-distributed.shSetting a plugin path
plugin.path=/opt/kafka/connector-plugins,/opt/connectors
The plugin.path configuration option can contain a comma-separated list of paths.
You can add more connector plugins if needed. Kafka Connect searches for and runs connector plugins at startup.
When running Kafka Connect in distributed mode, plugins must be made available on all worker nodes.
'%20d='M201.5%20305.5c-13.8%200-24.9-11.1-24.9-24.6%200-13.8%2011.1-24.9%2024.9-24.9%2013.6%200%2024.6%2011.1%2024.6%2024.9%200%2013.6-11.1%2024.6-24.6%2024.6zM504%20256c0%20137-111%20248-248%20248S8%20393%208%20256%20119%208%20256%208s248%20111%20248%20248zm-132.3-41.2c-9.4%200-17.7%203.9-23.8%2010-22.4-15.5-52.6-25.5-86.1-26.6l17.4-78.3%2055.4%2012.5c0%2013.6%2011.1%2024.6%2024.6%2024.6%2013.8%200%2024.9-11.3%2024.9-24.9s-11.1-24.9-24.9-24.9c-9.7%200-18%205.8-22.1%2013.8l-61.2-13.6c-3-.8-6.1%201.4-6.9%204.4l-19.1%2086.4c-33.2%201.4-63.1%2011.3-85.5%2026.8-6.1-6.4-14.7-10.2-24.1-10.2-34.9%200-46.3%2046.9-14.4%2062.8-1.1%205-1.7%2010.2-1.7%2015.5%200%2052.6%2059.2%2095.2%20132%2095.2%2073.1%200%20132.3-42.6%20132.3-95.2%200-5.3-.6-10.8-1.9-15.8%2031.3-16%2019.8-62.5-14.9-62.5zM302.8%20331c-18.2%2018.2-76.1%2017.9-93.6%200-2.2-2.2-6.1-2.2-8.3%200-2.5%202.5-2.5%206.4%200%208.6%2022.8%2022.8%2087.3%2022.8%20110.2%200%202.5-2.2%202.5-6.1%200-8.6-2.2-2.2-6.1-2.2-8.3%200zm7.7-75c-13.6%200-24.6%2011.1-24.6%2024.9%200%2013.6%2011.1%2024.6%2024.6%2024.6%2013.8%200%2024.9-11.1%2024.9-24.6%200-13.8-11-24.9-24.9-24.9z'/%3e%3c/svg%3e){kind=small}