Kafka-connect-jdbc: Question: Converting timestamp long value into string literal datetime expression

Created on 4 Jul 2019  路  11Comments  路  Source: confluentinc/kafka-connect-jdbc

Hi,

I'm trying to set up a realtime migration pipeline using Debezium + kafka-connect-jdbc sink connector. There is an issue in converting MySQL's DATETIME(6) column into PostgreSQL's TIMESTAMP type column, so I'm looking for your wisdom.

Let me explain the background a bit.

Description of the issue

Source Database
MySQL 5.6

CREATE TABLE test_datetime (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  dt DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
);

The dt DATETIME(6) column above means a datetime with microseconds precision.

When I insert a row into the test_datetime table like this: insert into test_datetime values (5, '2016-06-15 23:12:35.123897');, then Debezium captures it and produce a CDC data to a kafka topic.

Debezium CDC Data

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int64",
            "optional": false,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "default": 0,
            "field": "dt"
          }
        ],
        "optional": true,
        "name": "debezium.cdc.test_datetime.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int64",
            "optional": false,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "default": 0,
            "field": "dt"
          }
        ],
        "optional": true,
        "name": "debezium.cdc.test_datetime.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": true,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_sec"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "boolean",
            "optional": true,
            "default": false,
            "field": "snapshot"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "debezium.cdc.test_datetime.Envelope"
  },
  "payload": {
    "before": null,
    "after": {
      "id": 5,
      "dt": 1466032355123897
    },
    "source": {
      "version": "0.8.3.Final",
      "name": "debezium",
      "server_id": 223344,
      "ts_sec": 1562097541,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 619,
      "row": 0,
      "snapshot": false,
      "thread": 15,
      "db": "cdc",
      "table": "test_datetime",
      "query": null
    },
    "op": "c",
    "ts_ms": 1562097541760
  }
}

As you can see above, the generated value is a long type of which value is 1466032355123897. It is a unix timestamp with microseconds precision.

Then kafka-connect-jdbc sink connector tries to insert the above cdc data into my sink postgresql database below by generating this query: INSERT INTO test_datetime (id,dt) VALUES (5,1466032355123897) ON CONFLICT (id) DO UPDATE SET dt=EXCLUDED.dt

Sink Database
PostgreSQL 11.4

CREATE TABLE test_datetime (
  id SERIAL PRIMARY KEY,
  dt TIMESTAMP NOT NULL
);

However PostgreSQL cannot convert (or cast) this long type value 1466032355123897 into a TIMESTAMP type so I'm getting errors like following:

Caused by: java.sql.SQLException: java.sql.BatchUpdateException: Batch entry 0 INSERT INTO test_datetime (id,dt) VALUES (5,1466032355123897) ON CONFLICT (id) DO UPDATE SET dt=EXCLUDED.dt was aborted: ERROR: column "dt" is of type timestamp without time zone but expression is of type bigint
  Hint: You will need to rewrite or cast the expression.
  Position: 46  Call getNextException to see other errors in the batch.
org.postgresql.util.PSQLException: ERROR: column "dt" is of type timestamp without time zone but expression is of type bigint
  Hint: You will need to rewrite or cast the expression.
  Position: 46
org.postgresql.util.PSQLException: ERROR: column "dt" is of type timestamp without time zone but expression is of type bigint
  Hint: You will need to rewrite or cast the expression.
  Position: 46

Question

Can kafka-connect-jdbc sink connector convert a long type timestamp value into a string literal datetime expression?

For example: INSERT INTO test_datetime (id,dt) VALUES (5, '2016-06-15 23:12:35.123897') instead of INSERT INTO test_datetime (id,dt) VALUES (5, 1466032355123897)

picture myungjae1
👍1

Most helpful comment

Hey @mjaewon

You must have added ExtractNewRecordState or UnwrapFromEnvelope SMT in your sink connector to flatten the message.If you do not require addition information provided by debezium, you can have this SMT at source.
The problem here is, after flattening the message in sink connector, schema information is lost for sink. For e.g. schema has information about before.create_date and after.create_date but not for create_date.

Also, try adding/setting time.precision.mode=connect as kafka jdbc sink does not interpret Debezium DateTime Types. (Figured out from https://debezium.io/documentation/reference/1.0/connectors/mysql.html and https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/dialect/PostgreSqlDatabaseDialect.java)

I had similar problem while converting SQL Server DATETIME to MySQL DATETIME(3).

Here is my connector details:

Source: SQL Server

{
  "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "database.user": "admin",
  "database.dbname": "TEST",
  "tasks.max": "1",
  "database.history.kafka.bootstrap.servers": "localhost:9092",
  "database.history.kafka.topic": "sqlserver-cdc-dbhistory",
  "transforms": "route,unwrap",
  "time.precision.mode": "connect",
  "database.server.name": "SQL1",
  "database.port": "1433",
  "table.whitelist": "dbo.kafka_datatype",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
  "database.hostname": "host",
  "transforms.unwrap.drop.tombstones": "false",
  "database.password": "***",
  "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
  "transforms.route.replacement": "$3"
}

Sink: MySQL

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "delete.enabled": "true",
  "connection.password": "***",
  "auto.evolve": "false",
  "topics": "kafka_datatype",
  "tasks.max": "1",
  "connection.user": "admin",
  "auto.create": "true",
  "connection.url": "jdbc:mysql://localhost:3306/kafka_test",
  "insert.mode": "upsert",
  "pk.mode": "record_key"
}

Sample message

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "field": "id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "col_value"
      },
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "field": "create_date"
      }
    ],
    "optional": true,
    "name": "SQL1.dbo.kafka_datatype.Value"
  },
  "payload": {
    "id": 1,
    "col_value": "ravi",
    "create_date": 1578923527343
  }
}

Hope it helps!!

Thanks
Ravi

picture desairavi1101 on 14 Jan 2020
👍3 ❤1

All 11 comments

Take a look at https://docs.confluent.io/current/connect/transforms/timestampconverter.html#timestampconverter

aliasbadwolf picture aliasbadwolf on 31 Jul 2019

I already had a look at it, but Confluent's TimestampConverter is internally using SimpleDateFormat of Java, which does not support microseconds precision. Thank you though, @aliasbadwolf .

myungjae1 picture myungjae1 on 31 Jul 2019

@mjaewon did you manage to solve it using TimestampConverter?

viniciusbo picture viniciusbo on 7 Oct 2019
👀3

@mjaewon That is correct. Guess I didn't looked at the timestamp you were trying to convert.

Well another thing I can suggest, if you wish to stay within Confluent Platform components, (which is not as clean as having this capability out of the box) is to created a KStream application which will read from the topic having time as "long" and convert it to Formatted String using Java8 Date/Time libraries and push the data to a new topic which then can be consumed by the Sink Connector.

aliasbadwolf picture aliasbadwolf on 8 Oct 2019

@mjaewon Did you find any workaround for this problem?

alfishan-agile picture alfishan-agile on 12 Nov 2019

Hey @mjaewon

You must have added ExtractNewRecordState or UnwrapFromEnvelope SMT in your sink connector to flatten the message.If you do not require addition information provided by debezium, you can have this SMT at source.
The problem here is, after flattening the message in sink connector, schema information is lost for sink. For e.g. schema has information about before.create_date and after.create_date but not for create_date.

Also, try adding/setting time.precision.mode=connect as kafka jdbc sink does not interpret Debezium DateTime Types. (Figured out from https://debezium.io/documentation/reference/1.0/connectors/mysql.html and https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/dialect/PostgreSqlDatabaseDialect.java)

I had similar problem while converting SQL Server DATETIME to MySQL DATETIME(3).

Here is my connector details:

Source: SQL Server

{
  "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "database.user": "admin",
  "database.dbname": "TEST",
  "tasks.max": "1",
  "database.history.kafka.bootstrap.servers": "localhost:9092",
  "database.history.kafka.topic": "sqlserver-cdc-dbhistory",
  "transforms": "route,unwrap",
  "time.precision.mode": "connect",
  "database.server.name": "SQL1",
  "database.port": "1433",
  "table.whitelist": "dbo.kafka_datatype",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
  "database.hostname": "host",
  "transforms.unwrap.drop.tombstones": "false",
  "database.password": "***",
  "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
  "transforms.route.replacement": "$3"
}

Sink: MySQL

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "delete.enabled": "true",
  "connection.password": "***",
  "auto.evolve": "false",
  "topics": "kafka_datatype",
  "tasks.max": "1",
  "connection.user": "admin",
  "auto.create": "true",
  "connection.url": "jdbc:mysql://localhost:3306/kafka_test",
  "insert.mode": "upsert",
  "pk.mode": "record_key"
}

Sample message

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "field": "id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "col_value"
      },
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "field": "create_date"
      }
    ],
    "optional": true,
    "name": "SQL1.dbo.kafka_datatype.Value"
  },
  "payload": {
    "id": 1,
    "col_value": "ravi",
    "create_date": 1578923527343
  }
}

Hope it helps!!

Thanks
Ravi

desairavi1101 picture desairavi1101 on 14 Jan 2020
👍3 ❤1

Hey @mjaewon, did you ever find a solution or workaround to this?

The other comments talk about using the TimestampConverter but those options result in a loss of precision.

jayctran picture jayctran on 23 Mar 2020

@desairavi1101 Thank you so much, your answer helped me a lot.

mataralhawiti picture mataralhawiti on 5 Oct 2020

Hey @mjaewon

You must have added ExtractNewRecordState or UnwrapFromEnvelope SMT in your sink connector to flatten the message.If you do not require addition information provided by debezium, you can have this SMT at source.
The problem here is, after flattening the message in sink connector, schema information is lost for sink. For e.g. schema has information about before.create_date and after.create_date but not for create_date.

Also, try adding/setting time.precision.mode=connect as kafka jdbc sink does not interpret Debezium DateTime Types. (Figured out from https://debezium.io/documentation/reference/1.0/connectors/mysql.html and https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/dialect/PostgreSqlDatabaseDialect.java)

I had similar problem while converting SQL Server DATETIME to MySQL DATETIME(3).

Here is my connector details:

Source: SQL Server

{
  "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "database.user": "admin",
  "database.dbname": "TEST",
  "tasks.max": "1",
  "database.history.kafka.bootstrap.servers": "localhost:9092",
  "database.history.kafka.topic": "sqlserver-cdc-dbhistory",
  "transforms": "route,unwrap",
  "time.precision.mode": "connect",
  "database.server.name": "SQL1",
  "database.port": "1433",
  "table.whitelist": "dbo.kafka_datatype",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
  "database.hostname": "host",
  "transforms.unwrap.drop.tombstones": "false",
  "database.password": "***",
  "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
  "transforms.route.replacement": "$3"
}

Sink: MySQL

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "delete.enabled": "true",
  "connection.password": "***",
  "auto.evolve": "false",
  "topics": "kafka_datatype",
  "tasks.max": "1",
  "connection.user": "admin",
  "auto.create": "true",
  "connection.url": "jdbc:mysql://localhost:3306/kafka_test",
  "insert.mode": "upsert",
  "pk.mode": "record_key"
}

Sample message

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "field": "id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "col_value"
      },
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "field": "create_date"
      }
    ],
    "optional": true,
    "name": "SQL1.dbo.kafka_datatype.Value"
  },
  "payload": {
    "id": 1,
    "col_value": "ravi",
    "create_date": 1578923527343
  }
}

Hope it helps!!

Thanks
Ravi

This configuration did not work for Oracle source connector

mujahid picture mujahid on 8 Nov 2020

If you write a custom converter similarly to https://github.com/oryanmoshe/debezium-timestamp-converter, you can just return the original datetime string that comes from the binlogs, without any conversion. If you have microsecond precision in the source, the value would look like 2020-11-19 12:46:27.123456, you just need to return it as it is. The sink will be able to insert it to the database (at least for MySql, not sure about others).

zstoth picture zstoth on 19 Nov 2020

Finally leaving a reply to this long overdue task. Sorry for being late.

As @desairavi1101 mentioned, I was able to make this work by the changes below. Thanks again @desairavi1101 for pointing this out.

test_datetime.diff:

diff --git a/unwrap-smt/jdbc-sink.json b/unwrap-smt/jdbc-sink.json
index fdf98a9..f5d142d 100644
--- a/unwrap-smt/jdbc-sink.json
+++ b/unwrap-smt/jdbc-sink.json
@@ -3,11 +3,8 @@
     "config": {
         "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
         "tasks.max": "1",
-        "topics": "customers",
+        "topics": "test_datetime",
         "connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=postgrespw",
-        "transforms": "unwrap",
-        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
-        "transforms.unwrap.drop.tombstones": "false",
         "auto.create": "true",
         "insert.mode": "upsert",
         "delete.enabled": "true",
diff --git a/unwrap-smt/source.json b/unwrap-smt/source.json
index 14e5872..920e07f 100644
--- a/unwrap-smt/source.json
+++ b/unwrap-smt/source.json
@@ -12,9 +12,13 @@
         "database.include": "inventory",
         "database.history.kafka.bootstrap.servers": "kafka:9092",
         "database.history.kafka.topic": "schema-changes.inventory",
-        "transforms": "route",
+        "table.whitelist": "inventory.test_datetime",
+        "transforms": "route,unwrap",
         "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
         "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
-        "transforms.route.replacement": "$3"
+        "transforms.route.replacement": "$3",
+        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
+        "transforms.unwrap.drop.tombstones": "false",
+        "time.precision.mode": "connect"
     }
 }

My test is based on this great example code: https://github.com/debezium/debezium-examples (Big shout out to debezium authors!)

Exact reproducing steps are attached below:

git clone https://github.com/debezium/debezium-examples
git apply test_datetime.diff

# setup
cd debezium-examples/unwrap-smt
DEBEZIUM_VERSION=1.4 docker-compose -f docker-compose-jdbc.yaml up

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

# create table & insert row in the source mysql
docker exec -it unwrap-smt_mysql_1 bash -c 'exec mysql -u $MYSQL_USER  -p$MYSQL_PASSWORD'
mysql> CREATE TABLE inventory.test_datetime ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, dt DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) );
mysql> INSERT INTO inventory.test_datetime VALUES (5, '2016-06-15 23:12:35.123897');

# check table & row exist in the sink postgresql
# you should be able to see results like below
docker exec -it unwrap-smt_postgres_1 bash -c 'exec psql -U $POSTGRES_USER $POSTGRES_DB'
inventory=# \d test_datetime;
                                   Table "public.test_datetime"
 Column |            Type             |                         Modifiers
--------+-----------------------------+------------------------------------------------------------
 dt     | timestamp without time zone | default '1970-01-01 00:00:00'::timestamp without time zone
 id     | integer                     | not null
Indexes:
    "test_datetime_pkey" PRIMARY KEY, btree (id)

inventory=# select * from test_datetime;
           dt            | id
-------------------------+----
 2016-06-15 23:12:35.123 |  5
(1 row)

Closing this issue.

myungjae1 picture myungjae1 on 22 Feb 2021
Was this page helpful?
0 / 5 - 0 ratings

Related issues

maeh2k picture maeh2k  路  10Comments
jfinzel picture jfinzel  路  6Comments
venkrishy picture venkrishy  路  5Comments
bli24 picture bli24  路  6Comments
rdinkel picture rdinkel  路  8Comments