JDBC Kafka Source Connector: Stream Database Changes or ENTIRE Tables πŸš‚πŸ“‘



 Confluent has built some useful tools that rely on Kafka. One of these tools is Kafka Connect. And as I have shown you before one of the ways to use Kafka Connect as a consumer of Kafka Topics, I will show you how to use Kafka Connect as a producer to Kafka topics

Kafka Connect offers a very powerful feature which is connecting to databases using certain connectors as JDBC (Java Database Connectivity).  The Kafka Connect JDBC source connector allows you to import data from any relational database with a JDBC driver into an Apache Kafka topic. This connector supports a wide variety of databases. Data is loaded by periodically executing a SQL query and creating an output record for each row in the result set. It enables you to pull data (source) from a database into Kafka, and also push data (sink) from a Kafka topic to a database.

In this tutorial we'll be focusing on pulling data from a SQL database. I will be reusing the same setup as my previous Kafka Connect post so make sure to follow the steps in that post to prepare your application and the needed containers.

Running Kafka Connect πŸƒ

As I mentioned I will be reusing the YAML file I used to run Kafka Connect, Kafka and Zookeeper. But I will be editing the Dockerfile to add the JDBC connector along with the Elasticsearch connector although we will not be using the Elasticsearch connector, but I will just leave it there.

FROM confluentinc/cp-kafka-connect:latest

RUN confluent-hub install --no-prompt confluentinc/kafka-connect-elasticsearch:latest
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:latest

Great! Now, let's run the same docker-compose.yml file from the previous post using docker compose up.

After the containers run, we should be able be able to communicate with Kafka Connect and so let's for example get the connectors.


Perfect! Now that Kafka Connect is running and responding, let's go ahead and configure our JDBC connectors.

Configuring JDBC πŸ”§

There are several modes in JDBC connector but in this tutorial, I will be focusing on incrementing and bulk.

Incrementing means that when a certain specified column increments, a new message will be published on a topic with the same name as the table. Streaming database changes allows real-time analytics, monitoring, data warehousing, logging, ETL and more. 

Bulk mode constantly publishes all the records in the table in the form of messages. So, let's prepare the configs needed for both modes. Bulk streaming has the same benefits as incrementing streaming, but it also allows database replication.

I will be connecting to a database named kafka-jdbc which has a table named Documents. And this is the table that I will be streaming to a topic with the same name as the table but with a different prefix for each mode.

Let's start with the incrementing mode config.
{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:sqlserver://host.docker.internal;databaseName=kafka-jdbc;user=jdbc;password=123;",
    "connection.user": "jdbc",
    "connection.password": "123",
    "table.whitelist": "Documents",
    "mode": "incrementing",
    "incrementing.column.name": "Id",
    "topic.prefix": "sql-",
    "poll.interval.ms": 1000
  }
}
As you can see, I simply added a connection string, table name, a prefix to add to the name of the table when publishing to the topic, the name of the column that on its incrementing should publish a new message and finally a poll interval.

Let's go ahead and post this config to set up or incremental connector. Also, just like my previous post I will be posting this config by saving it in a file then posting this file via the Kafka Connect API.


Perfect! Now let's set up our bulk connector. I will be using almost the same configuration, but I will just increase the interval, change the prefix and change the name of the connector.
{
  "name": "jdbc-source-connector-bulk",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:sqlserver://host.docker.internal;databaseName=kafka-jdbc;user=jdbc;password=123;",
    "connection.user": "jdbc",
    "connection.password": "123",
    "table.whitelist": "Documents",
    "mode": "bulk",
    "topic.prefix": "sql-bulk-",
    "poll.interval.ms": 5000
  }
}
Note that I'm using the host.docker.internal server because my database is running on my local SQL server.

Let's also post this connector using the API.


Excellent! Now that our config files are posted successfully, let's run our application to listen to the topics and also create these topics because otherwise, Kafka Connect will not automatically create these topics.

Streaming Changes πŸ“

Let's first run the application and create the needed topics for Kafka Connect like we did in the previous post but this time we will create also a topic with the name sql-Documents so that our incrementing connector can publish messages on it. I will also run a consumer that can listen to that topic. If you don't know how to set up a consumer, follow the steps in this post but don't deserialize your message, just log it instead.

After running the application and creating the topics I will be adding a new record to my database to see if this new record has been published as a message or not.




My Documents table has already three records so when I add the fourth one, Kafka Connect should publish that change to the topic. So, I will run this command and see what happens.

Now I will take a look at my application's console to see if the consumer of the topic sql-Document has logged anything.

Perfect! My new record has been published as a message to the topic and that happened because the column (Id) that I specified in the config has been incremented, so the change was detected by the connector and published a message with the new record.

Streaming Entire Tables πŸ“¦

Now for the bulk streaming, we need to create a topic with the same name we configured in our connector which is sql-bulk-Documents and set up a consumer for this topic in our application.

It's important to note that once your topic is created, the connector will constantly push all the records in the table specified in the configuration every poll interval. And this is not a one-time operation. It will keep pushing the records until you pause the connector by putting to the API named pause.

curl -X PUT "http://your-kafka-connect-host:8083/connectors/my-connector/pause"

That being, let's go ahead and run the application which will allow the connector to stream the table from the database.


Excellent! All my table records have been streamed. But it will not stop. Every 5 seconds I will receive the records again until I pause the connector.

Seems like we were able to try two of modes that JDBC provides. But there are also other modes such as:
  • Timestamp Mode: In timestamp mode, the connector uses a timestamp column to detect only new rows. It compares the timestamp of each row with the last known timestamp value to identify changes. This mode is useful when your database table has a reliable timestamp column that reflects data modifications.

  • Timestamp+Incrementing Mode: This mode combines both a timestamp column and an incrementing column. It provides more accurate tracking of changes by considering both the timestamp and incremental values. Use this mode when you want to capture both INSERT and UPDATE events.
It's important to try out these modes and also others to see which one fits your business needs and your technical requirements.

Comments

Post a Comment

Popular posts

Google & Microsoft Token Validation Middleware in ASP.NET Applications

Image
   I have been working on a web app that allows 3rd party login. I had to validate the token that existed in the authorization header of each request hitting my backend services. The frontend team handled the 3rd party login and obtained a token from one of two providers: Google or Microsoft. Middleware The first step I decided to create middleware that intercepts all calls to my applications to validate the token before taking any further steps. And because ASP.NET Core calls middlewares in order just like a pipeline so it's very important to register this middleware before any other middleware you would like to register, so as not to do any unnecessary processing of requests which are yet to be authenticated. So basically, I created a class inside my API project under a folder named  TokenHandling. And right away always remember to register your middleware/services because sometimes you get  caught up in applying your solution that miss the step where you register ...
Read more

Publish and Consume Messages Using RabbitMQ as a Message Broker in 4 Simple Steps 🐰

Image
  SOLID principles seem to be the foundation that most engineers tend to build their solutions on. Using message brokers can help you achieve these core principles. Message brokers allow services and components to communicate directly, even if they were written in different languages or implemented on different platforms so it serves the Single-Responsibility Principle . Also, Message brokers act as intermediaries, allowing applications to communicate without direct dependencies. They provide a common abstraction for message handling which helps serve another principle which is  Dependency Inversion Principle . Of course, not to mention that most microservices, domain driven and event driven approaches are heavily dependent on message brokers. And message brokers come in all shapes and sizes and a lot of popular tools can help you set up a nice intermediatory space for your messages. Some of these tools are PubSub+ , Redis , Kafka , AWS SQS and more. Some of these are in...
Read more

Real Life Case Study: Synchronous Inter-Service Communication in Distributed Systems

Image
The Problem Inter-Service Communication  I was working on a project where I was a part of a team responsible for service B, which was one service of microservices A, B, C & D.  And one day the business required that service B should process ( using the business logic of the service ) an entity that existed in the domain of service A. Therefore, a question arose; how should we manage to do it? Just Fire and Forget, right? Now the services were set up to communicate using Apache Kafka . So, normally the initial thought was that service A should publish a message with the entity that needed processing and service B should listen to this topic, process the message and publish the processed response to another topic that service A is subscribed to. But wait, this has to be a synchronous job. Service A Can't Wait, Won't Wait Okay so it turns out that service A isn't that relaxed and isn't into the whole we'll-get-back-to-you-soon attitude and wants the response immedi...
Read more