Why ClickHouseSinkConnector with Debezium PostgresConnector replicates zero values or nulls

[rsurgiewicz]

Hi Guys,
I spent a lot of time trying to set up Kafka-connect with debezium connectors (In) to ClickhouseSinkConnector (Sink).

My reference docs were mainly:
https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-1
https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2
https://github.com/ClickHouse/examples/blob/main/cdc/postgresql/connectors/debezium_configuration.json
https://clickhouse.com/docs/en/integrations/kafka#clickhouse-kafka-connect-sink

Despite trying many different configs I got to the point where any operation on source Postgres Table produces a row in Clickhouse with 0 or empty values.

ClickhouseSinkConnector version: 1.0.0| 1.0.1 tested both
DebeziumPostgres connectors version: 2.3 | 2.4 tested both

I'm also not sure (docs are actually inconsistent) whether to use transforms.flatten or not.
Tried both with/without it and there's no change.

My configs (I'm using docker containers with no ssl, dev clickhouse is not exposed at all currently)

Postgres connector:

config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector", 
    "plugin.name": "pgoutput",
    "database.hostname": "xxxx", 
    "database.server.name": "postgres", 
    "database.port": "xxx", 
    "database.user": "xxxx",
    "database.password": "xxxx",
    "database.dbname" : "xxxx", 
    "table.include.list": "public.testradiomeas",
    "topic.prefix": "testprefix",
    "publication.name": "myint_logical_replication",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",  
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "snapshot.mode": "never",
    "tombstones.on.delete": "false",
    "decimal.handling.mode": "double",
    "transforms": "set_topic,flatten",
    "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
    "transforms.flatten.delimiter": ".",
    "transforms.set_topic.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.set_topic.regex": ".*",
    "transforms.set_topic.replacement": "testradiomeas",
    "value.converter.schemas.enable": "false",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter"
}

ClickhouseSinkConnector:

{
   "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "database": "default",
    "errors.retry.timeout": "60",
    "exactlyOnce": "false",
    "hostname": "clickhouse",
    "port": "8123",
    "ssl": "false",
    "username": "default",
    "password": "xxxx",
    "topics": "testradiomeas",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": ""
}

Sample message from PostgresTopic (without flattening)

{
    "before": {
        "measDate": 3,
        "measLongitude": 4.0,
        "measLatitude": 4.0,
        "measNetNoCoverage": 0,
        "measNetType": "4G"
    },
    "after": {
        "measDate": 3,
        "measLongitude": 4.0,
        "measLatitude": 4.0,
        "measNetNoCoverage": 1,
        "measNetType": "4G"
    },
    "source": {
        "version": "2.4.0.Beta2",
        "connector": "postgresql",
        "name": "xxxx",
        "ts_ms": 1694612998884,
        "snapshot": "false",
        "db": "xxxx",
        "sequence": "[\"58089216\",\"58089528\"]",
        "schema": "public",
        "table": "testradiomeas",
        "txId": 728,
        "lsn": 58089528,
        "xmin": null
    },
    "op": "u",
    "ts_ms": 1694612999037,
    "transaction": null
}

Sample message from PostgresTopic (with flattening)

{
    "before.measDate": 3,
    "before.measLongitude": 4.0,
    "before.measLatitude": 4.0,
    "before.measNetNoCoverage": 1,
    "before.measNetType": "4G",
    "after.measDate": 3,
    "after.measLongitude": 4.0,
    "after.measLatitude": 4.0,
    "after.measNetNoCoverage": 0,
    "after.measNetType": "4G",
    "source.version": "2.4.0.Beta2",
    "source.connector": "postgresql",
    "source.name": "gibbonapi",
    "source.ts_ms": 1694614700127,
    "source.snapshot": "false",
    "source.db": "xxxx",
    "source.sequence": "[\"58090632\",\"58090920\"]",
    "source.schema": "public",
    "source.table": "testradiomeas",
    "source.txId": 729,
    "source.lsn": 58090920,
    "source.xmin": null,
    "op": "u",
    "ts_ms": 1694614700664,
    "transaction.id": null,
    "transaction.total_order": null,
    "transaction.data_collection_order": null
}

What could be the reason behind is not replicating correctly
Both source and target table structure is the same.

Thanks!

[Paultagoras]

Hmmm, could you attach the "SHOW CREATE TABLE" so I can see the table structure?

[rsurgiewicz]

Sure

CREATE TABLE default.testradiomeas
(
    `measDate` Int32,
    `masLongitude` Float64,
    `measLatitude` Float64,
    `measNetNoCoverage` Int32,
    `measNetType` String
)
ENGINE = ReplacingMergeTree
ORDER BY (measDate, masLongitude, measLatitude)
SETTINGS index_granularity = 8192

[Paultagoras]

Ah, I wonder then - could you try adding a column for "before.measLongitude" for instance?

[rsurgiewicz]

I see your idea, but seems not to work anyways:

create table default.testradiomeas
(
    measDate          Int32,
    masLongitude      Float64,
    measLatitude      Float64,
    measNetNoCoverage Int32,
    measNetType       String,
    "before.measLongitude" Float64
)
    engine = ReplacingMergeTree ORDER BY (measDate,masLongitude,measLatitude)
        SETTINGS index_granularity = 8192;

[Paultagoras]

Hmm, interesting - let me set up something with a similar data structure to see if I can get more detail out of it.

[rsurgiewicz]

Just for the curiosity, shall "flatten" transformation be used as described here:
https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2
and https://github.com/ClickHouse/examples/blob/main/cdc/postgresql/connectors/debezium_configuration.json

or not?

[Paultagoras]

How you're flattening looks right - I think I might know what's going on though. Could you try this for the database?

CREATE TABLE default.testradiomeas
(
    `before.measDate` Int32,
    `before.measLongitude` Float64,
    `before.measLatitude` Float64,
    `before.measNetNoCoverage` Int32,
    `before.measNetType` String
)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}')
ORDER BY `before.measDate`

[rsurgiewicz]

Table creation works fine but it's still populated with zeros/empty values

+---------------+--------------------+-------------------+------------------------+------------------+
|before.measDate|before.measLongitude|before.measLatitude|before.measNetNoCoverage|before.measNetType|
+---------------+--------------------+-------------------+------------------------+------------------+
|0              |0                   |0                  |0                       |                  |
|0              |0                   |0                  |0                       |                  |
+---------------+--------------------+-------------------+------------------------+------------------+

[Paultagoras]

Hmm, could I see another sample of the data in the topic? Say the last... 5 messages?

[rsurgiewicz]

Sure! BTW. Kudos for trying to solve the case :-)

last message (now I tried with schema to check if it helps anyhow)
this is INSERT operation

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int64",
        "optional": true,
        "field": "before.measDate"
      },
      {
        "type": "double",
        "optional": true,
        "field": "before.measLongitude"
      },
      {
        "type": "double",
        "optional": true,
        "field": "before.measLatitude"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "before.measNetNoCoverage"
      },
      {
        "type": "string",
        "optional": true,
        "field": "before.measNetType"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "after.measDate"
      },
      {
        "type": "double",
        "optional": true,
        "field": "after.measLongitude"
      },
      {
        "type": "double",
        "optional": true,
        "field": "after.measLatitude"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "after.measNetNoCoverage"
      },
      {
        "type": "string",
        "optional": true,
        "field": "after.measNetType"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.version"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.connector"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.name"
      },
      {
        "type": "int64",
        "optional": false,
        "field": "source.ts_ms"
      },
      {
        "type": "string",
        "optional": true,
        "name": "io.debezium.data.Enum",
        "version": 1,
        "parameters": {
          "allowed": "true,last,false,incremental"
        },
        "default": "false",
        "field": "source.snapshot"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.db"
      },
      {
        "type": "string",
        "optional": true,
        "field": "source.sequence"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.schema"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.table"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "source.txId"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "source.lsn"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "source.xmin"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "string",
        "optional": true,
        "field": "transaction.id"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "transaction.total_order"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "transaction.data_collection_order"
      }
    ],
    "optional": false,
    "name": "noteldb.public.testradiomeas.Envelope",
    "version": 1
  },
  "payload": {
    "before.measDate": null,
    "before.measLongitude": null,
    "before.measLatitude": null,
    "before.measNetNoCoverage": null,
    "before.measNetType": null,
    "after.measDate": 4,
    "after.measLongitude": 5.0,
    "after.measLatitude": 6.0,
    "after.measNetNoCoverage": 7,
    "after.measNetType": "2G",
    "source.version": "2.2.1.Final",
    "source.connector": "postgresql",
    "source.name": "xxx",
    "source.ts_ms": 1694812175838,
    "source.snapshot": "false",
    "source.db": "xxx",
    "source.sequence": "[\"58144376\",\"58144432\"]",
    "source.schema": "public",
    "source.table": "testradiomeas",
    "source.txId": 757,
    "source.lsn": 58144432,
    "source.xmin": null,
    "op": "c",
    "ts_ms": 1694812176570,
    "transaction.id": null,
    "transaction.total_order": null,
    "transaction.data_collection_order": null
  }
}

this is update:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int64",
        "optional": true,
        "field": "before.measDate"
      },
      {
        "type": "double",
        "optional": true,
        "field": "before.measLongitude"
      },
      {
        "type": "double",
        "optional": true,
        "field": "before.measLatitude"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "before.measNetNoCoverage"
      },
      {
        "type": "string",
        "optional": true,
        "field": "before.measNetType"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "after.measDate"
      },
      {
        "type": "double",
        "optional": true,
        "field": "after.measLongitude"
      },
      {
        "type": "double",
        "optional": true,
        "field": "after.measLatitude"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "after.measNetNoCoverage"
      },
      {
        "type": "string",
        "optional": true,
        "field": "after.measNetType"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.version"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.connector"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.name"
      },
      {
        "type": "int64",
        "optional": false,
        "field": "source.ts_ms"
      },
      {
        "type": "string",
        "optional": true,
        "name": "io.debezium.data.Enum",
        "version": 1,
        "parameters": {
          "allowed": "true,last,false,incremental"
        },
        "default": "false",
        "field": "source.snapshot"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.db"
      },
      {
        "type": "string",
        "optional": true,
        "field": "source.sequence"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.schema"
      },
      {
        "type": "string",
        "optional": false,
        "field": "source.table"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "source.txId"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "source.lsn"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "source.xmin"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "string",
        "optional": true,
        "field": "transaction.id"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "transaction.total_order"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "transaction.data_collection_order"
      }
    ],
    "optional": false,
    "name": "noteldb.public.testradiomeas.Envelope",
    "version": 1
  },
  "payload": {
    "before.measDate": 3,
    "before.measLongitude": 4.0,
    "before.measLatitude": 2.0,
    "before.measNetNoCoverage": 0,
    "before.measNetType": "4G",
    "after.measDate": 3,
    "after.measLongitude": 4.0,
    "after.measLatitude": 2.0,
    "after.measNetNoCoverage": 0,
    "after.measNetType": "5G",
    "source.version": "2.2.1.Final",
    "source.connector": "postgresql",
    "source.name": "xxxx",
    "source.ts_ms": 1694812097921,
    "source.snapshot": "false",
    "source.db": "xxxx",
    "source.sequence": "[\"58144128\",\"58144184\"]",
    "source.schema": "public",
    "source.table": "testradiomeas",
    "source.txId": 756,
    "source.lsn": 58144184,
    "source.xmin": null,
    "op": "u",
    "ts_ms": 1694812098061,
    "transaction.id": null,
    "transaction.total_order": null,
    "transaction.data_collection_order": null
  }
}

[Paultagoras]

Hmm, what's interesting is that the first one has null values for before while the second actually fills them in.

"payload": {
    "before.measDate": 3,
    "before.measLongitude": 4.0,
    "before.measLatitude": 2.0,
    "before.measNetNoCoverage": 0,
    "before.measNetType": "4G",
    "after.measDate": 3,
    "after.measLongitude": 4.0,
    "after.measLatitude": 2.0,
    "after.measNetNoCoverage": 0,
    "after.measNetType": "5G",
    "source.version": "2.2.1.Final",
    "source.connector": "postgresql",
    "source.name": "xxxx",
    "source.ts_ms": 1694812097921,
    "source.snapshot": "false",
    "source.db": "xxxx",
    "source.sequence": "[\"58144128\",\"58144184\"]",
    "source.schema": "public",
    "source.table": "testradiomeas",
    "source.txId": 756,
    "source.lsn": 58144184,
    "source.xmin": null,
    "op": "u",
    "ts_ms": 1694812098061,
    "transaction.id": null,
    "transaction.total_order": null,
    "transaction.data_collection_order": null
  }

That looks like it should be fine (and when I tested it, it inserted no problem) - could you poll a few messages from the kafka topic, just to see what messages are coming in?

[rsurgiewicz]

Hmm, what's interesting is that the first one has null values for before while the second actually fills them in.
The first one is INSERT of a new row, therefore there are nulls for "before" stage.

I've setup a isolated local environment so the only messages in this kafka topic are the one that are generated by my manual changes to postgres table and then processed by debezium plugin.

[Paultagoras]

Just to recap - the flatten mentioned converts the data to the above format. The columns need to reflect this (as in the sample CREATE statement earlier) or else it just uses "default" values.

[rsurgiewicz]

How can I debug the connector plugin ?
I'm just guessing that the problem is either on:

• digesting incoming message --> it's not parsed correctly
• pushing data to CH --> it's parsed but problem is in commiting to CH table

[Paultagoras]

So the easiest way on the clickhouse-connector side is to enable debugging on the kafka connect instance - CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR,com.clickhouse=DEBUG

Another option is to using https://docs.confluent.io/platform/current/tools/kafkacat-usage.html to view the messages in the topic (separate from the connector) - that would help make sure the data is valid in the queue

[rsurgiewicz]

Ok there's some small breakthrough:

After disabling schema it works with

CREATE TABLE default.testradiomeas
(
    `before.measDate` Int32,
    `before.measLongitude` Float64,
    `before.measLatitude` Float64,
    `before.measNetNoCoverage` Int32,
    `before.measNetType` String
)
ENGINE = MergeTree
ORDER BY `before.measDate`;

or similar with 'after'

So it looks like that sink connector does not interpret after.xxx and before.xxx correctly.

Example here
https://github.com/ClickHouse/examples/blob/main/cdc/postgresql/connectors/debezium_configuration.json
and also article: https://clickhouse.com/blog/clickhouse-postgresql-change-data-capture-cdc-part-2

clearly confirms that "before" "after" fields feature are meant to be used by forcing "after.state.only": "false",

[elmaghrabii]

Hi,

i have the same issue , please can you help me ?

[elmaghrabii]

on postgres
mag=# insert into mag_tables values (666,'amr',1989);
INSERT 0 1

====================

on kafkak tpoic
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"string","optional":true,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"int32","optional":true,"field":"no"}],"optional":true,"name":"mag.public.mag_tables.Value","field":"before"},{"type":"struct","fields":[{"type":"string","optional":true,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"int32","optional":true,"field":"no"}],"optional":true,"name":"mag.public.mag_tables.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"}],"optional":false,"name":"mag.public.mag_tables.Envelope","version":1},"payload":{"before":null,"after":{"id":"666","name":"amr","no":1989},"source":{"version":"2.5.1.Final","connector":"postgresql","name":"mag","ts_ms":1711012855881,"snapshot":"false","db":"mag","sequence":"["26941288","26941576"]","schema":"public","table":"mag_tables","txId":800,"lsn":26941576,"xmin":null},"op":"c","ts_ms":1711012855904,"transaction":null}}
% Reached end of topic mag.public.mag_tables [0] at offset 62

============================

on clickhouse
┌─id─┬─name─┬─no─┐
│ │ │ 0 │

[Paultagoras]

Hi @elmaghrabii, I would suggest trying our latest version (1.1.0) to see if that resolves the issue - if it doesn't, please let us know 🙂