Add key to data ingested through Kafka Connect

Question:

How can I stream data from a source system (such as a database) into Kafka using Kafka Connect, and add a key to the data as part of the ingest?

Edit this page

Example use case:

Kafka Connect is the integration API for Apache Kafka. It enables you to stream data from source systems (such databases, message queues, SaaS platforms, and flat files) into Kafka, and from Kafka to target systems. When you stream data into Kafka you often need to set the key correctly for partitioning and application logic reasons. In this example there is data about cities in a database, and we want to key the resulting Kafka message by the city_id field. There are different ways to set the key correctly and these tutorials will show you how. It will also cover how to declare the schema and use Kafka Streams to process the data using SpecificAvro.

Code example:

Try it

1
Initialize the project

To get started, make a new directory anywhere you’d like for this project:

mkdir connect-add-key-to-source && cd connect-add-key-to-source

Then make the following directories to set up its structure:

mkdir src 

2
Prepare the source data

Create a file cities.sql with commands to pre-populate the database table with city information:

DROP TABLE IF EXISTS cities;
CREATE TABLE cities (city_id INTEGER PRIMARY KEY NOT NULL, name VARCHAR(255), state VARCHAR(255));
INSERT INTO cities (city_id, name, state) VALUES (1, 'Raleigh', 'NC');
INSERT INTO cities (city_id, name, state) VALUES (2, 'Mountain View', 'CA');
INSERT INTO cities (city_id, name, state) VALUES (3, 'Knoxville', 'TN');
INSERT INTO cities (city_id, name, state) VALUES (4, 'Houston', 'TX');
INSERT INTO cities (city_id, name, state) VALUES (5, 'Olympia', 'WA');
INSERT INTO cities (city_id, name, state) VALUES (6, 'Bismarck', 'ND');

3
Get Confluent Platform

Next, create the following docker-compose.yml file to obtain Confluent Platform. Make sure that you create this file in the same place as the cities.sql file that you created above.

---
version: '2'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka:5.5.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  schema-registry:
    image: confluentinc/cp-schema-registry:5.5.0
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - zookeeper
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'

  ksqldb-server:
    image: confluentinc/ksqldb-server:0.11.0
    hostname: ksqldb
    container_name: ksqldb
    depends_on:
      - broker
    ports:
      - "8088:8088"
    environment:
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_BOOTSTRAP_SERVERS: broker:9092
      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
      KSQL_KSQL_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      KSQL_KSQL_SERVICE_ID: confluent_rmoff_01
      KSQL_KSQL_HIDDEN_TOPICS: '^_.*'
      # Setting KSQL_KSQL_CONNECT_WORKER_CONFIG enables embedded Kafka Connect
      KSQL_KSQL_CONNECT_WORKER_CONFIG: "/connect/connect.properties"
      # Kafka Connect config below
      KSQL_CONNECT_BOOTSTRAP_SERVERS: "broker:9092"
      KSQL_CONNECT_REST_ADVERTISED_HOST_NAME: 'ksqldb'
      KSQL_CONNECT_REST_PORT: 8083
      KSQL_CONNECT_GROUP_ID: ksqldb-kafka-connect-group-01
      KSQL_CONNECT_CONFIG_STORAGE_TOPIC: _ksqldb-kafka-connect-group-01-configs
      KSQL_CONNECT_OFFSET_STORAGE_TOPIC: _ksqldb-kafka-connect-group-01-offsets
      KSQL_CONNECT_STATUS_STORAGE_TOPIC: _ksqldb-kafka-connect-group-01-status
      KSQL_CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      KSQL_CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      KSQL_CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      KSQL_CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      KSQL_CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: '1'
      KSQL_CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: '1'
      KSQL_CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: '1'
      KSQL_CONNECT_LOG4J_APPENDER_STDOUT_LAYOUT_CONVERSIONPATTERN: "[%d] %p %X{connector.context}%m (%c:%L)%n"
      KSQL_CONNECT_PLUGIN_PATH: '/home/appuser/share/java,/home/appuser/confluent-hub-components/,/data/connect-jars'
    command:
      # In the command section, $ are replaced with $$ to avoid the error 'Invalid interpolation format for "command" option'
      - bash
      - -c
      - |
        echo "Installing connector plugins"
        mkdir -p ~/confluent-hub-components/
        /home/appuser/bin/confluent-hub install --no-prompt --component-dir confluent-hub-components/ confluentinc/kafka-connect-jdbc:5.4.1
        #
        echo "Launching ksqlDB"
        /usr/bin/docker/run &
        #
        sleep infinity

  ksqldb-cli:
    image: confluentinc/ksqldb-cli:0.11.0
    container_name: ksqldb-cli
    depends_on:
      - broker
      - ksqldb-server
    entrypoint: /bin/sh
    tty: true
    environment:
      KSQL_CONFIG_DIR: "/etc/ksqldb"
    volumes:
      - ./src:/opt/app/src

  kafkacat:
    image: edenhill/kafkacat:1.5.0
    container_name: kafkacat
    links:
      - broker
    entrypoint:
      - /bin/sh
      - -c
      - |
        apk add jq;
        while [ 1 -eq 1 ];do sleep 60;done

  postgres:
    # *-----------------------------*
    # To connect to the DB:
    #   docker exec -it postgres bash -c 'psql -U $POSTGRES_USER $POSTGRES_DB'
    # *-----------------------------*
    image: postgres:11
    container_name: postgres
    environment:
     - POSTGRES_USER=postgres
     - POSTGRES_PASSWORD=postgres
    volumes:
     - ./cities.sql:/docker-entrypoint-initdb.d/cities.sql

Now launch Confluent Platform by running:

docker-compose up -d

4
Check the source data

Check the data in the source database. Observe the city_id primary key:

echo 'SELECT * FROM cities;' | docker exec -i postgres bash -c 'psql -U $POSTGRES_USER $POSTGRES_DB'
 city_id |     name      | state
---------+---------------+-------
       1 | Raleigh       | NC
       2 | Mountain View | CA
       3 | Knoxville     | TN
       4 | Houston       | TX
       5 | Olympia       | WA
       6 | Bismarck      | ND
(6 rows)

5
Create the connector

Launch the ksqlDB CLI:

docker exec -it ksqldb-cli ksql http://ksqldb-server:8088

From the ksqlDB prompt you can create the JDBC source connector. There are a couple of points to note:

  1. The transforms stanza, which is responsible for setting the key to the value of the city_id field. They run in the order defined by transforms:

    • - copyFieldToKey sets the key to a struct containing the city_id field from the value.

    • - extractKeyFromStruct sets the key to just the city_id field of the struct set by the previous step.

    • - removeKeyFromValue removes the city_id from the message value, as it’s now stored in the message key.

  2. Since the key is an integer we override the default serialization and instead use the IntegerConverter for the key field

CREATE SOURCE CONNECTOR JDBC_SOURCE_POSTGRES_01 WITH (
    'connector.class'= 'io.confluent.connect.jdbc.JdbcSourceConnector',
    'connection.url'= 'jdbc:postgresql://postgres:5432/postgres',
    'connection.user'= 'postgres',
    'connection.password'= 'postgres',
    'mode'= 'incrementing',
    'incrementing.column.name'= 'city_id',
    'topic.prefix'= 'postgres_',
    'transforms'= 'copyFieldToKey,extractKeyFromStruct,removeKeyFromValue',
    'transforms.copyFieldToKey.type'= 'org.apache.kafka.connect.transforms.ValueToKey',
    'transforms.copyFieldToKey.fields'= 'city_id',
    'transforms.extractKeyFromStruct.type'= 'org.apache.kafka.connect.transforms.ExtractField$Key',
    'transforms.extractKeyFromStruct.field'= 'city_id',
    'transforms.removeKeyFromValue.type'= 'org.apache.kafka.connect.transforms.ReplaceField$Value',
    'transforms.removeKeyFromValue.blacklist'= 'city_id',
    'key.converter' = 'org.apache.kafka.connect.converters.IntegerConverter'
);

Check that the connector is running:

SHOW CONNECTORS;

You should see that the state is RUNNING:

 Connector Name          | Type   | Class                                         | Status
----------------------------------------------------------------------------------------------------------------
 JDBC_SOURCE_POSTGRES_01 | SOURCE | io.confluent.connect.jdbc.JdbcSourceConnector | RUNNING (1/1 tasks RUNNING)
----------------------------------------------------------------------------------------------------------------

You can also inspect further details about the connector including to which topics it is writing:

DESCRIBE CONNECTOR JDBC_SOURCE_POSTGRES_01;
Name                 : JDBC_SOURCE_POSTGRES_01
Class                : io.confluent.connect.jdbc.JdbcSourceConnector
Type                 : source
State                : RUNNING
WorkerId             : ksqldb:8083

 Task ID | State   | Error Trace
---------------------------------
 0       | RUNNING |
---------------------------------

 Related Topics
-----------------
 postgres_cities
-----------------

6
Consume events from the output topic

With the connector running let’s now inspect the data on the Kafka topic. ksqlDB’s PRINT command will show the contents of a topic:

PRINT postgres_cities FROM BEGINNING LIMIT 6;

The output should resemble:

Key format: KAFKA_INT or KAFKA_STRING
Value format: AVRO or KAFKA_STRING
rowtime: 3/25/20 11:53:36 AM UTC, key: 1, value: {"name": "Raleigh", "state": "NC"}
rowtime: 3/25/20 11:53:36 AM UTC, key: 2, value: {"name": "Mountain View", "state": "CA"}
rowtime: 3/25/20 11:53:36 AM UTC, key: 3, value: {"name": "Knoxville", "state": "TN"}
rowtime: 3/25/20 11:53:36 AM UTC, key: 4, value: {"name": "Houston", "state": "TX"}
rowtime: 3/25/20 11:53:36 AM UTC, key: 5, value: {"name": "Olympia", "state": "WA"}
rowtime: 3/25/20 11:53:36 AM UTC, key: 6, value: {"name": "Bismarck", "state": "ND"}
Topic printing ceased

Notice how each Kafka message’s key has been set to the city_id.

7
Declare the topic as a ksqlDB table

Now that we have a topic with data in from the source system and the keys appropriately set, we can declare a ksqlDB table over the topic.

We only need to specify the datatype of the table’s key (CITY_ID) – the rest of the schema is picked up automagically from the Schema Registry since we’re using Avro to serialize the value part of the payload.

CREATE TABLE CITIES (CITY_ID INT PRIMARY KEY) WITH (KAFKA_TOPIC='postgres_cities', VALUE_FORMAT='AVRO');

With this table object created, we can query it or use it in queries such as joining to other objects.

SET 'auto.offset.reset' = 'earliest';

SELECT CITY_ID, NAME, STATE FROM CITIES EMIT CHANGES LIMIT 6;
+--------------------+--------------------+--------------------+
|CITY_ID             |NAME                |STATE               |
+--------------------+--------------------+--------------------+
|1                   |Raleigh             |NC                  |
|2                   |Mountain View       |CA                  |
|3                   |Knoxville           |TN                  |
|4                   |Houston             |TX                  |
|5                   |Olympia             |WA                  |
|6                   |Bismarck            |ND                  |
Limit Reached
Query terminated

8
Clean up

Exit the ksqlDB CLI by entering exit and shut down the stack by running:

docker-compose down

Take it to production

1
Write your statements to a file

Now that you have a series of statements that’s doing the right thing, the last step is to put them into a file so that they can be used outside the CLI session. Create a file at src/statements.sql with the following content:

CREATE SOURCE CONNECTOR JDBC_SOURCE_POSTGRES_01 WITH (
    'connector.class'= 'io.confluent.connect.jdbc.JdbcSourceConnector',
    'connection.url'= 'jdbc:postgresql://postgres:5432/postgres',
    'connection.user'= 'postgres',
    'connection.password'= 'postgres',
    'mode'= 'incrementing',
    'incrementing.column.name'= 'city_id',
    'topic.prefix'= 'postgres_',
    'transforms'= 'copyFieldToKey,extractKeyFromStruct,removeKeyFromValue',
    'transforms.copyFieldToKey.type'= 'org.apache.kafka.connect.transforms.ValueToKey',
    'transforms.copyFieldToKey.fields'= 'city_id',
    'transforms.extractKeyFromStruct.type'= 'org.apache.kafka.connect.transforms.ExtractField$Key',
    'transforms.extractKeyFromStruct.field'= 'city_id',
    'transforms.removeKeyFromValue.type'= 'org.apache.kafka.connect.transforms.ReplaceField$Value',
    'transforms.removeKeyFromValue.blacklist'= 'city_id',
    'key.converter' = 'org.apache.kafka.connect.converters.IntegerConverter'
);

2
Send the statements to the REST endpoint

Launch your statements into production by sending them to the REST API with the following command:

tr '\n' ' ' < src/statements.sql | \
sed 's/;/;\'$'\n''/g' | \
while read stmt; do
    echo '{"ksql":"'$stmt'", "streamsProperties": {}}' | \
        curl -s -X "POST" "http://localhost:8088/ksql" \
             -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \
             -d @- | \
        jq
done

Deploy on Confluent Cloud

1
Run your app to Confluent Cloud

Instead of running a local Kafka cluster, you may use Confluent Cloud, a fully-managed Apache Kafka service.

First, create your Kafka cluster in Confluent Cloud. Use the promo code CC100KTS to receive an additional $100 free usage (details).

Next, from the Confluent Cloud UI, click on Tools & client config to get the cluster-specific configurations, e.g. Kafka cluster bootstrap servers and credentials, Confluent Cloud Schema Registry and credentials, etc., and set the appropriate parameters in your client application.

Now you’re all set to your run application locally while your Kafka topics and stream processing is backed to your Confluent Cloud instance.