connector-lostevent-kafka-reset-offset.sh

connector-lostevent-kafka-reset-offset.sh

# Find the partition for the connector (server name is the key) that stores its latest offset. Let that be p
$ kafkacat -b localhost -C -t <connect.offsets.storage.topic>  -f 'Partition(%p) %k \n %s\n'

# Sample output where the connector has stored offsets till 46242 and we've missed from 42000-45000
# Partition <p>
# ["my-db-connector",{"server":"server.key"}]
# {"ts_sec":,"file":"mysql-bin.000038","pos":46242, "row":1,"server_id":121,"event":3}

# Write a poison pill message: NULL to that partition. This essentially means to clear offsets
$ echo '["my-db-connector",{"server":"server.key"}]|'  \| 
    kafkacat -P -Z -b localhost -t <connect.offsets.storage.topic> -K \| 
    -p <partition: p>