How to do an initial snapshot load with debezium mysql connector kafka?

user_7601181 0
Viewed 2385

I'm using Kafka Cluster and Debezium MySql connector to get messages from database updates to an Elasticsearch. For a while I was doing some test and I ended up with a mixed ES so I wanted to do a fresh full load to Elasticsearch.

So I wanted to stop the debezium connector to stop feeding ES and remove all the indexes so when I start again the connector it could do a full load. As far as I know, the connector only works with pause/resume actions and those does not do an initial load.

At this moment I will remove manually the connector and the topics created and re-create them so it can load as it was initial but any ideas of how to perform this clean up process in the right steps?

debezium apache-kafka-connect apache-kafka mysql
1 Answers

I don't know if there is a better way but this solution worked for me:

  1. Stopped the services for the /etc/kafka/connect-distributed.properties daemon.
sudo systemctl stop confluent-connect-distributed.service
  1. Delete the connector of Debezium 
curl -X DELETE http://localhost:8083/connectors/<connector name>
  1. delete all the topics related to my work, in this case this kafka cluster was a dev one so I deleted all the topics, even the defaults ones.
__consumer_offsets
_confluent-command
_schemas
connect-configs
connect-offsets
connect-status

kafka-topics --bootstrap-server <kafka bootstrap> --delete --topic <topic name>
  1. Create again the default topics
kafka-topics --create --bootstrap-server <boostrap kafka> --topic connect-configs --replication-factor 3 --partitions 1 --config cleanup.policy=compact
kafka-topics --create --bootstrap-server <boostrap kafka> --topic connect-offsets --replication-factor 3 --partitions 50 --config cleanup.policy=compact
kafka-topics --create --bootstrap-server <boostrap kafka> --topic connect-status --replication-factor 3 --partitions 10 --config cleanup.policy=compact
  1. Start service again
sudo systemctl start confluent-connect-distributed.service
  1. Create connector again
curl -s -i -X PUT -H  "Content-Type:application/json" \
    http://localhost:8083/connectors/<connector name>/config \
    -d '<json>'

This way you will end up with a initial snapshot in elasticsearch.

user_76011810
Related