email-message-scheduler Debezium connector

delayed_emails_history.json

{
  "name": "delayed-email-message",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "rootpass",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.whitelist": "emails",
    "database.history.kafka.bootstrap.servers": "kafka:9093",
    "database.history.kafka.topic": "delayed.emails.history",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false",
    "table.whitelist": "emails.delayed_messages",
    "transforms": "Reroute, filter, unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "true",
    "transforms.Reroute.type": "io.debezium.transforms.ByLogicalTableRouter",
    "transforms.Reroute.topic.regex": "(.*)(delayed_messages)$",
    "transforms.Reroute.topic.replacement": "email.execution",
    "transforms.filter.type": "io.debezium.transforms.Filter",
    "transforms.filter.language": "jsr223.groovy",
    "transforms.filter.topic.regex": "email.execution",
    "transforms.filter.condition": "value.op == \"u\" && value.before.is_ready == false && value.after.is_ready == true"
  }
}