A client for the Confluent Platform Kafka Connect REST API.
Project description
Kafka Connect Python
The Kafka Connect REST API allows you to manage connectors that move data between Apache Kafka and other systems.
The Kafka Connect command line tool, also known as kc
or kafka-connect
, allows users to manage their Kafka Connect cluster and connectors. With this tool, users can retrieve information about the cluster and connectors, create new connectors, update existing connectors, delete connectors, and perform other actions.
This project aims to supported all features of the Kafka Connect REST API.
Install
pip install kafka-connect-py
Command Line Usage
Getting Basic Connect Cluster Information
To get basic Connect cluster information including the worker version, the commit it’s on, and its Kafka cluster ID, use the following command:
kc info
Listing Installed Plugins
To list the plugins installed on the worker, use the following command:
kc list-plugins
To format the result of the installed plugin list for easier readability, pipe the output to the jq
command:
kc list-plugins | jq
Create a Connector Instance
To create a connector instance with JSON data containing the connector’s configuration:
kc update source-debezium-orders-00 -d '{
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"value.converter": "io.confluent.connect.json.JsonSchemaConverter",
"value.converter.schemas.enable": "true",
"value.converter.schema.registry.url": "'$SCHEMA_REGISTRY_URL'",
"value.converter.basic.auth.credentials.source": "'$BASIC_AUTH_CREDENTIALS_SOURCE'",
"value.converter.basic.auth.user.info": "'$SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO'",
"database.hostname": "mysql",
"database.port": "3306",
"database.user": "debezium",
"database.password": "dbz",
"database.server.id": "42",
"database.server.name": "asgard",
"table.whitelist": "demo.orders",
"database.history.kafka.bootstrap.servers": "'$BOOTSTRAP_SERVERS'",
"database.history.consumer.security.protocol": "SASL_SSL",
"database.history.consumer.sasl.mechanism": "PLAIN",
"database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"'$CLOUD_KEY'\" password=\"'$CLOUD_SECRET'\";",
"database.history.producer.security.protocol": "SASL_SSL",
"database.history.producer.sasl.mechanism": "PLAIN",
"database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"'$CLOUD_KEY'\" password=\"'$CLOUD_SECRET'\";",
"database.history.kafka.topic": "dbhistory.demo",
"topic.creation.default.replication.factor": "3",
"topic.creation.default.partitions": "3",
"decimal.handling.mode": "double",
"include.schema.changes": "true",
"transforms": "unwrap,addTopicPrefix",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.addTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
"transforms.addTopicPrefix.regex":"(.*)",
"transforms.addTopicPrefix.replacement":"mysql-debezium-$1"
}'
Or create/update a connector instance with a JSON file:
kc update <connector> --config-file <config_file>
Update a Connector
As mentioned above, if there’s a connector to update, you can use the update
sub-command to amend the configuration (see Create a Connector Instance above). Because update is used to both create and update connectors, it’s the standard command that you should use most of the time (which also means that you don’t have to completely rewrite your configs).
List Connector Instances
Use the following command to list of all extant connectors:
kc list [--expand=info|status] [--pattern=regex] [--state=running|paused|unassigned|failed]
Inspect Config and Status for a Connector
Inspect the config for a given connector as follows:
kc config sink-elastic-orders-00
You can also look at a connector’s status. While the config command shows a connector’s static configuration, the status shows the connector as a runtime entity:
kc status sink-elastic-orders-00
You can also use list
with the --expand=status
option to show the status of many connectors at once. We can filter down the response using a regex pattern and/or connector state.
Use the following to show all connector names prefixed with the word sink-
and that are in a FAILED
connector state.
kc list --expand=status -p sink-.* -s failed
Delete a Connector
If something is wrong in your setup and you don’t think a config change would help, or if you simply don’t need a connector to run anymore, you can delete it by name:
kc delete sink-elastic-orders-00
The delete
sub-command also supports multiple deletions using the --all
option. On its own it will apply the sub-command to all connectors.
The following will delete all connector names prefixed with the word sink-
and that are in a PAUSED
connector state.
kc delete --all --pattern sink-.* -s paused
The --all
option is supported by several sub-commands, including delete
, restart
, resume
, and pause
. However, for better testing and control over the outcome of your actions, we recommend using the list filtering option before executing any of these sub-commands. This way, you can ensure that your filters are working as intended and avoid unintended consequences. To use list filtering, simply run the list
sub-command and apply your filters.
Inspect Task Details
The following command returns the connector status:
kc status source-debezium-orders-00 | jq
If your connector fails, the details of the failure belong to the task. So to inspect the problem, you’ll need to find the stack trace for the task. The task is the entity that is actually running the connector and converter code, so the state for the stack trace lives in it.
kc task-status source-debezium-orders-00 <task-id> | jq
Restart the Connector and Tasks
If after inspecting a task, you have determined that it has failed and you have fixed the reason for the failure (perhaps restarted a database), you can restart the connector with the following:
kc restart source-debezium-orders-00
Keep in mind though that restarting the connector doesn’t restart all of its tasks. You will also need to restart the failed task and then get its status again as follows:
kc task-status source-debezium-orders-00 <task-id>
What's more, you can restart the connector and all its failed tasks with the following:
kc restart source-debezium-orders-00 --include-tasks --failed-only
and check the status again:
kc status source-debezium-orders-00 | jq
Pause and Resume a Connector
Unlike restarting, pausing a connector does pause its tasks. This happens asynchronously, though, so when you pause a connector, you can’t rely on it pausing all of its tasks at exactly the same time. The tasks are running in a thread pool, so there’s no fancy mechanism to make this happen simultaneously.
A connector and its tasks can be paused as follows:
kc pause source-debezium-orders-00
Just as easily, a connector and its tasks can be resumed:
kc resume source-debezium-orders-00
Display All of a Connector’s Tasks
A convenient way to display all of a connector’s tasks at once is as follows:
kc list-tasks source-debezium-orders-00 | jq
This information is similar to what you can get from other APIs, but it is broken down by task, and configs for each are shown. Get a List of Topics Used by a Connector
As of Apache Kafka 2.5, it is possible to get a list of topics used by a connector:
kc list-topics source-debezium-orders-00 | jq
This shows the topics that a connector is consuming from or producing to. This may not be particularly useful for connectors that are consuming from or producing to a single topic. However, some developers, for example, use regular expressions for topic names in Connect, so this is a major benefit in situations where topic names are derived computationally.
This could also be useful with a source connector that is using SMTs to dynamically change the topic names to which it is producing.
Python
# Import the class
from kafka_connect import KafkaConnect
import json
# Instantiate the client
client = KafkaConnect(url="http://localhost:8083")
# Get the version and other details of the Kafka Connect cluster
cluster = client.get_cluster_info()
print(cluster)
# Get a list of active connectors
connectors = client.list_connectors(expand="status")
print(json.dumps(connectors, indent=2))
# Create a new connector
config = {
"name": "my-connector",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "1",
"connection.url": "jdbc:postgresql://localhost:5432/mydatabase",
"connection.user": "myuser",
"connection.password": "mypassword",
"table.whitelist": "mytable",
"mode": "timestamp+incrementing",
"timestamp.column.name": "modified_at",
"validate.non.null": "false",
"incrementing.column.name": "id",
"topic.prefix": "my-connector-",
},
}
response = client.create_connector(config)
print(response)
# Update an existing connector
new_config = {
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "1",
"connection.url": "jdbc:postgresql://localhost:5432/mydatabase",
"connection.user": "myuser",
"connection.password": "mypassword",
"table.whitelist": "mytable",
"mode": "timestamp+incrementing",
"timestamp.column.name": "modified_at",
"validate.non.null": "false",
"incrementing.column.name": "id",
"topic.prefix": "my-connector-",
},
}
response = client.update_connector("my-connector", new_config)
print(response)
# Get status for a connector
response = client.get_connector_status("my-connector")
print(json.dumps(response, indent=2))
# Restart a connector
response = client.restart_connector("my-connector")
print(response)
# Delete a connector
response = client.delete_connector("my-connector")
print(response)
License
Apache 2.0 License - aidanmelen/kafka-connect-py
Credits
The entire Command Line Usage section was copied directly from the Confluence's Kafka Connect’s REST API course.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for kafka_connect_py-0.10.11-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | ad43723b61a96bce9251a8c27691085ed598011b02458651bd52a9577cd12886 |
|
MD5 | 0b52a1a165d4370f61990573f1b88f6e |
|
BLAKE2b-256 | c92f24cc278051ac790141f471c588200de1da9de7ecc3ef3bb806ddebfdc670 |