kcctl lets you manage all your Apache Kafka Connect instances in one command line tool, whether on-prem, Aiven, or other clouds. Find out how to use it.
Francesco Tisiot
|RSS FeedHead of Developer Experience at Aiven
Subscribe for the latest news and insights on open source, Aiven offerings, and more.
Apache Kafka is widely used as a company data backbone with Kafka Connect acting as a bridge. This way Kafka can be integrated with other technologies easily, reliably and scalably. The Kafka Connect REST APIs provide a way to manage connectors via web calls, but crafting URLs on a terminal can sometimes be tricky.
In this blog post we explore kcctl, a new open source command line tool for Kafka Connect. You'll find out how to integrate it with Apache Kafka and manage connections to other systems.
Aiven offers similar functionality with the Aiven Command Line Interface. You can use the CLI to create, drop, change, verify, pause and restore any Kafka Connect connector running on Aiven services. If all your Kafka Connect instances are Aiven services, then the Aiven command line interface is all you need.
But if you want to use the same tool for any Kafka Connect instance, be it on-premises, on Aiven, or in other cloud providers, then kcctl is your friend.
To start following the process in this article, make sure you already have an Apache Kafka environment with Kafka Connect up and running. If you don't have one, don't worry: Aiven can provide one in minutes. Just create one in the Aiven Console, or in the Aiven Command Line Interface with the following command:
Loading code...
This command creates an Aiven for Apache Kafka cluster named demo-kafka with three nodes (using the business-4 plan), in the google-europe-west3 region and enables the automatic creation of topics and Kafka Connect. With Aiven, you can deploy Kafka Connect as part of your Kafka cluster for business and premium plans, or as a separate, standalone cluster. To read more about Aiven for Apache Kafka and related Kafka Connect topics, check out the dedicated page.
Let's wait until the service is ready:
Loading code...
kcctlAt the time of writing, kcctl is an early access release. The current set of installation instructions can be found in its GitHub repository.
Once kcctl is installed, test that it's working in the terminal by adding the bin subdirectory to PATH and executing:
Loading code...
If we did everything correctly, then we should see the usage information. Now it's time to connect to our Kafka Connect cluster.
In order to plug in to Kafka Connect, first retrieve the cluster URL, which can be found using the via Aiven CLI and jq, to parse the JSON output:
Loading code...
Now create a kcctl configuration context by executing the following command, replacing the cluster parameter accordingly.
Loading code...
The above creates a context named my_kafka_cluster pointing to the demo-kafka instance. To verify the configuration:
Loading code...
This retrieves its definition of the current kcctl configuration context:
Loading code...
To see the connectors in action, create a PostgreSQL database and configure a Kafka Connect JDBC source connector to bring the data into Kafka. The connector takes data from a table named pasta stored in a PostgreSQL database and includes it in a Kafka topic.
If you don't have a PostgreSQL database handy, you can create one at Aiven with the following Aiven CLI command:
Loading code...
Once the demo-pg PostgreSQL instance is up and running (use avn service wait demo-pg to wait for it), connect to it:
Loading code...
Now create a sample pasta table and fill it with data, using the following statements in our terminal:
Loading code...
Once the source data is available, create the Kafka Connect JDBC source connector, sourcing in incremental mode the pasta table based on the id column. To get the required PostgreSQL connection details such as hostname, port, user and password, use this command:
Loading code...
Create a file named my_jdbc_connect_source.json with the following JSON content (substituting the <HOST>, <PORT> and <PASSWORD> with the actual information retrieved in the previous step):
Loading code...
Now invoke the connector creation via kcctl in a new terminal window:
Loading code...
Verify that the connector was successfully created:
Loading code...
The command output shows the pg-incremental-source connector in RUNNING state and all the details associated with it.
You can also check for a new Kafka topic called pg_source_pasta with the same data stored in PostgreSQL, via kcat. Start by first downloading the required certificates:
Loading code...
Then create a kcat.config file containing the following entries:
Loading code...
And reading from the pg_source_pasta topic with the following kcat invocation:
Loading code...
If we now insert some rows in the PostgreSQL pasta table, the same changes appear in Kafka via kcat:
Creating connectors is only part of the game with kcctl - You can also manage them! Need a list of all the connectors deployed? Just run the following command:
Loading code...
Need to pause and resume connectors? The code below, for example, pauses the one named pg-incremental-source:
Loading code...
What type of connectors can we create? Glad you asked! The full plugin list is available with:
Loading code...
The command shows all the connector plugins available with the related type (source or sink) and version. With this command you'll be able to check the list of the managed Kafka Connect connector types you can create with Aiven for Apache Kafka.
Loading code...
The ability to manage Kafka Connect connectors from the terminal is just a few commands away. kcctl makes it easy to inspect, deploy, update, pause and restore any connector to our environments. This unifies the end-user experience for Apache Kafka instances deployed on-premises, self-hosted or in Aiven.
Further reading
Not using Aiven services yet? Sign up now for your free trial at https://console.aiven.io/signup!
In the meantime, make sure you follow our changelog and blog RSS feeds or our LinkedIn and Twitter accounts to stay up-to-date with product and feature-related news.
avn service create demo-kafka \
--service-type kafka \
--cloud google-europe-west3 \
--plan business-4 \
-c kafka_connect=true \
-c kafka.auto_create_topics_enable=true avn service wait demo-kafkakcctlavn service get demo-kafka --json | jq '.connection_info.kafka_connect_uri'kcctl config set-context \
--cluster https://avnadmin:PASSWORD@demo-kafka-<PROJECT_NAME>.aivencloud.com:443 \
my_kafka_clusterkcctl infoURL: https://avnadmin:PASSWORD@demo-kafka-<PROJECT_NAME>.aivencloud.com:443
Version: 2.7.2-SNAPSHOT
Commit: d15ddddd3ef3f5ef
Kafka Cluster ID: -DvILyiXQxSpnFSK9M1qgQavn service create demo-pg \
--service-type pg \
--cloud google-europe-west3 \
--plan hobbyistavn service cli demo-pgcreate table pasta (id serial, name varchar, cooking_minutes int);
insert into pasta (name, cooking_minutes) values ('spaghetti', 8);
insert into pasta (name, cooking_minutes) values ('spaghettini', 6);
insert into pasta (name, cooking_minutes) values ('fusilli', 9);
insert into pasta (name, cooking_minutes) values ('trofie', 5);avn service get demo-pg --format '{service_uri_params}'{
"connector.class": "io.aiven.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:postgresql://<HOST>:<PORT>/defaultdb?sslmode=require",
"connection.user": "avnadmin",
"connection.password": "<PASSWORD>",
"table.whitelist": "pasta",
"mode": "incrementing",
"incrementing.column.name":"id",
"poll.interval.ms": "2000",
"topic.prefix": "pg_source_"
}kcctl apply -f my_jdbc_connect_source.json --name pg-incremental-sourcekcctl describe connector pg-incremental-sourceavn service user-creds-download demo-kafka \
--username avnadmin \
-d certsbootstrap.servers=<HOST>:<PORT>
security.protocol=ssl
ssl.key.location=certs/service.key
ssl.certificate.location=certs/service.cert
ssl.ca.location=certs/ca.pemkcat -F kcat.config -C -t pg_source_pastakcctl get connectorskcctl pause connector pg-incremental-sourcekcctl get pluginsTYPE CLASS VERSION
source com.couchbase.connect.kafka.CouchbaseSourceConnector 4.0.6
source com.datamountaineer.streamreactor.connect.cassandra.source.CassandraSourceCon 2.1.3
nector
source com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector 2.1.3
source com.google.pubsub.kafka.source.CloudPubSubSourceConnector 2.7.2-SNAPSHOT
source com.google.pubsublite.kafka.source.PubSubLiteSourceConnector 2.7.2-SNAPSHOT
...
sink io.aiven.kafka.connect.gcs.GcsSinkConnector 0.9.0
sink io.aiven.kafka.connect.http.HttpSinkConnector 0.4.0
sink io.aiven.kafka.connect.s3.AivenKafkaConnectS3SinkConnector 2.12.0
...