This message was deleted.

I think this is already supported if you use the

kafka

Input Format. You can definitely get headers and stuff. I don't think it will automatically use the kafka timestamp as the timestamp field, but you could probably make that work with a transformspec. https://druid.apache.org/docs/latest/development/extensions-core/kafka-ingestion.html#kafka-input-format-supervisor-spec-example

👆 and AFAICT you can use it as the primary timestamp by referencing it with

kafka.timestamp

Oh man, I scoured the documentation for this and couldn't find it. Thanks for the direct link!

😁 It is fairly new, 0.23 or 24, if I remember correctly.

I'm using 0.23. I replaced:

"inputFormat": {
        "type": "json"
      }

With the example from the documentation:

"inputFormat": {
        "type": "kafka",
        "headerLabelPrefix": "kafka.header.",
        "timestampColumnName": "kafka.timestamp",
        "keyColumnName": "kafka.key",
        "headerFormat": {
          "type": "string"
        },
        "keyFormat": {
          "type": "json"
        },
        "valueFormat": {
          "type": "json"
        },
        "findColumnsFromHeader": false
      }

Unfortunately when I do this it breaks the UI in 0.23 (it doesn't know how to parse it as kafka format). I also notice kafka is not a valid Input format on the "Parse data" tab in 0.23. I pushed it through but I just get errors:

org.apache.druid.java.util.common.parsers.ParseException: Unable to parse row [fb5d5708-6a51-4334-afd7-217e1f61ecd2] (Record: 1, Line: 1)

So I might have to upgrade my version of Druid before I try this again 😞

based on the release notes it looks like support was added for the kafka format in the web console in 26.0.0. it probably still works if you can submit to the task endpoint

I think this is the relevant PR (which was merged into 0.23.0) https://github.com/apache/druid/pull/11630, so you should be able to submit the spec programmatically, but not through the UI which caught up later.