rzepinskip commented on code in PR #11857:
URL: https://github.com/apache/nifi/pull/11857#discussion_r4415461526
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/consumer/Kafka3ConsumerService.java:
##########
@@ -375,11 +375,10 @@ public ByteRecord next() {
recordHeaders.add(recordHeader);
});
- // Support Kafka tombstones
- byte[] value = consumerRecord.value();
- if (value == null) {
- value = new byte[0];
- }
+ // A null Kafka value is a tombstone. Keep that distinct from a
non-null empty value.
+ final byte[] originalValue = consumerRecord.value();
+ final boolean isTombstone = originalValue == null;
+ final byte[] value = isTombstone ? new byte[0] : originalValue;
Review Comment:
Right:
- FLOW_FILE needs a fix in
[FlowFileStreamKafkaMessageConverter](https://github.com/rzepinskip/nifi/blob/59b3217e5faeb2b1904b60fbacd0b374c551702f/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/convert/FlowFileStreamKafkaMessageConverter.java#L67-L71);
to be handled as part of https://issues.apache.org/jira/browse/NIFI-16517
- DEMARCATOR needs a fix in
[ByteRecordBundler](https://github.com/rzepinskip/nifi/blob/59b3217e5faeb2b1904b60fbacd0b374c551702f/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/consumer/bundle/ByteRecordBundler.java#L64-L70);
to be handled as part of https://issues.apache.org/jira/browse/NIFI-16521
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
