vbhanuchander-lang opened a new pull request, #7770:
URL: https://github.com/apache/hop/pull/7770
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:
```json
[{"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:
- [x] 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)*
- [x] 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)*
- [x] 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](http://www.apache.org/licenses/LICENSE-2.0)
you have to acknowledge this by using the following check-box.
- [x] I hereby declare this contribution to be licensed under the [Apache
License Version 2.0, January 2004](http://www.apache.org/licenses/LICENSE-2.0)
- [ ] In any other case, please file an [Apache Individual Contributor
License Agreement](https://www.apache.org/licenses/icla.pdf).
--
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]
