This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 4776001c55e NIFI-16026 - Add Hex header encoding option to
ConsumeKafka (#11356)
4776001c55e is described below
commit 4776001c55e1657c0e81935711cff9f05f636eb1
Author: Alaksiej Ščarbaty <[email protected]>
AuthorDate: Mon Jun 22 11:13:37 2026 +0200
NIFI-16026 - Add Hex header encoding option to ConsumeKafka (#11356)
Add a 'Header Format' property (String/Hex) to ConsumeKafka so binary
Kafka Record Header values can be written as a lowercase hexadecimal
string instead of being corrupted by charset decoding. The existing
'Header Encoding' property keeps its meaning (the character set used for
the String format) and is shown only when Header Format is String, so
existing flows are unchanged and no property migration is needed.
Header decoding is resolved once into a HeaderValueConverter in
onScheduled and applied uniformly to FlowFile attributes and the
wrapper/inject-metadata record header fields. Internal batch headers
(kafka.max.offset, kafka.count) decode as fixed UTF-8, matching how they
are written.
Also add unit tests for KafkaUtils.toKeyString and correct the KeyEncoding
HEX description, which claimed uppercase output although HexFormat.of()
emits lowercase.
---
.../nifi/kafka/processors/ConsumeKafkaIT.java | 32 ++++++++++++++++
.../ConsumeKafkaInjectMetadataRecordIT.java | 35 +++++++++++++++++
.../processors/ConsumeKafkaWrapperRecordIT.java | 34 +++++++++++++++++
.../apache/nifi/kafka/processors/ConsumeKafka.java | 44 +++++++++++++++++-----
.../processors/common/HeaderValueConverter.java | 25 ++++++++++++
.../nifi/kafka/processors/common/KafkaUtils.java | 13 +++----
.../consumer/bundle/ByteRecordBundler.java | 10 ++---
.../AbstractRecordStreamKafkaMessageConverter.java | 10 ++---
.../FlowFileStreamKafkaMessageConverter.java | 10 ++---
...ectOffsetRecordStreamKafkaMessageConverter.java | 6 +--
.../convert/RecordStreamKafkaMessageConverter.java | 8 ++--
.../WrapperRecordStreamKafkaMessageConverter.java | 10 ++---
.../consumer/wrapper/ConsumeWrapperRecord.java | 10 ++---
.../producer/wrapper/InjectMetadataRecord.java | 5 ++-
.../kafka/processors/common/KafkaUtilsTest.java | 36 +++++++++++++++---
.../RecordStreamKafkaMessageConverterTest.java | 2 +-
...apperRecordStreamKafkaMessageConverterTest.java | 2 +-
.../{KeyEncoding.java => HeaderFormat.java} | 11 +++---
.../nifi/kafka/shared/property/KeyEncoding.java | 2 +-
19 files changed, 241 insertions(+), 64 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
index e4e12898e38..76fe12d06ad 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaIT.java
@@ -24,6 +24,7 @@ import
org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
import org.apache.nifi.kafka.service.Kafka3ConnectionService;
import org.apache.nifi.kafka.service.api.consumer.AutoOffsetReset;
import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
+import org.apache.nifi.kafka.shared.property.HeaderFormat;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
@@ -164,6 +165,37 @@ class ConsumeKafkaIT extends AbstractConsumeKafkaIT {
flowFile.assertAttributeNotExists("custom"); // Non-core attributes
should not exist without prefix
}
+ @Test
+ void testProcessingStrategyFlowFileHexHeaderEncoding() throws
InterruptedException, ExecutionException {
+ final String topic = UUID.randomUUID().toString();
+ final String groupId = topic.substring(0, topic.indexOf("-"));
+
+ runner.setProperty(ConsumeKafka.GROUP_ID, groupId);
+ runner.setProperty(ConsumeKafka.TOPICS, topic);
+ runner.setProperty(ConsumeKafka.PROCESSING_STRATEGY,
ProcessingStrategy.FLOW_FILE.getValue());
+ runner.setProperty(ConsumeKafka.HEADER_NAME_PATTERN, ".*");
+ runner.setProperty(ConsumeKafka.HEADER_NAME_PREFIX, "kafka.header.");
+ runner.setProperty(ConsumeKafka.HEADER_FORMAT,
HeaderFormat.HEX.getValue());
+
+ runner.run(1, false, true);
+
+ final byte[] binaryValue = new byte[] {0x01, 0x02, (byte) 0xab, (byte)
0xff};
+ final List<Header> headers = List.of(new RecordHeader("binary",
binaryValue));
+ produceOne(topic, 0, null, RECORD_VALUE, headers);
+ while (runner.getFlowFilesForRelationship("success").isEmpty()) {
+ runner.run(1, false, false);
+ }
+
+ runner.run(1, true, false);
+
+ final Iterator<MockFlowFile> flowFiles =
runner.getFlowFilesForRelationship(ConsumeKafka.SUCCESS).iterator();
+ assertTrue(flowFiles.hasNext());
+
+ final MockFlowFile flowFile = flowFiles.next();
+ flowFile.assertContentEquals(RECORD_VALUE);
+ flowFile.assertAttributeEquals("kafka.header.binary", "0102abff");
+ }
+
/**
* Test ability to specify a topic regular expression to query for
messages.
*/
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaInjectMetadataRecordIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaInjectMetadataRecordIT.java
index 84a4c8b2753..1c4524c44e8 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaInjectMetadataRecordIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaInjectMetadataRecordIT.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
import org.apache.nifi.kafka.processors.producer.wrapper.InjectMetadataRecord;
import org.apache.nifi.kafka.service.api.consumer.AutoOffsetReset;
+import org.apache.nifi.kafka.shared.property.HeaderFormat;
import org.apache.nifi.kafka.shared.property.KeyFormat;
import org.apache.nifi.kafka.shared.property.OutputStrategy;
import org.apache.nifi.provenance.ProvenanceEventRecord;
@@ -36,6 +37,7 @@ import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -186,4 +188,37 @@ class ConsumeKafkaInjectMetadataRecordIT extends
AbstractConsumeKafkaIT {
final ProvenanceEventRecord provenanceEvent =
provenanceEvents.getFirst();
assertEquals(ProvenanceEventType.RECEIVE,
provenanceEvent.getEventType());
}
+
+ @Test
+ void testInjectMetadataRecordHexHeaderEncoding()
+ throws InterruptedException, ExecutionException, IOException {
+ final String topic = UUID.randomUUID().toString();
+ final String groupId = topic.substring(0, topic.indexOf("-"));
+
+ runner.setProperty(ConsumeKafka.GROUP_ID, groupId);
+ runner.setProperty(ConsumeKafka.TOPICS, topic);
+ runner.setProperty(ConsumeKafka.PROCESSING_STRATEGY,
ProcessingStrategy.RECORD);
+ runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET,
AutoOffsetReset.EARLIEST);
+ runner.setProperty(ConsumeKafka.OUTPUT_STRATEGY,
OutputStrategy.INJECT_METADATA);
+ runner.setProperty(ConsumeKafka.KEY_FORMAT, KeyFormat.STRING);
+ runner.setProperty(ConsumeKafka.HEADER_FORMAT, HeaderFormat.HEX);
+
+ runner.run(1, false, true);
+ final String message = new
String(IOUtils.toByteArray(Objects.requireNonNull(
+ getClass().getClassLoader().getResource(TEST_RESOURCE))),
StandardCharsets.UTF_8);
+ final byte[] binaryValue = new byte[] {0x01, 0x02, (byte) 0xab, (byte)
0xff};
+ produceOne(topic, 0, MESSAGE_KEY, message, Collections.singletonList(
+ new RecordHeader("binary", binaryValue)));
+ while (runner.getFlowFilesForRelationship("success").isEmpty()) {
+ runner.run(1, false, false);
+ }
+ runner.run(1, true, false);
+
+ final MockFlowFile flowFile =
runner.getFlowFilesForRelationship(ConsumeKafka.SUCCESS).getFirst();
+ final ArrayNode arrayNode = assertInstanceOf(ArrayNode.class,
objectMapper.readTree(flowFile.getContent()));
+ final ObjectNode record = assertInstanceOf(ObjectNode.class,
arrayNode.get(0));
+ final ObjectNode metadata = assertInstanceOf(ObjectNode.class,
record.get(InjectMetadataRecord.METADATA));
+ final ObjectNode headers = assertInstanceOf(ObjectNode.class,
metadata.get(InjectMetadataRecord.HEADERS));
+ assertEquals("0102abff", headers.get("binary").asText());
+ }
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaWrapperRecordIT.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaWrapperRecordIT.java
index 7ce0469bbca..46ee56e3488 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaWrapperRecordIT.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-3-integration/src/test/java/org/apache/nifi/kafka/processors/ConsumeKafkaWrapperRecordIT.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
import org.apache.nifi.kafka.processors.producer.wrapper.InjectMetadataRecord;
import org.apache.nifi.kafka.service.api.consumer.AutoOffsetReset;
+import org.apache.nifi.kafka.shared.property.HeaderFormat;
import org.apache.nifi.kafka.shared.property.KeyFormat;
import org.apache.nifi.kafka.shared.property.OutputStrategy;
import org.apache.nifi.provenance.ProvenanceEventRecord;
@@ -36,6 +37,7 @@ import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -186,4 +188,36 @@ class ConsumeKafkaWrapperRecordIT extends
AbstractConsumeKafkaIT {
final ProvenanceEventRecord provenanceEvent =
provenanceEvents.getFirst();
assertEquals(ProvenanceEventType.RECEIVE,
provenanceEvent.getEventType());
}
+
+ @Test
+ void testWrapperRecordHexHeaderEncoding()
+ throws InterruptedException, ExecutionException, IOException {
+ final String topic = UUID.randomUUID().toString();
+ final String groupId = topic.substring(0, topic.indexOf("-"));
+
+ runner.setProperty(ConsumeKafka.GROUP_ID, groupId);
+ runner.setProperty(ConsumeKafka.TOPICS, topic);
+ runner.setProperty(ConsumeKafka.PROCESSING_STRATEGY,
ProcessingStrategy.RECORD);
+ runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET,
AutoOffsetReset.EARLIEST);
+ runner.setProperty(ConsumeKafka.OUTPUT_STRATEGY,
OutputStrategy.USE_WRAPPER);
+ runner.setProperty(ConsumeKafka.KEY_FORMAT, KeyFormat.STRING);
+ runner.setProperty(ConsumeKafka.HEADER_FORMAT, HeaderFormat.HEX);
+
+ runner.run(1, false, true);
+ final String message = new
String(IOUtils.toByteArray(Objects.requireNonNull(
+ getClass().getClassLoader().getResource(TEST_RESOURCE))),
StandardCharsets.UTF_8);
+ final byte[] binaryValue = new byte[] {0x01, 0x02, (byte) 0xab, (byte)
0xff};
+ produceOne(topic, 0, MESSAGE_KEY, message, Collections.singletonList(
+ new RecordHeader("binary", binaryValue)));
+ while (runner.getFlowFilesForRelationship("success").isEmpty()) {
+ runner.run(1, false, false);
+ }
+ runner.run(1, true, false);
+
+ final MockFlowFile flowFile =
runner.getFlowFilesForRelationship(ConsumeKafka.SUCCESS).getFirst();
+ final ArrayNode arrayNode = assertInstanceOf(ArrayNode.class,
objectMapper.readTree(flowFile.getContent()));
+ final ObjectNode wrapper = assertInstanceOf(ObjectNode.class,
arrayNode.get(0));
+ final ObjectNode headers = assertInstanceOf(ObjectNode.class,
wrapper.get("headers"));
+ assertEquals("0102abff", headers.get("binary").asText());
+ }
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
index 9a50eb6060c..b489edbc662 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/ConsumeKafka.java
@@ -32,6 +32,7 @@ import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.Validator;
import org.apache.nifi.components.connector.components.ConnectorMethod;
import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.common.KafkaUtils;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.processors.consumer.ProcessingStrategy;
@@ -50,6 +51,7 @@ import
org.apache.nifi.kafka.service.api.consumer.RebalanceCallback;
import org.apache.nifi.kafka.service.api.consumer.SessionContext;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
+import org.apache.nifi.kafka.shared.property.HeaderFormat;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
import org.apache.nifi.kafka.shared.property.KeyFormat;
import org.apache.nifi.kafka.shared.property.OutputStrategy;
@@ -75,6 +77,7 @@ import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HexFormat;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -94,7 +97,9 @@ import static
org.apache.nifi.expression.ExpressionLanguageScope.NONE;
+ "configured Record Reader or Record Writer, the contents of the
message will be written to a separate FlowFile, and that FlowFile will be
transferred to the "
+ "'parse.failure' relationship. Otherwise, each FlowFile is sent to
the 'success' relationship and may contain many individual messages within the
single FlowFile. "
+ "A 'record.count' attribute is added to indicate how many messages
are contained in the FlowFile. No two Kafka messages will be placed into the
same FlowFile if they "
- + "have different schemas, or if they have different values for a
message header that is included by the <Headers to Add as Attributes>
property.")
+ + "have different schemas, or if they have different values for a
message header that is included by the <Headers to Add as Attributes> property.
"
+ + "Kafka Record Header values selected for output are represented
according to the Header Format property: as text decoded with the configured
Header Encoding "
+ + "character set, or as a lowercase hexadecimal string for binary-safe
output.")
@Tags({"Kafka", "Get", "Record", "csv", "avro", "json", "Ingest", "Ingress",
"Topic", "PubSub", "Consume"})
@WritesAttributes({
@WritesAttribute(attribute = "record.count", description = "The number
of records received"),
@@ -194,12 +199,21 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
.addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
.build();
+ static final PropertyDescriptor HEADER_FORMAT = new
PropertyDescriptor.Builder()
+ .name("Header Format")
+ .description("Specifies how Kafka Record Header values are
represented when written as FlowFile attributes or record fields")
+ .required(true)
+ .defaultValue(HeaderFormat.STRING)
+ .allowableValues(HeaderFormat.class)
+ .build();
+
static final PropertyDescriptor HEADER_ENCODING = new
PropertyDescriptor.Builder()
.name("Header Encoding")
.description("Character encoding applied when reading Kafka Record
Header values and writing FlowFile attributes")
.addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
.defaultValue(StandardCharsets.UTF_8.name())
.required(true)
+ .dependsOn(HEADER_FORMAT, HeaderFormat.STRING)
.build();
static final PropertyDescriptor HEADER_NAME_PATTERN = new
PropertyDescriptor.Builder()
@@ -322,6 +336,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
MAX_UNCOMMITTED_SIZE,
MAX_UNCOMMITTED_TIME,
HEADER_NAME_PATTERN,
+ HEADER_FORMAT,
HEADER_ENCODING,
PROCESSING_STRATEGY,
HEADER_NAME_PREFIX,
@@ -338,7 +353,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
private static final Set<Relationship> SUCCESS_RELATIONSHIP =
Set.of(SUCCESS);
private static final Set<Relationship> SUCCESS_FAILURE_RELATIONSHIPS =
Set.of(SUCCESS, PARSE_FAILURE);
- private volatile Charset headerEncoding;
+ private volatile HeaderValueConverter headerValueConverter;
private volatile Pattern headerNamePattern;
private volatile String headerNamePrefix;
private volatile ProcessingStrategy processingStrategy;
@@ -373,10 +388,21 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
}
}
+ private static HeaderValueConverter createHeaderValueConverter(final
ProcessContext context) {
+ final HeaderFormat headerFormat =
context.getProperty(HEADER_FORMAT).asAllowableValue(HeaderFormat.class);
+ return switch (headerFormat) {
+ case STRING -> {
+ final Charset charset =
Charset.forName(context.getProperty(HEADER_ENCODING).getValue());
+ yield value -> new String(value, charset);
+ }
+ case HEX -> value -> HexFormat.of().formatHex(value);
+ };
+ }
+
@OnScheduled
public void onScheduled(final ProcessContext context) {
pollingContext = createPollingContext(context);
- headerEncoding =
Charset.forName(context.getProperty(HEADER_ENCODING).getValue());
+ headerValueConverter = createHeaderValueConverter(context);
final String headerNamePatternProperty =
context.getProperty(HEADER_NAME_PATTERN).getValue();
if (StringUtils.isNotBlank(headerNamePatternProperty)) {
@@ -634,7 +660,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
for (final ByteRecord byteRecord : records) {
recordIndex++;
final Map<String, String> recordAttributes =
KafkaUtils.toAttributes(
- byteRecord, keyEncoding, headerNamePattern, headerEncoding,
commitOffsets);
+ byteRecord, keyEncoding, headerNamePattern,
headerValueConverter, commitOffsets);
try (final InputStream inputStream = new
ByteArrayInputStream(byteRecord.getValue());
final RecordReader reader =
readerFactory.createRecordReader(recordAttributes, inputStream,
byteRecord.getValue().length, verificationLogger)) {
@@ -765,7 +791,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
final byte[] demarcator =
demarcatorValue.getBytes(StandardCharsets.UTF_8);
final boolean separateByKey =
context.getProperty(SEPARATE_BY_KEY).asBoolean();
- return new ByteRecordBundler(demarcator, separateByKey, keyEncoding,
headerNamePattern, headerEncoding, commitOffsets).bundle(consumerRecords);
+ return new ByteRecordBundler(demarcator, separateByKey, keyEncoding,
headerNamePattern, headerValueConverter, commitOffsets).bundle(consumerRecords);
}
private void processInputRecords(final ProcessContext context, final
ProcessSession session, final OffsetTracker offsetTracker, final
Iterator<ByteRecord> consumerRecords) {
@@ -774,13 +800,13 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
final KafkaMessageConverter converter;
if (outputStrategy == OutputStrategy.USE_VALUE) {
- converter = new RecordStreamKafkaMessageConverter(readerFactory,
writerFactory, headerEncoding, headerNamePattern,
+ converter = new RecordStreamKafkaMessageConverter(readerFactory,
writerFactory, headerValueConverter, headerNamePattern,
keyEncoding, commitOffsets, offsetTracker, getLogger(),
brokerUri);
} else if (outputStrategy == OutputStrategy.INJECT_OFFSET) {
converter = new InjectOffsetRecordStreamKafkaMessageConverter(
readerFactory,
writerFactory,
- headerEncoding,
+ headerValueConverter,
headerNamePattern,
keyEncoding,
commitOffsets,
@@ -793,7 +819,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
?
context.getProperty(KEY_RECORD_READER).asControllerService(RecordReaderFactory.class)
: null;
converter = new
WrapperRecordStreamKafkaMessageConverter(readerFactory, writerFactory,
keyReaderFactory,
- headerEncoding, headerNamePattern, keyFormat, keyEncoding,
commitOffsets, offsetTracker, getLogger(), brokerUri, outputStrategy);
+ headerValueConverter, headerNamePattern, keyFormat,
keyEncoding, commitOffsets, offsetTracker, getLogger(), brokerUri,
outputStrategy);
}
converter.toFlowFiles(session, consumerRecords);
@@ -801,7 +827,7 @@ public class ConsumeKafka extends AbstractProcessor
implements VerifiableProcess
private void processInputFlowFile(final ProcessSession session, final
OffsetTracker offsetTracker, final Iterator<ByteRecord> consumerRecords) {
final KafkaMessageConverter converter = new
FlowFileStreamKafkaMessageConverter(
- headerEncoding, headerNamePattern, headerNamePrefix, keyEncoding,
commitOffsets, offsetTracker, brokerUri);
+ headerValueConverter, headerNamePattern, headerNamePrefix,
keyEncoding, commitOffsets, offsetTracker, brokerUri);
converter.toFlowFiles(session, consumerRecords);
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/HeaderValueConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/HeaderValueConverter.java
new file mode 100644
index 00000000000..eeca62a4aa2
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/HeaderValueConverter.java
@@ -0,0 +1,25 @@
+/*
+ * 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.nifi.kafka.processors.common;
+
+/**
+ * Converts a Kafka Record Header value to the String form written as a
FlowFile attribute or record field.
+ */
+@FunctionalInterface
+public interface HeaderValueConverter {
+ String convert(byte[] value);
+}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/KafkaUtils.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/KafkaUtils.java
index 40e1e373678..853640f4110 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/KafkaUtils.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/common/KafkaUtils.java
@@ -21,7 +21,6 @@ import org.apache.nifi.kafka.service.api.record.ByteRecord;
import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
-import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
@@ -72,14 +71,14 @@ public class KafkaUtils {
}
public static Map<String, String> toAttributes(final ByteRecord
consumerRecord, final KeyEncoding keyEncoding,
- final Pattern
headerNamePattern, final Charset headerEncoding,
+ final Pattern
headerNamePattern, final HeaderValueConverter headerValueConverter,
final boolean
commitOffsets) {
- return toAttributes(consumerRecord, keyEncoding, headerNamePattern,
null, headerEncoding, commitOffsets);
+ return toAttributes(consumerRecord, keyEncoding, headerNamePattern,
null, headerValueConverter, commitOffsets);
}
public static Map<String, String> toAttributes(final ByteRecord
consumerRecord, final KeyEncoding keyEncoding,
final Pattern
headerNamePattern, final String headerNamePrefix,
- final Charset
headerEncoding, final boolean commitOffsets) {
+ final HeaderValueConverter
headerValueConverter, final boolean commitOffsets) {
final Map<String, String> attributes = new HashMap<>();
attributes.put(KafkaFlowFileAttribute.KAFKA_TOPIC,
consumerRecord.getTopic());
attributes.put(KafkaFlowFileAttribute.KAFKA_PARTITION,
Long.toString(consumerRecord.getPartition()));
@@ -87,10 +86,10 @@ public class KafkaUtils {
attributes.put(KafkaFlowFileAttribute.KAFKA_CONSUMER_OFFSETS_COMMITTED,
String.valueOf(commitOffsets));
consumerRecord.getHeaders().stream()
.filter(h ->
h.key().equals(KafkaFlowFileAttribute.KAFKA_MAX_OFFSET)).findFirst()
- .ifPresent(h ->
attributes.put(KafkaFlowFileAttribute.KAFKA_MAX_OFFSET, new String(h.value(),
headerEncoding)));
+ .ifPresent(h ->
attributes.put(KafkaFlowFileAttribute.KAFKA_MAX_OFFSET, new String(h.value(),
StandardCharsets.UTF_8)));
consumerRecord.getHeaders().stream()
.filter(h ->
h.key().equals(KafkaFlowFileAttribute.KAFKA_COUNT)).findFirst()
- .ifPresent(h ->
attributes.put(KafkaFlowFileAttribute.KAFKA_COUNT, new String(h.value(),
headerEncoding)));
+ .ifPresent(h ->
attributes.put(KafkaFlowFileAttribute.KAFKA_COUNT, new String(h.value(),
StandardCharsets.UTF_8)));
attributes.put(KafkaFlowFileAttribute.KAFKA_TIMESTAMP,
Long.toString(consumerRecord.getTimestamp()));
Optional.ofNullable(toKeyString(consumerRecord.getKey().orElse(null),
keyEncoding))
.ifPresent(keyAttribute ->
attributes.put(KafkaFlowFileAttribute.KAFKA_KEY, keyAttribute));
@@ -102,7 +101,7 @@ public class KafkaUtils {
for (final RecordHeader header : headers) {
final String name = header.key();
if (headerNamePattern.matcher(name).matches()) {
- final String value = new String(header.value(),
headerEncoding);
+ final String value =
headerValueConverter.convert(header.value());
final String attributeName = headerNamePrefix != null ?
headerNamePrefix + name : name;
attributes.put(attributeName, value);
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
index a0144aa48ef..c47e308cfe6 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.kafka.processors.consumer.bundle;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.common.KafkaUtils;
import org.apache.nifi.kafka.service.api.common.TopicPartitionSummary;
import org.apache.nifi.kafka.service.api.header.RecordHeader;
@@ -23,7 +24,6 @@ import org.apache.nifi.kafka.service.api.record.ByteRecord;
import org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
-import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Iterator;
@@ -36,18 +36,18 @@ public class ByteRecordBundler {
private final boolean separateByKey;
private final KeyEncoding keyEncoding;
private final Pattern headerNamePattern;
- private final Charset headerEncoding;
+ private final HeaderValueConverter headerValueConverter;
private final boolean commitOffsets;
private final Map<BundleKey, BundleValue> bundles;
public ByteRecordBundler(final byte[] demarcator, final boolean
separateByKey, final KeyEncoding keyEncoding,
- final Pattern headerNamePattern, final Charset
headerEncoding, final boolean commitOffsets) {
+ final Pattern headerNamePattern, final
HeaderValueConverter headerValueConverter, final boolean commitOffsets) {
this.demarcator = demarcator;
this.separateByKey = separateByKey;
this.keyEncoding = keyEncoding;
this.headerNamePattern = headerNamePattern;
- this.headerEncoding = headerEncoding;
+ this.headerValueConverter = headerValueConverter;
this.commitOffsets = commitOffsets;
this.bundles = new HashMap<>();
}
@@ -74,7 +74,7 @@ public class ByteRecordBundler {
final List<RecordHeader> headers = byteRecord.getHeaders();
final List<RecordHeader> headersFiltered =
KafkaUtils.toHeadersFiltered(byteRecord, headerNamePattern);
final byte[] messageKey = (separateByKey ?
byteRecord.getKey().orElse(null) : null);
- final Map<String, String> attributes =
KafkaUtils.toAttributes(byteRecord, keyEncoding, headerNamePattern,
headerEncoding, commitOffsets);
+ final Map<String, String> attributes =
KafkaUtils.toAttributes(byteRecord, keyEncoding, headerNamePattern,
headerValueConverter, commitOffsets);
final BundleKey bundleKey = new BundleKey(topicPartition,
byteRecord.getTimestamp(), headers, headersFiltered, attributes, messageKey);
if (bundles.containsKey(bundleKey)) {
update(bundles, byteRecord, bundleKey);
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
index 5163bdf7353..c1f8baff634 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/AbstractRecordStreamKafkaMessageConverter.java
@@ -19,6 +19,7 @@ package org.apache.nifi.kafka.processors.consumer.convert;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.kafka.processors.ConsumeKafka;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.common.KafkaUtils;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
@@ -42,7 +43,6 @@ import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -55,7 +55,7 @@ public abstract class
AbstractRecordStreamKafkaMessageConverter implements Kafka
protected final RecordReaderFactory readerFactory;
protected final RecordSetWriterFactory writerFactory;
- protected final Charset headerEncoding;
+ protected final HeaderValueConverter headerValueConverter;
protected final Pattern headerNamePattern;
protected final KeyEncoding keyEncoding;
protected final boolean commitOffsets;
@@ -66,7 +66,7 @@ public abstract class
AbstractRecordStreamKafkaMessageConverter implements Kafka
public AbstractRecordStreamKafkaMessageConverter(
final RecordReaderFactory readerFactory,
final RecordSetWriterFactory writerFactory,
- final Charset headerEncoding,
+ final HeaderValueConverter headerValueConverter,
final Pattern headerNamePattern,
final KeyEncoding keyEncoding,
final boolean commitOffsets,
@@ -75,7 +75,7 @@ public abstract class
AbstractRecordStreamKafkaMessageConverter implements Kafka
final String brokerUri) {
this.readerFactory = readerFactory;
this.writerFactory = writerFactory;
- this.headerEncoding = headerEncoding;
+ this.headerValueConverter = headerValueConverter;
this.headerNamePattern = headerNamePattern;
this.keyEncoding = keyEncoding;
this.commitOffsets = commitOffsets;
@@ -96,7 +96,7 @@ public abstract class
AbstractRecordStreamKafkaMessageConverter implements Kafka
// shared attribute extraction
final Map<String, String> attributes = KafkaUtils.toAttributes(
- consumerRecord, keyEncoding, headerNamePattern,
headerEncoding, commitOffsets);
+ consumerRecord, keyEncoding, headerNamePattern,
headerValueConverter, commitOffsets);
// hook for subclasses to expose headers (if desired)
final Map<String, String> extraAttrs =
extractHeaders(consumerRecord);
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/FlowFileStreamKafkaMessageConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/FlowFileStreamKafkaMessageConverter.java
index 815ddcb32be..f0289950bb0 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/FlowFileStreamKafkaMessageConverter.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/FlowFileStreamKafkaMessageConverter.java
@@ -18,6 +18,7 @@ package org.apache.nifi.kafka.processors.consumer.convert;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.kafka.processors.ConsumeKafka;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.common.KafkaUtils;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
@@ -25,13 +26,12 @@ import
org.apache.nifi.kafka.shared.attribute.KafkaFlowFileAttribute;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
import org.apache.nifi.processor.ProcessSession;
-import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Map;
import java.util.regex.Pattern;
public class FlowFileStreamKafkaMessageConverter implements
KafkaMessageConverter {
- private final Charset headerEncoding;
+ private final HeaderValueConverter headerValueConverter;
private final Pattern headerNamePattern;
private final String headerNamePrefix;
private final KeyEncoding keyEncoding;
@@ -40,14 +40,14 @@ public class FlowFileStreamKafkaMessageConverter implements
KafkaMessageConverte
private final String brokerUri;
public FlowFileStreamKafkaMessageConverter(
- final Charset headerEncoding,
+ final HeaderValueConverter headerValueConverter,
final Pattern headerNamePattern,
final String headerNamePrefix,
final KeyEncoding keyEncoding,
final boolean commitOffsets,
final OffsetTracker offsetTracker,
final String brokerUri) {
- this.headerEncoding = headerEncoding;
+ this.headerValueConverter = headerValueConverter;
this.headerNamePattern = headerNamePattern;
this.headerNamePrefix = headerNamePrefix;
this.keyEncoding = keyEncoding;
@@ -71,7 +71,7 @@ public class FlowFileStreamKafkaMessageConverter implements
KafkaMessageConverte
}
final Map<String, String> attributes = KafkaUtils.toAttributes(
- consumerRecord, keyEncoding, headerNamePattern,
headerNamePrefix, headerEncoding, commitOffsets);
+ consumerRecord, keyEncoding, headerNamePattern,
headerNamePrefix, headerValueConverter, commitOffsets);
flowFile = session.putAllAttributes(flowFile, attributes);
session.getProvenanceReporter().receive(flowFile, brokerUri + "/"
+ consumerRecord.getTopic());
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/InjectOffsetRecordStreamKafkaMessageConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/InjectOffsetRecordStreamKafkaMessageConverter.java
index 49b4f86b82a..a71d9c791c6 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/InjectOffsetRecordStreamKafkaMessageConverter.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/InjectOffsetRecordStreamKafkaMessageConverter.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.kafka.processors.consumer.convert;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
import org.apache.nifi.kafka.shared.property.KeyEncoding;
@@ -29,7 +30,6 @@ import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
-import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -48,7 +48,7 @@ public class InjectOffsetRecordStreamKafkaMessageConverter
extends AbstractRecor
public InjectOffsetRecordStreamKafkaMessageConverter(
final RecordReaderFactory readerFactory,
final RecordSetWriterFactory writerFactory,
- final Charset headerEncoding,
+ final HeaderValueConverter headerValueConverter,
final Pattern headerNamePattern,
final KeyEncoding keyEncoding,
final boolean commitOffsets,
@@ -59,7 +59,7 @@ public class InjectOffsetRecordStreamKafkaMessageConverter
extends AbstractRecor
super(
readerFactory,
writerFactory,
- headerEncoding,
+ headerValueConverter,
headerNamePattern,
keyEncoding,
commitOffsets,
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverter.java
index 4bf1fe882c6..5309db99dc9 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverter.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverter.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.kafka.processors.consumer.convert;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.service.api.header.RecordHeader;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
@@ -28,7 +29,6 @@ import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordSchema;
import java.io.IOException;
-import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;
@@ -38,14 +38,14 @@ public class RecordStreamKafkaMessageConverter extends
AbstractRecordStreamKafka
public RecordStreamKafkaMessageConverter(
final RecordReaderFactory readerFactory,
final RecordSetWriterFactory writerFactory,
- final Charset headerEncoding,
+ final HeaderValueConverter headerValueConverter,
final Pattern headerNamePattern,
final KeyEncoding keyEncoding,
final boolean commitOffsets,
final OffsetTracker offsetTracker,
final ComponentLog logger,
final String brokerUri) {
- super(readerFactory, writerFactory, headerEncoding, headerNamePattern,
keyEncoding, commitOffsets, offsetTracker, logger, brokerUri);
+ super(readerFactory, writerFactory, headerValueConverter,
headerNamePattern, keyEncoding, commitOffsets, offsetTracker, logger,
brokerUri);
}
@Override
@@ -72,7 +72,7 @@ public class RecordStreamKafkaMessageConverter extends
AbstractRecordStreamKafka
for (final RecordHeader h : consumerRecord.getHeaders()) {
final String name = h.key();
if (headerNamePattern.matcher(name).matches()) {
- headers.put(name, new String(h.value(), headerEncoding));
+ headers.put(name, headerValueConverter.convert(h.value()));
}
}
return headers;
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverter.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverter.java
index 0ed9fc84938..a4a3b4dd99a 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverter.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverter.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.kafka.processors.consumer.convert;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.consumer.OffsetTracker;
import org.apache.nifi.kafka.processors.consumer.wrapper.ConsumeWrapperRecord;
import
org.apache.nifi.kafka.processors.consumer.wrapper.WrapperRecordKeyReader;
@@ -36,7 +37,6 @@ import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.util.Tuple;
import java.io.IOException;
-import java.nio.charset.Charset;
import java.util.Map;
public class WrapperRecordStreamKafkaMessageConverter extends
AbstractRecordStreamKafkaMessageConverter {
@@ -49,7 +49,7 @@ public class WrapperRecordStreamKafkaMessageConverter extends
AbstractRecordStre
final RecordReaderFactory readerFactory,
final RecordSetWriterFactory writerFactory,
final RecordReaderFactory keyReaderFactory,
- final Charset headerEncoding,
+ final HeaderValueConverter headerValueConverter,
final java.util.regex.Pattern headerNamePattern,
final KeyFormat keyFormat,
final KeyEncoding keyEncoding,
@@ -58,7 +58,7 @@ public class WrapperRecordStreamKafkaMessageConverter extends
AbstractRecordStre
final ComponentLog logger,
final String brokerUri,
final OutputStrategy outputStrategy) {
- super(readerFactory, writerFactory, headerEncoding, headerNamePattern,
keyEncoding, commitOffsets, offsetTracker, logger, brokerUri);
+ super(readerFactory, writerFactory, headerValueConverter,
headerNamePattern, keyEncoding, commitOffsets, offsetTracker, logger,
brokerUri);
this.keyReaderFactory = keyReaderFactory;
this.keyFormat = keyFormat;
this.outputStrategy = outputStrategy;
@@ -84,8 +84,8 @@ public class WrapperRecordStreamKafkaMessageConverter extends
AbstractRecordStre
final WrapperRecordKeyReader keyReader = new
WrapperRecordKeyReader(keyFormat, keyReaderFactory, keyEncoding, logger);
final Tuple<RecordField, Object> recordKey =
keyReader.toWrapperRecordKey(consumerRecord.getKey().orElse(null), attributes);
return outputStrategy == OutputStrategy.USE_WRAPPER
- ? new
ConsumeWrapperRecord(headerEncoding).toWrapperRecord(consumerRecord, record,
recordKey)
- : InjectMetadataRecord.toWrapperRecord(headerEncoding,
consumerRecord, record, recordKey);
+ ? new
ConsumeWrapperRecord(headerValueConverter).toWrapperRecord(consumerRecord,
record, recordKey)
+ :
InjectMetadataRecord.toWrapperRecord(headerValueConverter, consumerRecord,
record, recordKey);
} catch (IOException | SchemaNotFoundException |
MalformedRecordException e) {
throw new IOException("Unable to convert record", e);
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/wrapper/ConsumeWrapperRecord.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/wrapper/ConsumeWrapperRecord.java
index edbc086bf33..0018859e57a 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/wrapper/ConsumeWrapperRecord.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/wrapper/ConsumeWrapperRecord.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.kafka.processors.consumer.wrapper;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.processors.producer.wrapper.WrapperRecord;
import org.apache.nifi.kafka.service.api.header.RecordHeader;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
@@ -27,7 +28,6 @@ import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.util.Tuple;
-import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -36,10 +36,10 @@ public class ConsumeWrapperRecord {
private static final RecordSchema EMPTY_SCHEMA = new
SimpleRecordSchema(List.of());
- private final Charset headerCharacterSet;
+ private final HeaderValueConverter headerValueConverter;
- public ConsumeWrapperRecord(final Charset headerCharacterSet) {
- this.headerCharacterSet = headerCharacterSet;
+ public ConsumeWrapperRecord(final HeaderValueConverter
headerValueConverter) {
+ this.headerValueConverter = headerValueConverter;
}
public MapRecord toWrapperRecord(final ByteRecord consumerRecord,
@@ -72,7 +72,7 @@ public class ConsumeWrapperRecord {
final RecordField recordField = new RecordField(WrapperRecord.HEADERS,
RecordFieldType.MAP.getMapDataType(RecordFieldType.STRING.getDataType()));
final Map<String, String> headers = new HashMap<>();
for (final RecordHeader header : consumerRecord.getHeaders()) {
- headers.put(header.key(), new String(header.value(),
headerCharacterSet));
+ headers.put(header.key(),
headerValueConverter.convert(header.value()));
}
return new Tuple<>(recordField, headers);
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/producer/wrapper/InjectMetadataRecord.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/producer/wrapper/InjectMetadataRecord.java
index b13c16348e2..a637cd60b8e 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/producer/wrapper/InjectMetadataRecord.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/producer/wrapper/InjectMetadataRecord.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.kafka.processors.producer.wrapper;
+import org.apache.nifi.kafka.processors.common.HeaderValueConverter;
import org.apache.nifi.kafka.service.api.header.RecordHeader;
import org.apache.nifi.kafka.service.api.record.ByteRecord;
import org.apache.nifi.serialization.SimpleRecordSchema;
@@ -108,7 +109,7 @@ public class InjectMetadataRecord extends MapRecord {
return new SimpleRecordSchema(valueFieldsWithInjectedMetadata);
}
- public static MapRecord toWrapperRecord(final Charset headerCharacterSet,
final ByteRecord consumerRecord, final Record record, final Tuple<RecordField,
Object> tupleKey) {
+ public static MapRecord toWrapperRecord(final HeaderValueConverter
headerValueConverter, final ByteRecord consumerRecord, final Record record,
final Tuple<RecordField, Object> tupleKey) {
final RecordSchema schema = record.getSchema();
RecordSchema finalSchema =
InjectMetadataRecord.toWrapperSchema(tupleKey.getKey(), schema);
@@ -123,7 +124,7 @@ public class InjectMetadataRecord extends MapRecord {
final Map<String, Object> valuesHeaders = new HashMap<>();
for (RecordHeader header : consumerRecord.getHeaders()) {
- valuesHeaders.put(header.key(), new String(header.value(),
headerCharacterSet));
+ valuesHeaders.put(header.key(),
headerValueConverter.convert(header.value()));
}
valuesMetadata.put(InjectMetadataRecord.HEADERS, valuesHeaders);
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
index 7aeedeb3823..e319fd44a4f 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/common/KafkaUtilsTest.java
@@ -29,6 +29,7 @@ import java.util.regex.Pattern;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
class KafkaUtilsTest {
@@ -37,6 +38,8 @@ class KafkaUtilsTest {
private static final long TEST_OFFSET = 100L;
private static final long TEST_TIMESTAMP = System.currentTimeMillis();
+ private static final HeaderValueConverter STRING_CONVERTER = value -> new
String(value, StandardCharsets.UTF_8);
+
@Test
void testToAttributesWithoutPrefix() {
final List<RecordHeader> headers = List.of(
@@ -48,7 +51,7 @@ class KafkaUtilsTest {
final Pattern headerPattern = Pattern.compile(".*");
final Map<String, String> attributes = KafkaUtils.toAttributes(
- record, KeyEncoding.UTF8, headerPattern,
StandardCharsets.UTF_8, true);
+ record, KeyEncoding.UTF8, headerPattern, STRING_CONVERTER,
true);
// Verify standard Kafka attributes
assertEquals(TEST_TOPIC,
attributes.get(KafkaFlowFileAttribute.KAFKA_TOPIC));
@@ -73,7 +76,7 @@ class KafkaUtilsTest {
final String prefix = "kafka.header.";
final Map<String, String> attributes = KafkaUtils.toAttributes(
- record, KeyEncoding.UTF8, headerPattern, prefix,
StandardCharsets.UTF_8, true);
+ record, KeyEncoding.UTF8, headerPattern, prefix,
STRING_CONVERTER, true);
// Verify standard Kafka attributes are NOT prefixed
assertEquals(TEST_TOPIC,
attributes.get(KafkaFlowFileAttribute.KAFKA_TOPIC));
@@ -105,7 +108,7 @@ class KafkaUtilsTest {
final String prefix = "msg.";
final Map<String, String> attributes = KafkaUtils.toAttributes(
- record, KeyEncoding.UTF8, headerPattern, prefix,
StandardCharsets.UTF_8, true);
+ record, KeyEncoding.UTF8, headerPattern, prefix,
STRING_CONVERTER, true);
// Verify matching headers are added with prefix
assertEquals("test-uuid", attributes.get("msg.uuid"));
@@ -130,7 +133,7 @@ class KafkaUtilsTest {
// Explicitly pass null for prefix
final Map<String, String> attributes = KafkaUtils.toAttributes(
- record, KeyEncoding.UTF8, headerPattern, null,
StandardCharsets.UTF_8, true);
+ record, KeyEncoding.UTF8, headerPattern, null,
STRING_CONVERTER, true);
// Verify header is added without prefix when prefix is null
assertEquals("value1", attributes.get("header1"));
@@ -148,12 +151,35 @@ class KafkaUtilsTest {
// Pass empty string for prefix
final Map<String, String> attributes = KafkaUtils.toAttributes(
- record, KeyEncoding.UTF8, headerPattern, "",
StandardCharsets.UTF_8, true);
+ record, KeyEncoding.UTF8, headerPattern, "", STRING_CONVERTER,
true);
// Verify header is added with empty prefix (effectively no prefix)
assertEquals("value1", attributes.get("header1"));
}
+ @Test
+ void testToKeyStringUtf8() {
+ final byte[] key = "my-key".getBytes(StandardCharsets.UTF_8);
+ assertEquals("my-key", KafkaUtils.toKeyString(key, KeyEncoding.UTF8));
+ }
+
+ @Test
+ void testToKeyStringHexIsLowercase() {
+ final byte[] key = new byte[] {0x01, 0x02, (byte) 0xab, (byte) 0xff};
+ assertEquals("0102abff", KafkaUtils.toKeyString(key, KeyEncoding.HEX));
+ }
+
+ @Test
+ void testToKeyStringDoNotAdd() {
+ final byte[] key = "my-key".getBytes(StandardCharsets.UTF_8);
+ assertNull(KafkaUtils.toKeyString(key, KeyEncoding.DO_NOT_ADD));
+ }
+
+ @Test
+ void testToKeyStringNullKey() {
+ assertNull(KafkaUtils.toKeyString(null, KeyEncoding.UTF8));
+ }
+
private ByteRecord createByteRecord(final List<RecordHeader> headers) {
return new ByteRecord(
TEST_TOPIC,
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
index e85ce4a85fd..2f694ed310a 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/RecordStreamKafkaMessageConverterTest.java
@@ -72,7 +72,7 @@ class RecordStreamKafkaMessageConverterTest {
final RecordStreamKafkaMessageConverter converter = new
RecordStreamKafkaMessageConverter(
readerFactory,
writerFactory,
- StandardCharsets.UTF_8,
+ value -> new String(value, StandardCharsets.UTF_8),
Pattern.compile(".*"),
KeyEncoding.UTF8,
true,
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
index 81295ba5144..ef135ceb077 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/kafka/processors/consumer/convert/WrapperRecordStreamKafkaMessageConverterTest.java
@@ -75,7 +75,7 @@ class WrapperRecordStreamKafkaMessageConverterTest {
readerFactory,
writerFactory,
keyReaderFactory,
- StandardCharsets.UTF_8,
+ value -> new String(value, StandardCharsets.UTF_8),
Pattern.compile(".*"),
KeyFormat.STRING,
KeyEncoding.UTF8,
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/HeaderFormat.java
similarity index 71%
copy from
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
copy to
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/HeaderFormat.java
index a42b5275f2c..810fdc08099 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/HeaderFormat.java
@@ -19,18 +19,17 @@ package org.apache.nifi.kafka.shared.property;
import org.apache.nifi.components.DescribedValue;
/**
- * Enumeration of supported Kafka Key Encoding Strategies
+ * Enumeration of supported Kafka Record Header Value Formats
*/
-public enum KeyEncoding implements DescribedValue {
- UTF8("utf-8", "UTF-8 Encoded", "The key is interpreted as a UTF-8 Encoded
string."),
- HEX("hex", "Hex Encoded", "The key is interpreted as arbitrary binary data
and is encoded using hexadecimal characters with uppercase letters"),
- DO_NOT_ADD("do-not-add", "Do Not Add Key as Attribute", "The key will not
be added as an Attribute");
+public enum HeaderFormat implements DescribedValue {
+ STRING("string", "String", "The header value is decoded as a string using
the configured Header Encoding character set."),
+ HEX("hex", "Hex Encoded", "The header value is interpreted as arbitrary
binary data and is encoded as a lowercase hexadecimal string");
private final String value;
private final String displayName;
private final String description;
- KeyEncoding(final String value, final String displayName, final String
description) {
+ HeaderFormat(final String value, final String displayName, final String
description) {
this.value = value;
this.displayName = displayName;
this.description = description;
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
index a42b5275f2c..68b79316593 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/property/KeyEncoding.java
@@ -23,7 +23,7 @@ import org.apache.nifi.components.DescribedValue;
*/
public enum KeyEncoding implements DescribedValue {
UTF8("utf-8", "UTF-8 Encoded", "The key is interpreted as a UTF-8 Encoded
string."),
- HEX("hex", "Hex Encoded", "The key is interpreted as arbitrary binary data
and is encoded using hexadecimal characters with uppercase letters"),
+ HEX("hex", "Hex Encoded", "The key is interpreted as arbitrary binary data
and is encoded as a lowercase hexadecimal string"),
DO_NOT_ADD("do-not-add", "Do Not Add Key as Attribute", "The key will not
be added as an Attribute");
private final String value;
