Kafka-connect-jdbc: Timestamp field in InsertField transformation is always null

[jfconavarrete]

Hello,

I'm trying to add a timestamp field to every record read from a JDBC source connector using Kafka Connect's org.apache.kafka.connect.transforms.InsertField. The timestamp field does appear in the record, however it is always null.

Here is my connector's config:

{
  "name": "test-mysql-source1",
  "config": {
    "group.id": "test-mysql-group1",
    "tasks.max": "1",
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://172.17.0.7:3306/accounting?user=root&password=root",
    "mode": "bulk",
    "topic.prefix": "test-mysql-accounting-",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "internal.key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "internal.value.converter": "org.apache.kafka.connect.json.JsonConverter"
    "transforms": "InsertSource",
    "transforms.InsertSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertSource.static.field": "SMT_DATABASE",
    "transforms.InsertSource.static.value": "XE",
    "transforms.InsertSource.timestamp.field": "SMT_TIMESTAMP"
  }
}

And here is the resulting record:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "field": "id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "name"
      },
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "field": "SMT_TIMESTAMP"
      },
      {
        "type": "string",
        "optional": true,
        "field": "SMT_DATABASE"
      }
    ],
    "optional": false,
    "name": "accounts"
  },
  "payload": {
    "id": 2,
    "name": "bob",
    "SMT_TIMESTAMP": null,
    "SMT_DATABASE": "XE"
  }
}

I'm using Kafka 0.10.2.0

Any help would be greatly appreciated, thanks!

[bradacina]

+1

[hazefully]

I ran into the same issue. I tracked it down and I think I understand why this happens.

The InsertField populates the timestamp field using this line of code:

if (timestampField != null && record.timestamp() != null) {
    updatedValue.put(timestampField.name, new Date(record.timestamp()));

Therefore, when a record is first created it should have its timestamp field populated, the records are created in the classes implementing the TableQuerier interface (for now the BulkTableQuerier and TimestampIncrementingTableQuerier). For example, the BulkTableQuerier uses this line of code:
return new SourceRecord(partition, null, topic, record.schema(), record);

I think this can be fixed easily by just adding a timestamp to the the line of code that generates records as follows:

return new SourceRecord(partition, null, topic, record.schema(), record, System.currentTimeMillis());

I am more than happy to make the PR for this.

[thomas-chauvet]

Hello,

I know this is a pretty old issue, but is there a workaround? I am struggling with the same issue.

Thanks a lot!

Here is my connector configuration:

"transforms": "InsertSource,InsertTopic,InsertTimestamp",
"transforms.InsertSource.type": "org.apache.kafka.connect.transforms.InsertField\$Value",
"transforms.InsertSource.static.field": "message_source",
"transforms.InsertSource.static.value": "sql_connector",
"transforms.InsertTopic.type":"org.apache.kafka.connect.transforms.InsertField\$Value",
"transforms.InsertTopic.topic.field":"message_topic",
"transforms.InsertTimestamp.type": "org.apache.kafka.connect.transforms.InsertField\$Value",
"transforms.InsertTimestamp.timestamp.field": "message_timestamp"

[ncliang]

I would like to understand what the use-case of this timestamp would be. Note that the timestamp added would not be the time which the row was inserted into the database, or when broker received the record, it would be the timestamp at which the connector happened to poll and find the row. This could differ from the actual insertion/update time by up to a polling interval. Or, if it is the initial scan, the entire lifetime of the table. It would be more or less just an arbitrary timestamp. I think having the insertion/update time recorded on the column itself would be much more useful.

Note that the timestamp is also optional on the SourceRecord itself. By not filling it in, I feel we are reducing confusion that an arbitrary timestamp might cause downstream when processing the record. Would like to hear why having this timestamp is desired from others on this thread.