Kafka - migrating date from sqlserver (type: [datetime2](7)) to cockroach database table with "Timestamp" column

Question

We have data stored in SQL Server and we are in need to migrate these data into cockroach database.

For this requirement, we are trying to set up a Kafka Pipeline to stream the data from source database to target database.

At source end, we have Debezium connector and at target end, we have JDBC Sink connector

We are facing the issue for migrating the Datetime2 column from SQLServer.

[created_on] [datetime2](7) NOT NULL,

At target table, the associated column is Timestamp typed.

created_on TIMESTAMP NOT NULL DEFAULT timezone('utc':::STRING, now():::TIMESTAMPTZ)

Initially we have tried to use TimestampConverter transform at JDBC sink connector to convert the date before inserting it into the target table.

transforms.myTimestampConverter1.type: "org.apache.kafka.connect.transforms.TimestampConverter$Value"
transforms.myTimestampConverter1.field: "created_on"
transforms.myTimestampConverter1.target.type: "Timestamp"
transforms.myTimestampConverter1.format: "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"

This works for Datetime columns where the typical date would look like "2024-05-30 06:29:06.603" But not for the Datetime2(7) columns.

The other data is moving fine but we are facing the issue for a date column (type Datetime2(7)) . The date in this column looks like "2024-05-30 06:29:06.6033333"

While inserting the data into target table, we are facing the error

ERROR: error in argument for $3: could not parse "54413135-12-02 11:26:40+00" as type timestamp: timestamp "54413135-12-02T11:26:40Z" exceeds supported timestamp bounds

the whole message from deadletterqueue:

"Exception chain:\njava.sql.BatchUpdateException: Batch entry 0 
INSERT INTO \"public\".\"users\" 
(\"id\",\"created_by\",\"created_on\",\"tenant_id\",\"updated_by\",\"updated_on\",\"version\",\"status\",\"alias\",\"email\",\"locale\",\"user_name\",\"description\",\"is_deleted\",\"fail_login_attempts\",\"disabled\",\"user_type\",\"tenant_uuid\") 
VALUES 
    (994,1,'54413135-12-02 11:26:40+00',1,1,'54413135-12-02 11:26:40+00',1,NULL,NULL,NULL,NULL,'user994','','FALSE',0,'FALSE','HUMAN','E100FBCE-008C-04EC-4063-7AF0AF91FB2F'::uuid) 
    ON CONFLICT (\"id\") DO UPDATE SET \"created_by\"=EXCLUDED.\"created_by\",\"created_on\"=EXCLUDED.\"created_on\",\"tenant_id\"=EXCLUDED.\
    "tenant_id\",\"updated_by\"=EXCLUDED.\"updated_by\",\"updated_on\"=EXCLUDED.\"updated_on\",\"version\"=EXCLUDED.\"version\",\"status\"=EXCLUDED.
    \"status\",\"alias\"=EXCLUDED.\"alias\",\"email\"=EXCLUDED.\"email\",\"locale\"=EXCLUDED.\"locale\",\"user_name\"=EXCLUDED.\"user_name\",\"description
    \"=EXCLUDED.\"description\",\"is_deleted\"=EXCLUDED.\"is_deleted\",\"fail_login_attempts\"=EXCLUDED.\"fail_login_attempts\",\"disabled\"=EXCLUDED.\"disabled\",
    \"user_type\"=EXCLUDED.\"user_type\",\"tenant_uuid\"=EXCLUDED.\"tenant_uuid\" was aborted: 
ERROR: error in argument for $3: could not parse \"54413135-12-02 11:26:40+00\" as type timestamp: timestamp \"54413135-12-02T11:26:40Z\" exceeds supported timestamp bounds\n  
Detail: statement summary \"INSERT INTO public.users(id, created_by,...)\"  
    Call getNextException to see other errors in the batch.\norg.postgresql.util.PSQLException: ERROR: error in argument for $3: 
    could not parse \"54413135-12-02 11:26:40+00\" as type timestamp: timestamp \"54413135-12-02T11:26:40Z\" exceeds supported timestamp bounds\n  
    Detail: statement summary \"INSERT INTO public.users(id, created_by,...)\"\norg.postgresql.util.PSQLException: ERROR: error in argument for $3: 
    could not parse \"54413135-12-02 11:26:40+00\" as type timestamp: timestamp \"54413135-12-02T11:26:40Z\" exceeds supported timestamp bounds\n  
    Detail: statement summary \"INSERT INTO public.users(id, created_by,...)\"\n"

please note, if we don't have the Datetime2 column in the table, the data moves fine with no error.

One more observation i had when looking into the kafka topic, the date fields are converted into long (unix format) and dates of Datetime2(7) column have the long value of 19 digits while Datetime field have value of 13 digits.

Datetime2(7) seems to be quite commonly used datatype for dates, and I am hoping there would be a widely used way to migrate that through kafka.

• Can this be handled using TimestampConverter , if so am I missing anything at configuring it on JDBC sink connector?
• If TimestampConverter can't help, then how can i achieve the migration of this data.
• I would prefer already existing and widely tested way better over writing my own SMT, but if there may nothing else that can help me, is there any way to handle this via custom SMT other than dropping last 6 digits of the date (in unix format)?