Juhwan Kim created NIFI-16450:
---------------------------------
Summary: PublishKafka fails with UnsupportedEncodingException when
Kafka Key Attribute Encoding is Hex Encoded
Key: NIFI-16450
URL: https://issues.apache.org/jira/browse/NIFI-16450
Project: Apache NiFi
Issue Type: Bug
Components: Extensions
Affects Versions: 2.12.0, 2.11.0
Reporter: Juhwan Kim
h3. What happens
While migrating our flows from NiFi 1.x to 2.11.0 I hit a wall with
PublishKafka: setting *Kafka Key Attribute Encoding* to *Hex Encoded* looks
fine in the UI and passes validation, but as soon as the processor runs, every
FlowFile with a key goes straight to {{failure}}:
{code}
java.io.UnsupportedEncodingException: hex
at java.base/java.lang.String.getBytes(String.java:1795)
at
org.apache.nifi.kafka.processors.producer.key.AttributeKeyFactory.getKey(AttributeKeyFactory.java:45)
at
org.apache.nifi.kafka.processors.producer.convert.FlowFileStreamKafkaRecordConverter.convert(FlowFileStreamKafkaRecordConverter.java:61)
at
org.apache.nifi.kafka.processors.PublishKafka$PublishCallback.process(PublishKafka.java:620)
{code}
Easy to reproduce: GenerateFlowFile -> UpdateAttribute ({{kafka.key}} =
{{0a0b0c}}) -> PublishKafka with Hex Encoded. *Do Not Add Key as Attribute*
fails the same way.
h3. Why
{{AttributeKeyFactory}} receives the raw property value ({{utf-8}}, {{hex}},
{{do-not-add}}) and hands it to {{String.getBytes(charsetName)}}. Only
{{utf-8}} happens to be a real charset name, so UTF-8 works by luck and the
other two can never work. There is no hex decoding anywhere in the publish
path, which means a binary key read by ConsumeKafka with Hex Encoded cannot be
published back as the same bytes. PublishKafka_2_6 in 1.x handled this with
{{DatatypeConverter.parseHexBinary}}; the new PublishKafka (NIFI-11259,
property restored in NIFI-14053) lost that part.
Checked on rel/nifi-2.11.0; the same code is on rel/nifi-2.12.0 and main.
h3. Fix I have in mind
Read the property as {{KeyEncoding}} and branch on it: HEX ->
{{HexFormat.of().parseHex(value)}} (upper and lower case, so keys from both 1.x
and 2.x ConsumeKafka round-trip), UTF8 -> {{getBytes(StandardCharsets.UTF_8)}}.
No new dependency. Two small questions I would like to settle in review:
whether *Do Not Add Key as Attribute* should simply be removed from
PublishKafka's allowable values, and fixing the property description, which is
currently copied from ConsumeKafka.
h3. I want to take this one
I have already got a working version of this fix running in our environment
with unit tests, so I would really like to own this ticket and submit the pull
request myself. Please assign it to me (or grant contributor access so I can).
I will have the PR up right after.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
