rzepinskip commented on code in PR #11857:
URL: https://github.com/apache/nifi/pull/11857#discussion_r4413056279
##########
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:
What do you mean? There is no further changes needed here: we detect whether
Kafka record is tombstone then set ByteRecord's tombstone parameter
accordingly. The value we pass to ByteRecord has to be non-null so for
tombstone I am using empty byte array as previous code did.
For FLOW_FILE strategy the additional change needed is in
FlowFileStreamKafkaMessageConverter.java:
https://github.com/apache/nifi/pull/11860/changes#diff-db771f11851dc797c52ebb887ddf5691c3e68052efb12cfb4373aff5a0a61e2fL67
--
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]
