This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 39d377c3f5 Issue #6408 : Read and write Kafka record headers (#7770)
39d377c3f5 is described below
commit 39d377c3f5ef77fec3a75f702a23703c61f76be5
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Thu Aug 6 02:27:37 2026 -0400
Issue #6408 : Read and write Kafka record headers (#7770)
* Issue #6408 : Read Kafka record headers into an output field
The Kafka Consumer exposes key, message, topic, partition, offset and
timestamp,
but not the record headers, so anything carried in a header - trace context,
schema hints, routing metadata - is unreachable once a message is consumed.
Add an 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.
Headers are rendered as a JSON array of {"name":..,"value":..} objects
rather
than a JSON object. A record carries an ordered list of header pairs and the
same name may appear more than once, so an object would silently drop
repeats
and lose ordering; an array keeps both, which also lets the value round-trip
back out through the producer unchanged. Values are bytes on the wire and
are
decoded as UTF-8, with a null value rendered as JSON null.
Building the row also had to change. putFieldOnRowMeta leaves out any field
whose output name is empty, so the output row is variable length, but
processMessageAsRow wrote its six values at fixed indices regardless.
Clearing
the name of, say, the topic field was therefore 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, which fixes that and is what lets the new
field be optional at all.
* Issue #6408 : Write Kafka record headers from a field
Completes header support: the consumer can now read headers into a field
and the
producer can attach headers read back from one.
The exchange format lives in a new shared KafkaHeaders class rather than in
either transform, because it is a contract between the two: what the
consumer
writes into a field can be handed straight to the producer and yields the
same
headers, including ordering, repeated names, and the difference between a
null
and an empty value. Keeping both directions in one place stops them drifting
apart, and the round trip is covered by a test.
The producer's headers field is optional; leaving it empty sends messages
without headers, so existing pipelines are unaffected. A field that is
named but
absent from the input stream fails when the first row arrives, matching how
the
key, message and topic fields already behave. A value that is not a JSON
array
of name/value objects fails with the offending document in the message.
Also documents both transforms and makes the consumer's output field table
editable. That table was already built with an editable name column and a
disabled-listener limiting type changes, but the table itself was flagged
read-only and the write-back left as a TODO, so field names could not
actually
be changed from the dialog - which would have left the new Headers field
impossible to switch on outside metadata injection.
---
.../pages/pipeline/transforms/kafkaconsumer.adoc | 8 +-
.../pages/pipeline/transforms/kafkaproducer.adoc | 6 +
.../kafka/consumer/KafkaConsumerField.java | 3 +-
.../kafka/consumer/KafkaConsumerInput.java | 29 +++-
.../kafka/consumer/KafkaConsumerInputDialog.java | 30 +++-
.../kafka/consumer/KafkaConsumerInputMeta.java | 46 +++++-
.../kafka/producer/KafkaProducerOutput.java | 48 +++++-
.../kafka/producer/KafkaProducerOutputData.java | 1 +
.../kafka/producer/KafkaProducerOutputDialog.java | 42 ++++++
.../kafka/producer/KafkaProducerOutputMeta.java | 7 +
.../transforms/kafka/shared/KafkaHeaders.java | 115 ++++++++++++++
.../consumer/messages/messages_en_US.properties | 1 +
.../producer/messages/messages_en_US.properties | 4 +
.../shared/messages/messages_en_US.properties | 20 +++
.../transforms/kafka/shared/KafkaHeadersTest.java | 167 +++++++++++++++++++++
15 files changed, 512 insertions(+), 15 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
index 904caff683..94808fa8fb 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
@@ -134,8 +134,14 @@ You cannot have more consumers than the number of
partitions.
* **offset**: A sequential ID number assigned by Kafka to each record.
It uniquely identifies each record within the partition.
* **timestamp**: The time the message is received on the server.
+* **headers**: The record headers, as a JSON array of `{"name":..,"value":..}`
objects.
+An array is used rather than a JSON object because Kafka keeps headers ordered
and allows the same name to appear more than once.
+This value can be fed straight into the _Headers field_ of the Kafka Producer
transform to pass headers through unchanged.
-|Output name|The _Output name_ can be mapped to subscriber and member
requirements.
+|Output name a|The _Output name_ can be mapped to subscriber and member
requirements.
+
+Leave the output name empty to keep that field off the output row.
+**headers** ships with an empty output name, so pipelines created before
header support produce the same rows as before; give it a name to start reading
headers.
|Type a|The Type field defines the data format for streaming the record.
You must choose the same data type that produced the records.
This field applies to the ‘key’ and ‘message’ input names.
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
index fdb99fa38a..a5bcdcfb4e 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
@@ -69,6 +69,12 @@ This lets a single transform route rows across topics,
rather than needing one K
|The field holding the topic name for each row, used when *Get topic from
field* is enabled.
Rows where this field is empty cause the transform to fail.
+|Headers field
+|Optional.
+A field holding the record headers to attach to each message, as a JSON array
of `{"name":..,"value":..}` objects — the same shape the Kafka Consumer
transform produces, so headers survive a consume-then-produce round trip.
+An array is used rather than a JSON object because Kafka keeps headers ordered
and allows a name to repeat.
+Leave empty to send messages without headers.
+
|Key Field
|In Kafka, all messages can be keyed, allowing for messages to be distributed
to partitions based on their keys in a default routing scheme.
If no key is present, messages are randomly distributed to partitions.
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerField.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerField.java
index 627249b145..bb48f64ce2 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerField.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerField.java
@@ -135,7 +135,8 @@ public class KafkaConsumerField {
TOPIC("topic"),
PARTITION("partition"),
OFFSET("offset"),
- TIMESTAMP("timestamp");
+ TIMESTAMP("timestamp"),
+ HEADERS("headers");
private final String code;
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
index 375e7dc521..7288f916c5 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
@@ -43,6 +43,7 @@ import org.apache.hop.pipeline.transform.ITransformMeta;
import org.apache.hop.pipeline.transform.RowAdapter;
import org.apache.hop.pipeline.transform.TransformMeta;
import org.apache.hop.pipeline.transforms.injector.InjectorMeta;
+import org.apache.hop.pipeline.transforms.kafka.shared.KafkaHeaders;
import org.apache.hop.pipeline.transforms.kafka.shared.KafkaOption;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -410,14 +411,30 @@ public class KafkaConsumerInput
Object[] rowData = RowDataUtil.allocateRowData(data.outputRowMeta.size());
+ // Only fields carrying an output name are on the row, in the order
KafkaConsumerInputMeta
+ // adds them, so each value is placed conditionally rather than at a fixed
index.
int index = 0;
- rowData[index++] = record.key();
- rowData[index++] = record.value();
- rowData[index++] = record.topic();
- rowData[index++] = (long) record.partition();
- rowData[index++] = record.offset();
- rowData[index] = record.timestamp();
+ index = putIfNamed(rowData, index, meta.getKeyField(), record.key());
+ index = putIfNamed(rowData, index, meta.getMessageField(), record.value());
+ index = putIfNamed(rowData, index, meta.getTopicField(), record.topic());
+ index = putIfNamed(rowData, index, meta.getPartitionField(), (long)
record.partition());
+ index = putIfNamed(rowData, index, meta.getOffsetField(), record.offset());
+ index = putIfNamed(rowData, index, meta.getTimestampField(),
record.timestamp());
+ putIfNamed(rowData, index, meta.getHeadersField(),
KafkaHeaders.toJson(record.headers()));
return rowData;
}
+
+ /**
+ * Writes a value at the given index only when the field contributes a
column, and reports the
+ * next free index. A field with an empty output name is skipped by {@code
+ * KafkaConsumerInputMeta.getRowMeta}, so writing it here would shift every
later column.
+ */
+ private int putIfNamed(Object[] rowData, int index, KafkaConsumerField
field, Object value) {
+ if (field == null || StringUtils.isEmpty(field.getOutputName()) || index
>= rowData.length) {
+ return index;
+ }
+ rowData[index] = value;
+ return index + 1;
+ }
}
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
index fb3f48a5d4..b662f6d2e8 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
@@ -501,7 +501,9 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
// don't let any rows get deleted or added (this does not affect the
read-only state of the
// cells)
- fieldsTable.setReadonly(true);
+ // The output name column is editable so fields can be renamed, and so the
Headers field - which
+ // ships unnamed to keep existing pipelines unchanged - can be switched on
from the dialog.
+ fieldsTable.setReadonly(false);
}
private void buildOptionsTable(Composite parentWidget) {
@@ -888,9 +890,13 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
String outputType = row.getText(3);
try {
KafkaConsumerField.Name ref =
KafkaConsumerField.Name.valueOf(kafkaName.toUpperCase());
- KafkaConsumerField field =
- new KafkaConsumerField(ref, outputName,
KafkaConsumerField.Type.valueOf(outputType));
- // meta.setField(field); TODO FIXME
+ KafkaConsumerField field = fieldFor(ref);
+ if (field != null) {
+ field.setOutputName(outputName);
+ if (StringUtils.isNotEmpty(outputType)) {
+ field.setOutputType(KafkaConsumerField.Type.valueOf(outputType));
+ }
+ }
} catch (IllegalArgumentException e) {
if (isDebug()) {
logDebug(e.getMessage(), e);
@@ -899,6 +905,22 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
}
}
+ /**
+ * The metadata keeps one typed instance per Kafka field rather than a list,
so an edited table
+ * row has to be written back onto the matching instance.
+ */
+ private KafkaConsumerField fieldFor(KafkaConsumerField.Name name) {
+ return switch (name) {
+ case KEY -> meta.getKeyField();
+ case MESSAGE -> meta.getMessageField();
+ case TOPIC -> meta.getTopicField();
+ case PARTITION -> meta.getPartitionField();
+ case OFFSET -> meta.getOffsetField();
+ case TIMESTAMP -> meta.getTimestampField();
+ case HEADERS -> meta.getHeadersField();
+ };
+ }
+
private void setTopicsFromTable() {
int itemCount = topicsTable.getItemCount();
ArrayList<String> tableTopics = new ArrayList<>();
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
index 303fc18671..bc84cbd7c2 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
@@ -105,6 +105,8 @@ public class KafkaConsumerInputMeta
@HopMetadataProperty private TimestampConsumerField timestampField;
+ @HopMetadataProperty private HeadersConsumerField headersField;
+
@HopMetadataProperty(
key = "pipelinePath",
injectionKey = "pipelinePath",
@@ -236,6 +238,10 @@ public class KafkaConsumerInputMeta
timestampField =
new TimestampConsumerField(
BaseMessages.getString(PKG,
"KafkaConsumerInputDialog.TimestampField"));
+
+ // Left unnamed on purpose: an empty output name keeps the field off the
output row, so
+ // pipelines saved before headers existed produce exactly the same row as
before.
+ headersField = new HeadersConsumerField("");
}
public KafkaConsumerInputMeta(KafkaConsumerInputMeta m) {
@@ -271,6 +277,7 @@ public class KafkaConsumerInputMeta
putFieldOnRowMeta(getPartitionField(), rowMeta, origin, variables);
putFieldOnRowMeta(getOffsetField(), rowMeta, origin, variables);
putFieldOnRowMeta(getTimestampField(), rowMeta, origin, variables);
+ putFieldOnRowMeta(getHeadersField(), rowMeta, origin, variables);
return rowMeta;
}
@@ -301,7 +308,8 @@ public class KafkaConsumerInputMeta
getTopicField(),
getPartitionField(),
getOffsetField(),
- getTimestampField()));
+ getTimestampField(),
+ getHeadersField()));
}
@Override
@@ -512,6 +520,7 @@ public class KafkaConsumerInputMeta
case PARTITION -> partitionField;
case OFFSET -> offsetField;
case TIMESTAMP -> timestampField;
+ case HEADERS -> headersField;
};
field.setKafkaName(name);
field.setOutputType(type);
@@ -606,6 +615,41 @@ public class KafkaConsumerInputMeta
}
}
+ /**
+ * The record headers, rendered as a JSON array of {@code
{"name":..,"value":..}} objects.
+ *
+ * <p>A Kafka record carries an ordered list of header pairs and the same
name may appear more
+ * than once, which a flat row column cannot represent. An array of objects
keeps both the order
+ * and any repeats, so the value round-trips through the Kafka Producer
transform unchanged. Leave
+ * the output name empty to keep headers off the row entirely.
+ */
+ @Getter
+ @Setter
+ public static class HeadersConsumerField extends KafkaConsumerField {
+ @HopMetadataProperty(
+ key = "outputName",
+ injectionKey = "HEADERS.OUTPUT_NAME",
+ injectionKeyDescription =
"KafkaConsumerInputMeta.Injection.HEADERS.OUTPUT_NAME")
+ protected String outputName;
+
+ public HeadersConsumerField() {
+ super();
+ this.outputName = "";
+ super.outputType = Type.String;
+ super.kafkaName = Name.HEADERS;
+ }
+
+ public HeadersConsumerField(HeadersConsumerField f) {
+ super(f.kafkaName, f.outputName, f.outputType);
+ this.outputName = f.outputName;
+ }
+
+ public HeadersConsumerField(String outputName) {
+ this();
+ this.outputName = outputName;
+ }
+ }
+
@Getter
@Setter
public static class OffsetConsumerField extends KafkaConsumerField {
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
index b76e2a9301..dfb7870456 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
@@ -26,7 +26,9 @@ import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMeta;
import org.apache.hop.pipeline.transforms.kafka.consumer.KafkaConsumerField;
import org.apache.hop.pipeline.transforms.kafka.shared.KafkaFactory;
+import org.apache.hop.pipeline.transforms.kafka.shared.KafkaHeaders;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
public class KafkaProducerOutput
extends BaseTransform<KafkaProducerOutputMeta, KafkaProducerOutputData> {
@@ -84,6 +86,17 @@ public class KafkaProducerOutput
data.topic = resolve(meta.getTopic());
}
+ data.headersFieldIndex = -1;
+ String headersField = resolve(meta.getHeadersField());
+ if (StringUtils.isNotEmpty(headersField)) {
+ data.headersFieldIndex = getInputRowMeta().indexOfValue(headersField);
+ if (data.headersFieldIndex < 0) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "KafkaProducerOutput.Error.HeadersFieldNotFound",
headersField));
+ }
+ }
+
data.kafkaProducer =
kafkaFactory.producer(
meta,
@@ -100,13 +113,18 @@ public class KafkaProducerOutput
return false;
}
String topic = resolveTopic(r);
+ Iterable<Header> headers = resolveHeaders(r);
ProducerRecord<Object, Object> producerRecord;
// allow for null keys
if (data.keyFieldIndex < 0
|| getInputRowMeta().isNull(r, data.keyFieldIndex)
|| StringUtils.isEmpty(r[data.keyFieldIndex].toString())) {
- producerRecord = new ProducerRecord<>(topic, r[data.messageFieldIndex]);
+ // Explicit casts pick the (topic, partition, timestamp, key, value,
headers) constructor;
+ // bare nulls leave it ambiguous against the shorter overloads.
+ producerRecord =
+ new ProducerRecord<>(
+ topic, (Integer) null, (Long) null, null,
r[data.messageFieldIndex], headers);
} else {
Object nativeObject =
@@ -116,7 +134,12 @@ public class KafkaProducerOutput
producerRecord =
new ProducerRecord<>(
- topic, getInputRowMeta().getString(r, data.keyFieldIndex),
nativeObject);
+ topic,
+ (Integer) null,
+ (Long) null,
+ getInputRowMeta().getString(r, data.keyFieldIndex),
+ nativeObject,
+ headers);
}
data.kafkaProducer.send(producerRecord);
@@ -157,6 +180,27 @@ public class KafkaProducerOutput
return topic;
}
+ /**
+ * Reads the record headers for a row from the configured field, which holds
the same JSON array
+ * of {@code {"name":..,"value":..}} objects the Kafka Consumer produces, so
headers survive a
+ * consume-then-produce round trip.
+ *
+ * @param r the current input row
+ * @return the headers to attach, or null when no headers field is
configured or the value is
+ * empty
+ * @throws HopException if the field does not hold a usable headers document
+ */
+ private Iterable<Header> resolveHeaders(Object[] r) throws HopException {
+ if (data.headersFieldIndex < 0) {
+ return null;
+ }
+ String json = getInputRowMeta().getString(r, data.headersFieldIndex);
+ if (StringUtils.isBlank(json)) {
+ return null;
+ }
+ return KafkaHeaders.fromJson(json);
+ }
+
@Override
public void stopRunning() {
if (data.kafkaProducer != null && data.isOpen) {
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
index 20bc68c198..56af40aa12 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
@@ -30,6 +30,7 @@ public class KafkaProducerOutputData extends
BaseTransformData implements ITrans
public int keyFieldIndex;
public int messageFieldIndex;
public int topicFieldIndex;
+ public int headersFieldIndex;
public String topic;
public boolean isOpen;
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
index c00a064b2f..dafc099a95 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
@@ -77,6 +77,7 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
private ComboVar wTopicField;
private ComboVar wKeyField;
private ComboVar wMessageField;
+ private ComboVar wHeadersField;
private TableView optionsTable;
private CTabFolder wTabFolder;
@@ -300,6 +301,31 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
variables, pipelineMeta, wMessageField, transformName);
wMessageField.getCComboWidget().addListener(SWT.FocusIn, lsMessageFocus);
+ Label wlHeadersField = new Label(wSetupComp, SWT.RIGHT);
+ PropsUi.setLook(wlHeadersField);
+ wlHeadersField.setText(BaseMessages.getString(PKG,
"KafkaProducerOutputDialog.HeadersField"));
+ FormData fdlHeadersField = new FormData();
+ fdlHeadersField.left = new FormAttachment(0, 0);
+ fdlHeadersField.top = new FormAttachment(wMessageField, margin);
+ fdlHeadersField.right = new FormAttachment(middle, -margin);
+ wlHeadersField.setLayoutData(fdlHeadersField);
+
+ wHeadersField = new ComboVar(variables, wSetupComp, SWT.SINGLE | SWT.LEFT
| SWT.BORDER);
+ PropsUi.setLook(wHeadersField);
+ wHeadersField.addModifyListener(lsMod);
+ FormData fdHeadersField = new FormData();
+ fdHeadersField.left = new FormAttachment(wlHeadersField, margin);
+ fdHeadersField.top = new FormAttachment(wlHeadersField, 0, SWT.CENTER);
+ fdHeadersField.right = new FormAttachment(100, 0);
+ wHeadersField.setLayoutData(fdHeadersField);
+ wHeadersField
+ .getCComboWidget()
+ .addListener(
+ SWT.FocusIn,
+ e ->
+ KafkaDialogHelper.populateFieldsList(
+ variables, pipelineMeta, wHeadersField, transformName));
+
FormData fdSetupComp = new FormData();
fdSetupComp.left = new FormAttachment(0, 0);
fdSetupComp.top = new FormAttachment(0, 0);
@@ -418,6 +444,7 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
wTopicField.setText(Const.NVL(meta.getTopicField(), ""));
wKeyField.setText(Const.NVL(meta.getKeyField(), ""));
wMessageField.setText(Const.NVL(meta.getMessageField(), ""));
+ wHeadersField.setText(Const.NVL(meta.getHeadersField(), ""));
populateOptionsData();
enableFields();
@@ -540,6 +567,21 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
}
meta.setMessageField(wMessageField.getText());
+
+ if (!Utils.isEmpty(wHeadersField.getText()) &&
checkMissingField(wHeadersField.getText())) {
+ MessageBox mb = new MessageBox(shell, SWT.OK | SWT.ICON_ERROR);
+ mb.setMessage(
+ BaseMessages.getString(
+ PKG,
+ "KafkaProducerOutputDialog.HeadersFieldNotExists.Message",
+ wHeadersField.getText()));
+ mb.setText(
+ BaseMessages.getString(PKG,
CONST_KAFKA_PRODUCER_OUTPUT_DIALOG_FIELD_NOT_EXISTS_TITLE));
+ mb.open();
+ return;
+ }
+
+ meta.setHeadersField(wHeadersField.getText());
meta.setOptions(KafkaDialogHelper.getConfig(optionsTable));
transformName = wTransformName.getText();
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
index 625329f7af..bbf12bb110 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
@@ -87,6 +87,12 @@ public class KafkaProducerOutputMeta
injectionKeyDescription = "KafkaProducerOutputMeta.Injection.KEY_FIELD")
private String keyField;
+ @HopMetadataProperty(
+ key = "headersField",
+ injectionKey = "HEADERS_FIELD",
+ injectionKeyDescription =
"KafkaProducerOutputMeta.Injection.HEADERS_FIELD")
+ private String headersField;
+
@HopMetadataProperty(
key = "messageField",
injectionKey = "MESSAGE_FIELD",
@@ -112,6 +118,7 @@ public class KafkaProducerOutputMeta
this.topic = m.topic;
this.topicInField = m.topicInField;
this.topicField = m.topicField;
+ this.headersField = m.headersField;
this.keyField = m.keyField;
this.messageField = m.messageField;
m.options.forEach(option -> this.options.add(new KafkaOption(option)));
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/shared/KafkaHeaders.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/shared/KafkaHeaders.java
new file mode 100644
index 0000000000..63fd920188
--- /dev/null
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/shared/KafkaHeaders.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.kafka.shared;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+/**
+ * Converts Kafka record headers to and from the textual form carried in a
pipeline field.
+ *
+ * <p>A record carries an ordered list of name/value pairs and the same name
may appear more than
+ * once, which a flat row column cannot represent. The exchange format is
therefore a JSON array of
+ * {@code {"name":..,"value":..}} objects rather than a JSON object: an object
would silently drop
+ * repeats and lose ordering. Keeping both sides of the conversion here means
the Kafka Consumer and
+ * Kafka Producer cannot drift apart, so headers survive a
consume-then-produce round trip.
+ *
+ * <p>Header values are bytes on the wire and are treated as UTF-8 text. A
null value is preserved
+ * as JSON null, which is distinct from an empty value.
+ */
+public final class KafkaHeaders {
+
+ private static final Class<?> PKG = KafkaHeaders.class;
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private static final String NAME = "name";
+ private static final String VALUE = "value";
+
+ private KafkaHeaders() {
+ // Utility class
+ }
+
+ /**
+ * Renders record headers as a JSON array, preserving order and repeated
names.
+ *
+ * @param headers the headers to render, may be null
+ * @return a JSON array document, empty when there are no headers
+ */
+ public static String toJson(Headers headers) {
+ ArrayNode array = JsonNodeFactory.instance.arrayNode();
+ if (headers != null) {
+ for (Header header : headers) {
+ ObjectNode node = array.addObject();
+ node.put(NAME, header.key());
+ if (header.value() == null) {
+ node.putNull(VALUE);
+ } else {
+ node.put(VALUE, new String(header.value(), StandardCharsets.UTF_8));
+ }
+ }
+ }
+ return array.toString();
+ }
+
+ /**
+ * Parses the document produced by {@link #toJson(Headers)} back into record
headers.
+ *
+ * @param json the headers document
+ * @return the parsed headers, in document order
+ * @throws HopException if the document is not a JSON array of name/value
objects
+ */
+ public static List<Header> fromJson(String json) throws HopException {
+ JsonNode root;
+ try {
+ root = OBJECT_MAPPER.readTree(json);
+ } catch (JsonProcessingException e) {
+ throw new HopException(BaseMessages.getString(PKG,
"KafkaHeaders.Error.NotJson", json), e);
+ }
+ if (root == null || !root.isArray()) {
+ throw new HopException(BaseMessages.getString(PKG,
"KafkaHeaders.Error.NotArray", json));
+ }
+
+ List<Header> headers = new ArrayList<>();
+ for (JsonNode element : root) {
+ JsonNode name = element.get(NAME);
+ if (name == null || !name.isTextual()) {
+ throw new HopException(BaseMessages.getString(PKG,
"KafkaHeaders.Error.MissingName", json));
+ }
+ JsonNode value = element.get(VALUE);
+ byte[] bytes =
+ (value == null || value.isNull())
+ ? null
+ : value.asText().getBytes(StandardCharsets.UTF_8);
+ headers.add(new RecordHeader(name.asText(), bytes));
+ }
+ return headers;
+ }
+}
diff --git
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
index b642302d33..5703e114e7 100644
---
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
+++
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
@@ -80,6 +80,7 @@ KafkaConsumerInputMeta.Injection.TOPICS_LIST=The topics
KafkaConsumerInputMeta.Injection.TRANSFORMATION_PATH=The path to the
sub-pipeline.
KafkaConsumerInputMeta.Injection.VALUES=The value(s) of the Kafka consumer
configuration properties.
KafkaConsumerInputMeta.Injection.TOPIC.OUTPUT_NAME=Topic output name
+KafkaConsumerInputMeta.Injection.HEADERS.OUTPUT_NAME=Headers output name;
leave empty to keep the record headers off the output row
KafkaConsumerInputMeta.Injection.OFFSET.OUTPUT_NAME=Offset output name
KafkaConsumerInputMeta.Injection.PARTITION.OUTPUT_NAME=Partition output name
KafkaConsumerInputMeta.Injection.TIMESTAMP.OUTPUT_NAME=Timestamp output name
diff --git
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
index 7b62ef7716..7f41648629 100644
---
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
+++
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
@@ -19,6 +19,7 @@ KafkaConsumerOutput.Log.LineNumber=Linenr {0}
KafkaProducer.TypeLongDesc=Kafka producer
KafkaProducer.TypeTooltipDesc=Sends messages to a Kafka producer for a certain
topic
KafkaProducerOutput.Error.EmptyTopicInField=Field ''{0}'' contains no topic
name for the row at line {1}
+KafkaProducerOutput.Error.HeadersFieldNotFound=Field ''{0}'' containing the
record headers is not present in the input stream
KafkaProducerOutput.Error.NoTopicFieldSpecified=No field containing the topic
name has been specified
KafkaProducerOutput.Error.TopicFieldNotFound=Field ''{0}'' containing the
topic name is not present in the input stream
KafkaProducerOutputDialog.BootstrapServerMandatory.Message=''Bootstrap
servers'' field''s value is mandatory!
@@ -27,6 +28,8 @@ KafkaProducerOutputDialog.ClientId=Client ID
KafkaProducerOutputDialog.FailedToGetFields.DialogMessage=Unable to get fields
from previous transforms because of an error
KafkaProducerOutputDialog.FailedToGetFields.DialogTitle=Field not present
KafkaProducerOutputDialog.FieldNotExists.Title=Error
+KafkaProducerOutputDialog.HeadersField=Headers field
+KafkaProducerOutputDialog.HeadersFieldNotExists.Message=Field ''{0}''
containing the record headers is not present in input stream
KafkaProducerOutputDialog.KeyField=Key field
KafkaProducerOutputDialog.KeyFieldNotExists.Message=Field ''{0}'' containing
key value is not present in input stream
KafkaProducerOutputDialog.MessageField=Message field
@@ -48,6 +51,7 @@ KafkaProducerOutputMeta.Injection.CLUSTER_NAME=The cluster
containing the Kafka
KafkaProducerOutputMeta.Injection.CONFIGURATION_PROPERTIES=The Kafka producer
configuration properties.
KafkaProducerOutputMeta.Injection.CONNECTION_TYPE=Specify the connection type:
DIRECT or CLUSTER.
KafkaProducerOutputMeta.Injection.DIRECT_BOOTSTRAP_SERVERS=Specify the
Bootstrap Servers when Connection Type is DIRECT.
+KafkaProducerOutputMeta.Injection.HEADERS_FIELD=The name of the stream field
holding the record headers, as a JSON array of name/value objects.
KafkaProducerOutputMeta.Injection.KEY_FIELD=The name of the stream field used
as the Kafka key.
KafkaProducerOutputMeta.Injection.MESSAGE_FIELD=The name of the stream field
used as the Kafka message.
KafkaProducerOutputMeta.Injection.NAMES=The name(s) of the Kafka producer
configuration properties.
diff --git
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/shared/messages/messages_en_US.properties
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/shared/messages/messages_en_US.properties
new file mode 100644
index 0000000000..45b1264d7e
--- /dev/null
+++
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/shared/messages/messages_en_US.properties
@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+KafkaHeaders.Error.MissingName=A record header is missing its ''name'': {0}
+KafkaHeaders.Error.NotArray=The record headers must be a JSON array of
name/value objects: {0}
+KafkaHeaders.Error.NotJson=The record headers could not be read as JSON: {0}
diff --git
a/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/shared/KafkaHeadersTest.java
b/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/shared/KafkaHeadersTest.java
new file mode 100644
index 0000000000..4d3358cae4
--- /dev/null
+++
b/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/shared/KafkaHeadersTest.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.kafka.shared;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import org.apache.hop.core.exception.HopException;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.junit.jupiter.api.Test;
+
+class KafkaHeadersTest {
+
+ private static Headers headers(String... namesAndValues) {
+ RecordHeaders recordHeaders = new RecordHeaders();
+ for (int i = 0; i < namesAndValues.length; i += 2) {
+ String value = namesAndValues[i + 1];
+ recordHeaders.add(
+ namesAndValues[i], value == null ? null :
value.getBytes(StandardCharsets.UTF_8));
+ }
+ return recordHeaders;
+ }
+
+ private static String valueOf(Header header) {
+ return header.value() == null ? null : new String(header.value(),
StandardCharsets.UTF_8);
+ }
+
+ // --- rendering
-------------------------------------------------------------------------------
+
+ @Test
+ void testNoHeadersRenderEmptyArray() {
+ assertEquals("[]", KafkaHeaders.toJson(new RecordHeaders()));
+ }
+
+ @Test
+ void testNullHeadersRenderEmptyArray() {
+ assertEquals("[]", KafkaHeaders.toJson(null));
+ }
+
+ @Test
+ void testSingleHeaderIsRendered() {
+ assertEquals(
+ "[{\"name\":\"traceparent\",\"value\":\"00-abc\"}]",
+ KafkaHeaders.toJson(headers("traceparent", "00-abc")));
+ }
+
+ @Test
+ void testRenderingPreservesOrder() {
+ assertEquals(
+ "[{\"name\":\"b\",\"value\":\"1\"},{\"name\":\"a\",\"value\":\"2\"}]",
+ KafkaHeaders.toJson(headers("b", "1", "a", "2")));
+ }
+
+ @Test
+ void testRenderingKeepsRepeatedNames() {
+ // The reason the format is an array rather than an object: Kafka allows
the same header name
+ // more than once and an object would silently drop one.
+ assertEquals(
+
"[{\"name\":\"tag\",\"value\":\"one\"},{\"name\":\"tag\",\"value\":\"two\"}]",
+ KafkaHeaders.toJson(headers("tag", "one", "tag", "two")));
+ }
+
+ @Test
+ void testNullHeaderValueRendersJsonNull() {
+ assertEquals(
+ "[{\"name\":\"tombstone\",\"value\":null}]",
+ KafkaHeaders.toJson(headers("tombstone", null)));
+ }
+
+ @Test
+ void testRenderingEscapesJson() {
+ assertEquals(
+ "[{\"name\":\"quoted\",\"value\":\"he said \\\"hi\\\"\"}]",
+ KafkaHeaders.toJson(headers("quoted", "he said \"hi\"")));
+ }
+
+ // --- parsing
---------------------------------------------------------------------------------
+
+ @Test
+ void testEmptyArrayParsesToNoHeaders() throws HopException {
+ assertTrue(KafkaHeaders.fromJson("[]").isEmpty());
+ }
+
+ @Test
+ void testParsingKeepsOrderAndRepeatedNames() throws HopException {
+ List<Header> parsed =
+ KafkaHeaders.fromJson(
+
"[{\"name\":\"tag\",\"value\":\"one\"},{\"name\":\"a\",\"value\":\"x\"},"
+ + "{\"name\":\"tag\",\"value\":\"two\"}]");
+
+ assertEquals(3, parsed.size());
+ assertEquals("tag", parsed.get(0).key());
+ assertEquals("one", valueOf(parsed.get(0)));
+ assertEquals("a", parsed.get(1).key());
+ assertEquals("tag", parsed.get(2).key());
+ assertEquals("two", valueOf(parsed.get(2)));
+ }
+
+ @Test
+ void testJsonNullParsesToNullValue() throws HopException {
+ // A null header value is distinct from an empty one on the wire.
+
assertNull(KafkaHeaders.fromJson("[{\"name\":\"tombstone\",\"value\":null}]").get(0).value());
+ }
+
+ @Test
+ void testMissingValueParsesToNullValue() throws HopException {
+ assertNull(KafkaHeaders.fromJson("[{\"name\":\"bare\"}]").get(0).value());
+ }
+
+ @Test
+ void testNonJsonIsRejected() {
+ assertThrows(HopException.class, () -> KafkaHeaders.fromJson("not json at
all"));
+ }
+
+ @Test
+ void testJsonObjectIsRejected() {
+ // An object cannot express ordering or repeats, so it is rejected even
though it is the more
+ // obvious shape to hand-write.
+ assertThrows(HopException.class, () ->
KafkaHeaders.fromJson("{\"traceparent\":\"00-abc\"}"));
+ }
+
+ @Test
+ void testEntryWithoutNameIsRejected() {
+ assertThrows(HopException.class, () ->
KafkaHeaders.fromJson("[{\"value\":\"orphan\"}]"));
+ }
+
+ // --- round trip
------------------------------------------------------------------------------
+
+ @Test
+ void testRoundTripPreservesEverything() throws HopException {
+ // The point of keeping both directions together: what the Kafka Consumer
writes into a field
+ // can be handed straight back to the Kafka Producer and yield the same
headers.
+ Headers original =
+ headers("tag", "one", "tag", "two", "city", "München", "tombstone",
null, "empty", "");
+
+ List<Header> parsed = KafkaHeaders.fromJson(KafkaHeaders.toJson(original));
+
+ assertEquals(5, parsed.size());
+ assertEquals("tag", parsed.get(0).key());
+ assertEquals("one", valueOf(parsed.get(0)));
+ assertEquals("two", valueOf(parsed.get(1)));
+ assertEquals("München", valueOf(parsed.get(2)));
+ assertNull(parsed.get(3).value(), "a null value must not become an empty
one");
+ assertEquals("", valueOf(parsed.get(4)), "an empty value must not become
null");
+ }
+}
