CODEWITHSUNDEEP
postgresqlapache-kafka-connectdebeziumcdcheartbeat
javaguru
javaguru

Reputation: 23

Debezium heartbeat table not updating

There is already a question Debezium Heartbeat Action not firing but it did not resolve my issue.

Here is my source connector config for postgres. It is generating heatbeat events after every 5 seconds. I have confirmed that by checking kafka topic but the issue is that it is not updating the row in the database heartbeat table. Any suggestions?

{
    "name": "postgres-localdb-source-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max": "1",
        "database.hostname": "postgres",
        "database.port": "5432",
        "slot.name":"debezium",
        "database.user": "postgres",
        "database.password": "postgres",
        "database.dbname" : "postgres",
        "database.server.name": "dbserver2",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.dbserver2",
        "schema.include": "inventory",
        "tables.include": "customers,heartbeat",
        "publication.autocreate.mode" : "filtered",
        "max.batch.size":"20480",
        "max.queue.size":"81920",
        "poll.interval.ms":"100",
        "heartbeat.interval.ms": "5000",
        "heartbeat.action.query" :"INSERT INTO heartbeat (id, ts) VALUES (1, NOW()) ON CONFLICT(id) DO UPDATE SET ts=EXCLUDED.ts;"
    } }

Upvotes: 0

Views: 2061

Answers (1)

Yuri Niitsuma
Yuri Niitsuma

Reputation: 156

Try to share your heartbeat table with a DDL. Your heartbeat table have a primary key? Debezium only track updates and deletes if table have PK defined. Also try to share your debezium version because this propertiers change from version to version.

Try a UPDATE without WHERE to test if the problem is in your query. Check if your heartbeat in your schema public or inventory and add as a prefix in your query.

UPDATE inventory.heartbeat SET ts = NOW();

On your tables.include add for each table a prefix with schema.

"tables.include": "inventory.customers,inventory.heartbeat",

On tables.include try to change to tables.include.list. Source: https://debezium.io/documentation/reference/1.6/connectors/mysql.html#:~:text=connector%20configuration%20property.-,table.include.list,-empty%20string

"tables.include.list": "inventory.customers,inventory.heartbeat",

Upvotes: 1

Related Questions