In Part 1, we reviewed Kafka connectors focusing on AWS services integration. Among the available connectors, the suite of Apache Camel Kafka connectors and the Kinesis Kafka connector from the AWS Labs can be effective for building data ingestion pipelines on AWS. In this post, I will illustrate how to develop the Camel DynamoDB sink connector using Docker. Fake order data will be generated using the MSK Data Generator source connector, and the sink connector will be configured to consume the topic messages to ingest them into a DynamoDB table.
Part 2 Develop Camel DynamoDB Sink Connector using Docker (this post)
Part 4 Develop Kinesis Kafka Connector for OpenSearch using Docker
Part 5 Deploy Kinesis Kafka Connector for OpenSearch on MSK Connect
Kafka Cluster
We will create a Kafka cluster with 3 brokers and 1 Zookeeper node using the bitnami/kafka image. Kafka 2.8.1 is used as it is the recommended Kafka version by Amazon MSK. The following Docker Compose file is used to create the Kafka cluster, and the source can also be found in the GitHub repository of this post. The resources created by the compose file are illustrated below.
services
zookeeper
A Zookeeper node is created with minimal configuration. It allows anonymous login.
kafka-[id]
Each broker has a unique ID (KAFKA_CFG_BROKER_ID) and shares the same Zookeeper connect parameter (KAFKA_CFG_ZOOKEEPER_CONNECT). These are required to connect to the Zookeeper node.
Each has two listeners – INTERNAL and EXTERNAL. The former is accessed on port 9092, and it is used within the same Docker network. The latter is mapped from port 29092 to 29094, and it can be used to connect from outside the network.
Each can be accessed without authentication (ALLOW_PLAINTEXT_LISTENER).
The number of partitions (KAFKA_CFG_NUM_PARTITIONS) and default replica factor (KAFKA_CFG_DEFAULT_REPLICATION_FACTOR) are set to 3 respectively.
metworks
A network named kafka-network is created and used by all services. Having a custom network can be beneficial when services are launched by multiple Docker Compose files. This custom network can be referred to by services in other compose files.
volumes
Each service has its own volume that will be mapped to the container’s data folder. We can check the contents of the folder in the Docker volume path. More importantly data is preserved in the Docker volume unless it is deleted so that we don’t have to recreate data every time the Kafka cluster gets started.
Kafka Connect
We can use the same Docker image because Kafka Connect is included in the Kafka distribution. The Kafka Connect server runs as a separate docker compose service, and its key configurations are listed below.
We run it as the distributed mode, and it can be started by executing connect-distributed.sh on the Docker command.
The startup script requires the properties file (connect-distributed.properties). It includes configurations such as Kafka broker server addresses – see below for details.
The Connect server is accessible on port 8083, and we can manage connectors via a REST API as demonstrated below.
The properties file and connector sources are volume-mapped.
AWS credentials are added to environment variables as the sink connector requires permission to write data into DynamoDB.
Connect Properties File
The properties file includes configurations of the Connect server. Below shows key config values.
Bootstrap Server
I changed the Kafka bootstrap server addresses. As it shares the same Docker network, we can take the service names (e.g. kafka-0) on port 9092.
Cluster group id
In distributed mode, multiple worker processes use the same group.id, and they automatically coordinate to schedule execution of connectors and tasks across all available workers.
Converter-related properties
Converters are necessary to have a Kafka Connect deployment support a particular data format when writing to or reading from Kafka.
By default, org.apache.kafka.connect.json.JsonConverter is set for both the key and value converters and schemas are enabled for both of them.
As shown later, these properties can be overridden when creating a connector.
Topics for offsets, configs, status
Several topics are created to manage connectors by multiple worker processes.
Plugin path
Paths that contains plugins (connectors, converters, transformations) can be set to a list of filesystem paths separated by commas (,)
/opt/connectors is added and connector sources will be volume-mapped to it.
Download Connectors
The connector sources need to be downloaded into the ./connectors path so that they can be volume-mapped to the container’s plugin path (/opt/connectors). The MSK Data Generator is a single Jar file, and it can be kept as it is. On the other hand, the Camel DynamoDB sink connector is an archive file, and it should be decompressed. Note a separate zip file is made as well, and it will be used to create a custom plugin of MSK Connect in a later post. The following script downloads them into the host path.
Below shows the folder structure after the connectors are downloaded successfully.
Kafka Management App
A Kafka management app can be a good companion for development as it helps monitor and manage resources on an easy-to-use user interface. We’ll use kafka-ui in this post. It provides a docker image, and we can link one or more Kafka clusters and related resources to it. In the following compose file, we added connection details of the Kafka cluster and Kafka Connect server.
Start Services
There are 3 docker compose files for the Kafka cluster, Kafka Connect and management application. We can run the whole services by starting them in order as illustrated below.
Data Ingestion to Kafka Topic
Source Connector Creation
As mentioned earlier, Kafka Connect provides a REST API that manages connectors, and we can create a connector programmatically using it. The REST endpoint requires a JSON payload that includes connector configurations.
The connector class (connector.class) is required for any connector and I set it for the MSK Data Generator. Also, a single worker is allocated to the connector (tasks.max). As mentioned earlier, the converter-related properties are overridden. Specifically, the key converter is set to the string converter as the key of the topic is set to be primitive values (genkp). Also, schemas are not enabled for both the key and value.
Those properties in the middle are specific to the source connector. Basically it sends messages to a topic named order. The key is marked as to-replace as it will be replaced with the order_id attribute of the value – see below. The value has order_id, product_id, quantity, customer_id and customer_name attributes, and they are generated by the Java faker library.
It can be easier to manage messages if the same order ID is shared with the key and value. We can achieve it using single message transforms (SMTs). Specifically I used two transforms – ValueToKey and ExtractField to achieve it. As the name suggests, the former copies the order_id value into the key. The latter is used additionally because the key is set to have primitive string values. Finally, the last transform (Cast) is to change the quantity value into integer.
Once created successfully, we can check the connector status as shown below.
As we’ve added the connector URL, the Kafka Connect menu appears on kafka-ui. We can check the details of the connector on the app as well.
Kafka Topics
As configured, the source connector ingests messages to the order topic.
We can browse individual messages in the Messages tab of the topic.
Data Ingestion to DynamoDB
Table Creation
The destination table is named orders, and it has the primary key where order_id and ordered_at are the hash and range key respectively. It also has a global secondary index where customer_id and ordered_at constitute the primary key. Note that ordered_at is not generated by the source connector as the Java faker library doesn’t have a method to generate a current timestamp. As illustrated below it’ll be created by the sink connector using SMTs. The table can be created using the AWS CLI as shown below.
Sink Connector Creation
Similar to the source connector, we can create the sink connector using the REST API.
The connector is configured to write messages from the order topic into the DynamoDB table created earlier. It requires to specify the table name, AWS region, operation, write capacity and whether to use the default credential provider – see the documentation for details. Note that, if you don’t use the default credential provider, you have to specify the access key id and secret access key. Note further that, although the current LTS version is v3.18.2, the default credential provider option didn’t work for me, and I was recommended to use v3.20.3 instead. Finally, the camel.sink.unmarshal option is to convert data from the internal java.util.HashMap type into the required java.io.InputStream type. Without this configuration, the connector fails with org.apache.camel.NoTypeConversionAvailableException error.
Although the destination table has ordered_at as the range key, it is not created by the source connector because the Java faker library doesn’t have a method to generate a current timestamp. Therefore, it is created by the sink connector using two SMTs – InsertField and TimestampConverter. Specifically they add a timestamp value to the order_at attribute, format the value as yyyy-MM-dd HH:mm:ss:SSS, and convert its type into string.
Below shows the sink connector details on kafka-ui.
DynamoDB Destination
We can check the ingested records on the DynamoDB table items view. Below shows a list of scanned records. As expected, it has the order_id, ordered_at and other attributes.
We can also obtain an individual Json record by clicking an order_id value as shown below.
Summary
The suite of Apache Camel Kafka connectors and the Kinesis Kafka connector from the AWS Labs can be effective for building data ingestion pipelines that integrate AWS services. In this post, I illustrated how to develop the Camel DynamoDB sink connector using Docker. Fake order data was generated using the MSK Data Generator source connector, and the sink connector was configured to consume the topic messages to ingest them into a DynamoDB table.
