Issue #6408 : Read and write Kafka record headers

[vbhanuchander-lang]

Addresses #6408.

The Kafka transforms expose key, message, topic, partition, offset and timestamp, but not the record headers — so anything carried in a header (trace context, schema hints, routing or tenancy metadata) is unreachable on the way in and impossible to set on the way out. As the reporter put it, that rules Hop out of a fair number of Kafka use cases.

This adds headers to both sides.

Exchange format

A record carries an ordered list of header pairs and the same name may appear more than once, neither of which a flat row column can express. The value is therefore a JSON array of {"name":..,"value":..} objects rather than a JSON object:

[{"name":"traceparent","value":"00-abc"},{"name":"tag","value":"one"},{"name":"tag","value":"two"}]

A JSON object would have been friendlier to hand-write, but it silently drops repeats and loses ordering. The array is lossless, which means the consumer's output can be fed straight into the producer's Headers field and the headers come out unchanged — including a null value, which stays distinct from an empty one.

Both directions live in one new shared/KafkaHeaders class rather than in either transform, because the format is a contract between them; keeping them together is what stops them drifting apart. The round trip is covered by a test.

Consumer

A new optional headers output field. It ships with an empty output name, and getRowMeta already omits fields without one, so a pipeline saved before this change produces exactly the row it did before. Give it a name to start reading headers.

Producer

A new optional Headers field. Left empty, messages are sent without headers, so existing pipelines are unaffected. A field that is named but missing from the input stream fails when the first row arrives — matching how the key, message and topic fields already behave — and a value that is not a JSON array of name/value objects fails with the offending document in the message.

Two things I had to fix to make this work

The consumer's row builder used fixed indices. putFieldOnRowMeta omits any field whose output name is empty, so the output row is variable length, but processMessageAsRow wrote its six values at indices 0–5 regardless. Clearing the name of, say, the topic field was enough to shift every later column by one. Values are now placed only for fields that contribute a column, in the order getRowMeta adds them. That is a pre-existing defect rather than anything this feature introduced, but an optional field cannot sit on top of fixed indices, so it had to be fixed here.

The consumer's output field table was read-only. It was already built with an editable name column and a setDisabledListener restricting type changes to key/message, but the table itself was flagged setReadonly(true) and the write-back left as // meta.setField(field); TODO FIXME, so field names could not actually be changed from the dialog. That would have left the new Headers field impossible to switch on except through metadata injection. The write-back is now implemented and the table editable — which the indexing fix above makes safe, since clearing a name now correctly drops the column instead of corrupting the row.

Happy to split either of those into their own PR if you would rather keep this one to the feature.

Tests

15 cases in KafkaHeadersTest covering rendering, parsing and the round trip: ordering, repeated names, null vs. empty values, JSON escaping, non-ASCII, and the rejection cases (not JSON, a JSON object, an entry without a name). Plus the existing suite — 32 tests pass in the module.

mvn -pl plugins/transforms/kafka test, spotless:check and apache-rat:check (Unapproved: 0) all pass.

Documentation

Both kafkaconsumer.adoc and kafkaproducer.adoc are updated, including a note that an empty output name keeps a field off the row.

Not included

I have not added an integration test under integration-tests/kafka for this one yet — happy to add a produce-with-headers / consume-and-assert workflow in the same shape as 0004-kafka-test-topic-from-field if you would like it before merge.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Run mvn clean install apache-rat:check to make sure basic checks pass. A more thorough check will be performed on your pull request automatically. (ran the module build, spotless:check and apache-rat:check; full clean install not run locally)
• If you have a group of commits related to the same change, please squash your commits into one and force push your branch using git rebase -i. (two commits kept deliberately: read side, then write side — happy to squash)
• Mention the appropriate issue in your description (for example: addresses #123), if applicable.

To make clear that you license your contribution under the Apache License Version 2.0, January 2004
you have to acknowledge this by using the following check-box.

• I hereby declare this contribution to be licensed under the Apache License Version 2.0, January 2004
• In any other case, please file an Apache Individual Contributor License Agreement.