installations.txt

sudo systemctl start zookeeper.service kafka.service connect.service

Register mysql CDC connector:
curl -i -X POST -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/ \
    -d '{
      "name": "mysql-connector",
      "config": {
            "connector.class": "io.debezium.connector.mysql.MySqlConnector",
            "database.hostname": "localhost",
            "database.port": "3306",
            "database.serverTimezone": "Europe/Rome",
            "database.user": "sa18",
            "database.password": "software_architectures_18",
            "database.server.id": "42",
            "database.server.name": "sa18",
            "database.history.kafka.bootstrap.servers": "localhost:9092",
            "database.history.kafka.topic": "dbhistory.sa18" ,
            "include.schema.changes": "true",
            "max.request.size":"104857600"
       }
    }'


./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic translated_categories
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic sa18.raw_data.analytics
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic step_analytics
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic recipe_analytics
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic equip_analytics
./kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name sa18.raw_data.analytics --alter --add-config max.message.bytes=104857600


Register mongoDB sink connector:
curl -i -X POST -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/ \
    -d ' {
    "name": "mongodb-sink",
    "config": {
        "connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
        "tasks.max": "1",
        "topics": "translated_categories",
        "mongodb.connection.uri": "mongodb://localhost:27017/connect",
        "mongodb.document.id.strategy":"at.grahsl.kafka.connect.mongodb.processor.id.strategy.BsonOidStrategy",
        "mongodb.collection": "events",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": false,
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": false
    }
}'



Extra options for the mongoDB sink that could be useful:
ProvidedInValueStrategy
        "mongodb.post.processor.chain":"at.grahsl.kafka.connect.mongodb.processor.field.renaming.RenameByMapping",
        "mongodb.field.renamer.mapping": "[{\"oldName\":\"value.equipID\",\"newName\":\"_id\"}]"

mongodb.field.renamer.mapping=[{"oldName":"key.fieldA","newName":"field1"},{"oldName":"value.xyz","newName":"abc"}]