When you’re ready to deploy your documentation website, say in Docker with Nginx the following Dockerfile and Nginx default.conf should do.
Dockerfile
FROM python:3.9 as builder
WORKDIR /app
COPY . .
RUN pip install mkdocs mkdocs-material && mkdocs build
FROM nginx as deploy
# Copy the build to the nginx directory.
COPY --from=builder /app/site/ /usr/share/nginx/html/
# Copy the nginx configuration to the nginx config directory.
COPY default.conf /etc/nginx/conf.d/
EXPOSE 8080:8080/tcp
default.conf
server {
listen 8080;
root /usr/share/nginx/html/;
index index.html;
}
I thought that making videos will be easier that typing blog posts but to my surprise the difficulty is a bit higher. Fixing mistakes takes more time with videos and since I’m not that great of a presenter I struggle with presenting the content. Hopefully I will improve my skills with time and practice.
Click on mongodb-kafka-connect-mongodb-1.6.0.zip then unzip it and copy the directory into the plugin path /usr/share/java as defined in the CONNECT_PLUGIN_PATH: “/usr/share/java,/usr/share/confluent-hub-components” environment variable.
Connect needs to be restarted to pick-up the newly installed plugin. Verify that the connector plugin has been successfully installed:
➜ bin curl -s -X GET http://localhost:8083/connector-plugins | jq | head -n 20
[
{
"class": "com.mongodb.kafka.connect.MongoSinkConnector",
"type": "sink",
"version": "1.6.0"
},
{
"class": "com.mongodb.kafka.connect.MongoSourceConnector",
"type": "source",
"version": "1.6.0"
},
Note: If you don’t have jq installed you can omit it.
Creating the topics
Before starting the connector, let’s create the Kafka Topics events and events.deadletter, they will be used them in the connector.
To create the topics, we will need to download Confluent tools and run kafka-topics.
curl -s -O http://packages.confluent.io/archive/6.2/confluent-community-6.2.0.tar.gz
tar -xzf .\confluent-community-6.2.0.tar.gz
cd .\confluent-6.2.0\bin\
./kafka-topics --bootstrap-server localhost:9092 --list
__consumer_offsets
__transaction_state
_confluent-ksql-default__command_topic
_schemas
default_ksql_processing_log
docker-connect-configs
docker-connect-offsets
docker-connect-status
./kafka-topics --bootstrap-server localhost:9092 --create --topic events --partitions 3
Created topic events.
./kafka-topics --bootstrap-server localhost:9092 --create --topic events.deadletter --partitions 3
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues, it is best to use either, but not both.
Created topic events.deadletter.
Note: You will need Java to run the Confluent tools if you’re on Ubuntu you can type sudo apt install openjdk-8-jdk.
Starting the connector 🚙
To start the connector, it is enough to do a single post request to the connector’s API with the connector’s configuration.
The configuration that we will use is going to be:
In short, this POST will create a new connector named mongo-sink-connector using the com.mongodb.kafka.connect.MongoSinkConnector java class, run a single connector task that will get all the messages from the events topic and put them into the Mongo found at mongodb://mongodb:27017/my_events, database named my_events and collection named kafka_events. The records which will fail to be written into the database will be placed on a dead letter topic named events.deadletter, in my opinion this is better than discarding them, since we can inspect the topic to see what went wrong.
To verify that the connector is running, you can retrieve its first tasks status with:
➜ bin curl -s -X GET http://localhost:8083/connectors/mongo-sink-connector/tasks/0/status | jq
{
"id": 0,
"state": "RUNNING",
"worker_id": "connect:8083"
}
Querying the Database 🗃
Now that our Kafka Connect cluster is running and is configured, all that’s left to do is POST some dummy data into Kafka and check for it in the database.
That’s all! 🎉If we now connect to the database using mongosh or any other client, we can query the data.
mongosh
> use my_events
switched to db my_events
> db.kafka_events.findOne()
{
_id: ObjectId("6147242856623b0098fc756d"),
glossary: {
title: 'example glossary',
GlossDiv: {
title: 'S',
GlossList: {
GlossEntry: {
ID: 'SGML',
SortAs: 'SGML',
GlossTerm: 'Standard Generalized Markup Language',
Acronym: 'SGML',
Abbrev: 'ISO 8879:1986',
GlossDef: {
para: 'A meta-markup language, used to create markup languages such as DocBook.',
GlossSeeAlso: [ 'GML', 'XML' ]
},
GlossSee: 'markup'
}
}
}
}
}
Viewing Kafka Connect JMX Metrics
JConsole is a tool that can be used to view JMX metrics exposed by Kafka Connect, if you installed openjdk-8 it should come with it
Start JConsole and connect to localhost:9102. If you get a warning about an insecure connection, accept the connection, and ignore it.
After you’re connected click the MBeans tab and explore 🦹♀️
Summary
Getting into Kafka and Kafka Connect can be a bit overwhelming at first. I hope that this tutorial has provided you with the necessary basics so you can continue to play and explore on your own.
Spinning up a playground for Kafka and Connect using docker-compose isn’t that complicated, you can start from the confluent-cp-community repo, it will give you everything you need to get started. With some little modifications to the docker-compose file, we’ve spawned a MongoDB instance and exposed the JMX metrics in Kafka Connect.
Next, we’ve installed and configured the MongoDB connector and confirmed that it works as expected.
If you have any questions let me know in the comments.
In this article I will give you some tips on how to improve the throughput of a message producer.
I had to write a Golang based application which would consume messages from Apache Kafka and send them into a sink using HTTP JSON / HTTP Protocol Buffers.
To see if my idea works, I started using a naïve approach in which I polled Kafka for messages and then send each message into the sink, one at a time. This worked, but it was slow.
To better understand the system, a colleague has setup Grafana and a dashboard for monitoring, using Prometheus metrics provided by the Sink. This allowed us to test various versions of the producer and observe it’s behavior.
Let’s explore what we can do to improve the throughput.
Request batching 📪
A very important improvement is request batching.
Instead of sending one message at a time in a single HTTP request, try to send more, if the sink allows it.
As you can see in the image, this simple idea improved the throughput from 100msg/sec to ~4000msg/sec.
Batching is tricky, if your batches are large the receiver might be overwhelmed, or the producer might have a tough time building them. If your batches contain a few items you might not see an improvement. Try to choose a batch number which isn’t too high and not to low either.
Fast JSON libraries ⏩
If you’re using HTTP and JSON then it’s a good idea to replace the standard JSON library.
There are lots of open-source JSON libraries that provide much higher performance compared to standard JSON libraries that are built in the language.
There are several partitioning strategies that you can implement. It depends on your tech stack.
Kafka allows you to assign one consumer to one partition, if you have 3 partitions in a topic then you can run 3 consumer instances in parallel from that topic, in the same consumer group, this is called replication, I did not use this as the Sink does not allow it, only one instance of the Producer is running at a time.
If you have multiple topics that you want to consume from, you can partition on the topic name or topic name pattern by subscribing to multiple topics at once using regex. You can have 3 consumers consuming from highspeed.* and 3 consumer consuming from other.*. If each topic has 3 partitions.
Note: The standard build of librdkafka doesn’t support negative lookahead regex expressions, if that’s what you need you will need to build the library from source. See issues/2769. It’s easy to do and the confluent-kafka-go client supports custom builds of librdkafka.
Negative lookahead expressions allow you to ignore some patterns, see this example for a better understanding: regex101.com/r/jZ9AEz/1
Source Parallelization 🕊
The confluent-kafka-go client allows you to poll Kafka for messages. Since polling is thread safe, it can be done in multiple goroutines or threads for a performance improvement.
import (
"fmt"
"github.com/confluentinc/confluent-kafka-go/kafka"
)
func main() {
var wg sync.WaitGroup
c, err := kafka.NewConsumer(&kafka.ConfigMap{
"bootstrap.servers": "localhost",
"group.id": "myGroup",
"auto.offset.reset": "earliest",
})
if err != nil {
panic(err)
}
c.SubscribeTopics([]string{"myTopic", "^aRegex.*[Tt]opic"}, nil)
for i := 0; i < 5; i++ {
go func() {
wg.Add(1)
defer wg.Done()
for {
msg, err := c.ReadMessage(-1)
if err == nil {
fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
// TODO: Send data through a channel to be processed by another goroutine.
} else {
// The client will automatically try to recover from all errors.
fmt.Printf("Consumer error: %v (%v)\n", err, msg)
}
}
}()
}
wg.Wait()
c.Close()
}
Buffered Golang channels can also be used in this scenario in order to improve the performance.
Protocol Buffers 🔷
Finally, I saw a huge performance improvement when replacing the JSON body of the request with Protocol Buffers encoded and snappy compressed data.
If your Sink supports receiving protocol buffers, then it is a good idea to try sending it instead of JSON.
Honorable Mention: GZIP Compressed JSON 📚
The Sink supported receiving GZIP compressed JSON, but in my case I didn’t see any notable performance improvements.
I’ve compared the RAM and CPU usage of the Producer, the number of bytes sent over the network and the message throughput. While there were some improvements in some areas, I decided not to implement GZIP compression.
It’s all about trade-offs and needs.
Conclusion
As you could see, there are several things you can do to your producers in order to make them more efficient.
Request Batching
Fast JSON Libraries
Partitioning
Source Parallelization & Buffering
Protocol Buffers
Compression
I hope you’ve enjoyed this article and learned something! If you have some ideas, please let me know in the comments.
Nucu Labs 