Reset the JDBC Kafka Connector to start pulling rows from the beginning of time?

Question

The Kafka Connector can make use of a primary key and a timestamp to determine which rows need to be processed.

I'm looking for a way to reset the Connector so that it will process from the beginning of time.

Answer 1

Because the requirement is to run in distributed mode, the easiest thing to do is to update the connector name to a new value. This will prompt a new entry to be made into the connect-offsets topic as it looks like a totally new connector. Then the connector should start reading again as if nothing has been written to Kafka yet. You could also manually send a tombstone message to the key in the connect-offsets topic associated with that particular connector, but renaming is much easier than dealing with that. This method applies to all source connectors, not only the JDBC one described here.

Answer 2

I got a bit tired of renaming the connector every time during development so started using the tombstone method. This methods can be used for any source connector.

First check the format of the key/value of the connector:

kafka-console-consumer --bootstrap-server localhost:9092 --topic kafka-connect-offsets --from-beginning --property print.key=true
["demo",{"query":"query"}] {"timestamp_nanos":542000000,"timestamp":1535768081542}
["demo",{"query":"query"}] {"timestamp_nanos":171831000,"timestamp":1540435281171}
["demo",{"query":"query"}] {"timestamp_nanos":267775000,"timestamp":1579522539267}

Create the tombstone message by sending the key without any value:

echo '["demo",{"query":"query"}]#' | kafka-console-producer --bootstrap-server localhost:9092 --topic kafka-connect-offsets --property "parse.key=true" --property "key.separator=#"

Now restart or recreate the connector and it will start producing messages again.

Be very careful with this in production unless you really know what you're doing. There's some more information here: https://rmoff.net/2019/08/15/reset-kafka-connect-source-connector-offsets/