Reset Kafka Connect Source Connector Offsets

Kafka Connect in distributed mode uses Kafka itself to persist the offsets of any source connectors. This is a great way to do things as it means that you can easily add more workers, rebuild existing ones, etc without having to worry about where the state is persisted. I personally always recommend using distributed mode, even if just for a single worker instance - it just makes things easier, and more standard. Watch my talk online here to understand more about this. If you want to reset the offset of a source connector then you can do so by very carefully modifying the data in the Kafka topic itself.

In this example I’m using the very simple FileStreamSourceConnector.

As I add data to the file, it is ingested by Kafka Connect and written to a topic.

So now let’s imagine we want to get the connector to re-consume all of the source data. One very simple option is to just rename the connector, which then means it has no existing offsets stored, and thus will start from the beginning again. This can be a bit clumsy though depending on the use case. The alternative is to modify the actual offsets themselves.

The first thing is to determine the Kafka topic being used to persist the offsets. The default is usually connect-offsets but I’ve taken to overriding this to include an underscore prefix to make it easy to spot an internal topic. Regardless; you can look at your Connect worker config, and/or check the worker log for offset.storage.topic:

Now shutdown all connect workers that are using this topic. If you don’t do this then funny things might happen, since connect only periodically flushes offsets to the topic, and doesn’t read them from it except at startup.

With the connect worker shutdown, you can now examine the topic. I’m using kafkacat here because it is very flexible in how it works with both keys and partitions which will be very important in a moment.

We can see here using the consumer mode (-C) and a key separator character of # that the key of the message is the connector’s name plus file

and the value of the message is the position offset in the file

Now use the -f option of kafkacat to display this even more clearly, along with a bunch of other important metadata including the partition:

Take note of the partition number, because we’ll need that shortly.

Having seen how Kafka Connect stores this data, we can now amend it. For this, the Kafka Connect worker really does need to be shut down. We’re going to prepare our payload which we’ll pass to kafkacat in a similar way to above. The payload is going to be the key, which remains fixed, and a value. The value can either be a given offset, or it can be NULL which denotes nothing, nada, nowt—start from scratch. Here’s sending a NULL, which is known as a tombstone message. Very important is that you specify the same target partition (using -p) for the message as the one that you saw above:

In the echo we’re specifying the key (["source-file-01",{"filename":"/data/testdata.txt"}]) followed by the key separator (#, defined in the kafakcat argument -K#), and the -Z in kafkacat tells it to send the empty value as a NULL. As stated above, don’t forget to target the correct partition.

Now when we restart the Kafka Connect worker, we can see that the file has been re-processed (note the incrementing offset value but repeating Value payloads):