[Mohith26]

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:

"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.

---

Key changed/added classes in this PR

• KafkaInputFormat
• KafkaInputReader

---

This PR has:

• been self-reviewed.
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage 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.

[FrankChen021]

🟡 Changes recommended

Preserve the existing public constructor signatures before merging. Adding the optional prefix currently breaks direct Java or custom-extension callers even when keyColumnPrefix is unset.

Reviewed 7 of 7 changed files; findings in 1 file.

Validation: git diff --no-ext-diff --check 7e300b9ccf6cca26d7bbe632f790415b31f123c1 d2ff79db70f2b5c398a10247e23ee6cf77b2094b passed. No build or tests were run.

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

---

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

[FrankChen021]

[P2] Preserve the existing public constructor

Finding: Appending keyColumnPrefix removes the previous public KafkaInputFormat constructor signature, so Java callers compiled against the prior release can fail with NoSuchMethodError or fail to compile even when they do not use the new option. The corresponding public KafkaInputReader constructor is changed in the same way.

Suggestion: Retain overloads for the previous KafkaInputFormat and KafkaInputReader signatures, delegating to the new constructors with a null prefix.

[FrankChen021]

can ignore this finding from AI