How to use a unique sorting key for Debezium with GCP Pub/Sub

65 Views Asked by At

I am using Debezium to capture changes for all tables in my database. I am rerouting every event into a single GCP Pub/Sub topic so that my worker can consume those messages in the same order they were created. The worker is subscribing to new messages on a Pub/Sub topic using @google-cloud/pubsub .on method.

My problem is that the ordering is not working. When I pull and log a single message once without acknowledging it, stop my worker, wait 10 seconds (ackDeadline) and start my worker again I have different messages.

I tried logging the ordering key of these messages and the JSON object is different across messages.

Message 1 sorting key:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "default": 0,
        "field": "dl_seq"
      },
      {
        "type": "string",
        "optional": false,
        "field": "__dbz__physicalTableIdentifier"
      }
    ],
    "optional": false,
    "name": "prod.x.Key"
  },
  "payload": {
    "dl_seq": 20853238,
    "__dbz__physicalTableIdentifier": "prod.public.table1"
  }
}

Message 2 sorting key:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "default": 0,
        "field": "dosliv_seq"
      },
      {
        "type": "string",
        "optional": false,
        "field": "__dbz__physicalTableIdentifier"
      }
    ],
    "optional": false,
    "name": "prod.x.Key"
  },
  "payload": {
    "dosliv_seq": 6925801,
    "__dbz__physicalTableIdentifier": "prod.public.table2"
  }
}

Has anyone had to use Debezium to apply every changes to another database using GCP Pub/Sub and had the same problem ?

0

There are 0 best solutions below