Using Kafka Connect to connect to enterprise MQ systems

One of the big decisions that led to the Apache Kafka that we know today was to build the Kafka Connect framework for connecting to other systems right into the open-source Apache Kafka project. Kafka Connect gives a standard way of building connectors: source connectors that transfer data into Kafka and sink connectors that transfer data out of Kafka. The connectors themselves are not part of the project, but the framework and runtime components are. As a result, many people have built connectors to all kinds of other systems. If you’re introducing Kafka into an organization, you’re going to want to connect it to other stuff, so making this as easy as possible makes a lot of sense.

In IBM, we have some history connecting to Kafka before Kafka Connect had turned into a mature technology. For example, when Kafka Connect was first introduced, there was no real way to pause a connector and have it resume again. Our cloud Kafka service, IBM Event Streams for IBM Cloud, offers “bridges” from IBM MQ to Kafka, and from Kafka to a choice of cloud-based object storage services, but none of these use Kafka Connect. The design of the bridges puts the non-Kafka system in charge, and because those systems differ in behaviour, the Kafka features they use is also different. In contrast, Kafka Connect has an understandably Kafka-centric view and it dictates the way that the connector gets data into and out of Kafka.

A Kafka Connect connector is basically a pump with Kafka at one end and something else at the other. Because Kafka Connect is part of Kafka itself, the Apache Kafka project is responsible for writing the Kafka end of these connectors following best practice. All you need to do to make a new connector is to implement a standard interface and write the code that connects to the other system. It’s very easy to get data flowing, but it’s actually surprisingly hard to make a reliable, fast connector. How hard depends on how well the other system’s API matches the Kafka Connect way of doing things.

Let’s take the example of a Kafka Connect source connector. This takes data from another system and writes it onto a Kafka topic. Once you start delving into the details, it becomes clear that the job can be quite tricky if the API of the other system isn’t quite the same shape as the Kafka Connect source connector interface. The Javadoc is pretty good but there’s actually some subtlety in terms of threading, error recovery and the overall philosophy that takes a bit of getting used to.

I’m the author of IBM’s source and sink connectors for IBM MQ. They’re fully supported for use by customers of IBM Event Streams. They’ve been through several iterations and a lot of independent testing, but there’s always room for improvement. So, I recently dug deeply into how the source connector framework actually works in detail and it turned out to be a bit different than I’d expected from reading the documentation.

Observations from writing the MQ source connector

In summary, here are the important points I learnt from writing the source connector:

• A source connector basically works in batches and the idea of batches is quite fundamental to the way the connector works. Each call to the connector’s poll() method returns a batch of records and Kafka Connect treats them as a batch for delivery into Kafka. So, it looks like a good idea to figure out an appropriate batch size for your source system and return that amount of data on each call to poll(), without blocking for an inconveniently long time since that can introduce a lot of delay into your connector.
• If you want to observe messages successfully arriving in Kafka, implement the commitRecord() method. It may well be that this level of control isn’t much use to you, but if you want to make sure to minimise duplication as far as possible while ensuring no data is lost, this can be a very helpful tool.
• The connector uses the JMS API to talk to MQ. It turns out that there are a couple of characteristics of the JMS API that make it a bit tricky. It’s best to use a transacted session to confirm receipt of messages from MQ you’re sure Kafka has acknowledged them, but it takes some care to make sure that you commit at the right times so nothing is lost and duplication is kept to an absolute minimum.
• The opportunity for reporting errors in the connector is very limited. You end up having to do all of the connection management and exception handling in your connector code, while trying to fit in with the Kafka Connect way of working. This actually included implementing an exponential back-off loop for reconnection attempts in my MQ connector.
• Testing a connector properly is really hard. You can treat it as a black box and try to cause lots of interesting behaviour from outside, but there are error paths in Kafka Connect itself that are extremely difficult to exercise reliably. Because each task has multiple threads, this also makes testing difficult to be certain that you’ve covered all bases.
• Be careful to understand the commit() method. You really need a very clear idea of how it works to make proper use of it. I ended up just using it as a kind of “pulse” to check that progress continues to be made. Even though it looks like each task is single-threaded, it’s not and commit() is called on a different thread than poll() and is not synchronized with it. To understand exactly what a call to commit() was telling me, I had to do the following…
• Read the Kafka Connect source code in detail. Yes, really. If you want to understand properly how it works, it’s the only way. That’s the beauty of open source. For a source connector, pore over WorkerSourceTask.java, and for a sink connector, look at WorkerSinkTask.java. They’re not all that complicated, and I learnt so much about how it all hangs together that I wish I’d started here.

As my understanding of the inner workings developed, I changed the design of the connector to make its use of MQ better match the design of Kafka Connect. This Having done all of this, what difference did it make in terms of how the MQ source connector works? Well, the connector still uses the JMS interface to talk to MQ and it still uses transacted sessions, but the transactions are now smaller and the connector tracks the successful arrival of each message in Kafka. Because the transactions are smaller, the chances of message duplication are reduced. It’s a bit more complicated but I think it’s more faithful to the way that Kafka Connect is intended to be used.

Exactly-once delivery

It’s essentially impossible using Kafka Connect to ensure exactly-once message transfer between MQ and Kafka. You can get close, but you cannot readily overcome duplication of your data in some situations. You can’t quite tie the removal of messages from MQ to their delivery into Kafka with 100% accuracy.

Kafka Connect is set up to retry delivery in order to make absolutely sure that nothing is lost. That’s fine, but if you think you want exactly-once delivery, I suggest you first try and talk yourself out of it, because it’s common with Kafka to tolerate occasional message duplication caused by retries. If you’re still sure you absolutely need exactly-once delivery, I suggest writing your own connector from scratch using the Kafka idempotent producer API. If I wanted to build an exactly-once MQ to Kafka connector, that’s what I would do. I wouldn’t start with Kafka Connect. By making the job easier, it takes away some control and that control is important for exactly-once delivery.

Wrapping up

Kafka Connect is a very valuable part of the Apache Kafka project. The way that it’s an intrinsic part of the Kafka has led to a wide variety of connectors that you can just pick up and use, most of them open-source. Writing connectors yourself is definitely possible, but really take the time to understand the way that Kafka Connect works to evaluate how well the API for your chosen system matches Kafka Connect.