Using JSON Schema and passing datetime fields to Kafka Connect

DevOps Kafka
#1

Has anyone any pointers on using JSON Schema and passing through date/datetime fields to Kafka Connect to build the columns as a datetime and not a string?

#2

What is generating the json schema based messages? Are you using the jdbc sink connector, whats the destination db type?

#3

We have a custom python producer building the schema from a description. this is an example datetime field…

#4

The message is just standard json…

"2020-09-24T15:29:45.000+0000",
CreatedById:
"0054J000001GvU3QAK",
LastModifiedDate:
"2021-04-07T13:16:55.000+0000",
LastModifiedById:
"0054J0000012GKOQA2",
SystemModstamp:
"2021-06-01T09:28:44.000+0000"```
#5

What connector are you using?

#6

Jdbc sink going to a mysql DB

#7

Going to see what a source connector does with json schema, maybe that will help point in the right direction

#8

Ok, many thanks for your help…

#9

The jdbc source connector creates a schema like this

  "type" : "object",
  "title" : "outbox",
  "properties" : {
    "date_added" : {
      "connect.index" : 1,
      "oneOf" : [ {
        "type" : "null"
      }, {
        "type" : "integer",
        "title" : "org.apache.kafka.connect.data.Timestamp",
        "connect.version" : 1,
        "connect.type" : "int64"
      } ]
    },
    "id" : {
      "type" : "string",
      "connect.index" : 0
    }
  }
}```
#10

Which looks like it uses a millisecond epoch (ex: 1622737247818)

#11

If you don’t want to encode kafka specific stuff in your json schema i think the next best option would be to use an SMT, i think you can use the TimestampConverter

#12

SMT is probably the preferred method

#13

Something like this

"transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.TimestampConverter.format": "yyyy-MM-dd'T'HH:mm:ss.SSSz",
"transforms.TimestampConverter.target.type": "Timestamp",
"transforms.TimestampConverter.field": "SystemModstamp"```
#14

Thanks. Luckily all the tables have system fields so I can use this on those, but I was hoping not to have to have a custom option for the non-system Fields.