Stream of delete events from MySQL to PostgreSQL via Apache-kafka

Question

I am trying to pass events from MySQLto PostgreSQLusing Apache Kafka.Although inserts and updates work fine, I cannot figure out how to remove a record from MySQLand pass this event to PostgreSQL.

Assume the following topology:

               +-------------+
               |             |
               |    MySQL    |
               |             |
               +------+------+
                      |
                      |
                      |
      +---------------v------------------+
      |                                  |
      |           Kafka Connect          |
      |  (Debezium, JDBC connectors)     |
      |                                  |
      +---------------+------------------+
                      |
                      |
                      |
                      |
              +-------v--------+
              |                |
              |   PostgreSQL   |
              |                |
              +----------------+

I use the following docker images;

1. Apache zookeper
2. Apache Kafka
3. Debezium / JDBC Connectors

then

# Start the application
export DEBEZIUM_VERSION=0.6
docker-compose up

# Start PostgreSQL connector
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @jdbc-sink.json

# Start MySQL connector
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @source.json

Here is the contents of the MySQL database;

docker-compose -f docker-compose-jdbc.yaml exec mysql bash -c 'mysql -u $MYSQL_USER  -p$MYSQL_PASSWORD inventory -e "select * from customers"'
+------+------------+-----------+-----------------------+
| id   | first_name | last_name | email                 |
+------+------------+-----------+-----------------------+
| 1001 | Sally      | Thomas    | [e-mail] |
| 1002 | George     | Bailey    | [e-mail]    |
| 1003 | Edward     | Walker    | [e-mail]         |
| 1004 | Anne       | Kretchmar | [e-mail]    |
+------+------------+-----------+-----------------------+

And we can verify that the contents of PostgresSQL are identical;

docker-compose -f docker-compose-jdbc.yaml exec postgres bash -c 'psql -U $POSTGRES_USER $POSTGRES_DB -c "select * from customers"'
 last_name |  id  | first_name |         email         
-----------+------+------------+-----------------------
 Thomas    | 1001 | Sally      | [e-mail]
 Bailey    | 1002 | George     | [e-mail]
 Walker    | 1003 | Edward     | [e-mail]
 Kretchmar | 1004 | Anne       | [e-mail]
(4 rows)

Suppose I want to delete a record id=1004from a MySQL database;

docker-compose -f docker-compose-jdbc.yaml exec mysql bash -c 'mysql -u $MYSQL_USER  -p$MYSQL_PASSWORD inventory'
mysql> delete from customers where id = 1004;


docker-compose -f docker-compose-jdbc.yaml exec mysql bash -c 'mysql -u $MYSQL_USER  -p$MYSQL_PASSWORD inventory -e "select * from customers"'
+------+------------+-----------+-----------------------+
| id   | first_name | last_name | email                 |
+------+------------+-----------+-----------------------+
| 1001 | Sally      | Thomas    | [e-mail] |
| 1002 | George     | Bailey    | [e-mail]    |
| 1003 | Edward     | Walker    | [e-mail]         |
+------+------------+-----------+-----------------------+

Although the record is deleted from MySQL, the record still appears in PostgresSQL

docker-compose -f docker-compose-jdbc.yaml exec postgres bash -c 'psql -U $POSTGRES_USER $POSTGRES_DB -c "select * from customers"'

 last_name |  id  | first_name |         email         
-----------+------+------------+-----------------------
 Thomas    | 1001 | Sally      | [e-mail]
 Bailey    | 1002 | George     | [e-mail]
 Walker    | 1003 | Edward     | [e-mail]
 Kretchmar | 1004 | Anne       | [e-mail]
(4 rows)

, , , PostgresSQL ( del MySQL Apache-Kafka)?

:

source.json

{
    "name": "inventory-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "mysql",
        "database.port": "3306",
        "database.user": "debezium",
        "database.password": "dbz",
        "database.server.id": "184054",
        "database.server.name": "dbserver1",
        "database.whitelist": "inventory",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory",
        "transforms": "route",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
        "transforms.route.replacement": "$3"
    }
}

jdbc-sink.json

{
    "name": "jdbc-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "customers",
        "connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=postgrespw",
        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
        "auto.create": "true",
        "insert.mode": "upsert",
        "pk.fields": "id",
        "pk.mode": "record_value"
    }
}

"pk.mode": "record_key" "delete.enabled": "true" ( ), , , .

Answer 1

Connluent JDBC. ( ), .

JDBC , , , DELETE .

Answer 2

- jdbc, ?