Mohith26 opened a new pull request, #20839:
URL: https://github.com/apache/druid/pull/20839
Fixes #18578.
### Description
`KafkaInputFormat` parses the Kafka key with `keyFormat`, but only keeps the
first field of the parsed key (as `keyColumnName`). Topics with structured
keys, such as a JSON or Avro key with a tenant id and an entity id, lose every
field after the first one, so users have to duplicate key fields into the
payload or pre-process the topic.
This adds an optional `keyColumnPrefix` property to the `kafka` input
format. When it is set, every field of the parsed key is ingested as its own
column, named `keyColumnPrefix + fieldName`:
```json
"inputFormat": {
"type": "kafka",
"keyFormat": { "type": "json" },
"keyColumnPrefix": "kafka.key.",
"valueFormat": { "type": "json" }
}
```
A key of `{"tenant": "acme", "userId": 42}` becomes the columns
`kafka.key.tenant` and `kafka.key.userId`.
Behavior details:
- Not set (default): unchanged. Only the first key field is used, as
`keyColumnName`. Specs that don't set the property serialize exactly as before
(`@JsonInclude(NON_NULL)`).
- A `flattenSpec` on `keyFormat` is applied first, so flattened fields are
included.
- Precedence matches the existing key column: header and record-metadata
columns win over key columns with the same name, and payload columns win over
both.
- Prefixed key columns can be used as the `timestampSpec` column.
- `sample()` returns the same key columns as `read()` (it uses the parsed
key row, so flattened fields show up in the sampler too).
- Null keys add no key columns.
Alternative considered: changing the default to ingest all key fields. I
kept the default as is because it would change the schema of existing
datasources on upgrade, and an empty prefix (`""`) already allows unprefixed
key columns for users who want them.
#### Release note
The Kafka input format has a new optional `keyColumnPrefix` property. When
set, Druid ingests every field of a structured Kafka key (for example a
multi-field JSON or Avro key) as its own column instead of only the first field.
<hr>
##### Key changed/added classes in this PR
* `KafkaInputFormat`
* `KafkaInputReader`
<hr>
This PR has:
- [x] been self-reviewed.
- [x] added documentation for new or modified features or behaviors.
- [x] a release note entry in the PR description.
- [x] added Javadocs for most classes and all non-trivial methods. Linked
related entities via Javadoc links.
- [x] added comments explaining the "why" and the intent of the code
wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths,
ensuring the threshold for [code
coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md)
is met.
Testing: `mvn -pl extensions-core/kafka-indexing-service test` (checkstyle
and forbidden-apis pass). `KafkaInputFormatTest` has 9 new tests: serde,
multi-field JSON key, key `flattenSpec`, CSV key, timestamp from a prefixed key
field, precedence, null key, and sampling with and without a flattenSpec. The
Testcontainers-based tests in the module (`KafkaIndexTaskTest`,
`KafkaSamplerSpecTest`, the embedded supervisor tests) need Docker, which I
don't have locally, so those rely on CI. The web console doesn't expose the new
field yet; I can follow up with that if this approach looks right.
This change was written with help from an AI coding assistant; I reviewed
the code and ran the tests above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
