Sink connector fails to convert negative epoch values

[achunduru1]

Description

COSMOS Sink connector is failing to convert negative epoch values when it is part of composite primary key.

Kafka topic key schema :

{"subject":"TableA-key","version":1,"id":51,"schema":"{"type":"record","name":"ConnectDefault","namespace":"io.confluent.connect.avro","fields":[{"name":"TEST_CK","type":"long"},{"name":"TEST_VALUE","type":"string"},{"name":"TEST_EFF_DT","type":{"type":"long","connect.version":1,"connect.name":"org.apache.kafka.connect.data.Timestamp","logicalType":"timestamp-millis"}}]}"}

Message on Kafka Topic {key} {value}

{"TEST_CK":100,"TEST_VALUE":"R","TEST_EFF_DT":-1577905200000} {"TEST_CK":100,"TEST_VALUE":"R","TEST_EFF_DT":-1577905200000,.... other field values}

Key value after data moved to COSMOS container using COSMOS Sink Connector

"id": "{"TEST_CK":100,"TEST_VALUE":"R","TEST_EFF_DT":00:00:00.000Z}",
"TEST_CK": 100,
"TEST_VALUE": "R",
"TEST_EFF_DT": -1577905200000,
... other field values

Expected Behavior

Negative epoch values(ex: TEST_EFF_DT field) should get converted to correct date time after sink connector move data to COSMOS container.

Additional Context

We set log mode to TRACE\DEBUG but we didn't see any error on connector logs saying that it is failing to convert negative epoch values.

[TheovanKraay]

This is not a bug.

Please see https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/connect/data/Timestamp.html

A timestamp representing an absolute time, without timezone information. The corresponding Java type is a java.util.Date. The underlying representation is a long representing the number of milliseconds since Unix epoch.

You are attempting to convert a negative number into Unix Epoch - this is not possible.

If you need to preserve the negative number, one option might be to simplify the schema definition that field as a long, e.g:

{"name": "TEST_EFF_DT","type": "long"}

Then you can convert to whatever you need to on the client after reading back the value.

If you really need the document to be updated in Cosmos DB with something other the negative value, you can consider using a post trigger, see documentation here.

But, this is not going to work using org.apache.kafka.connect.data.Timestamp.