awelless commented on code in PR #11857:
URL: https://github.com/apache/nifi/pull/11857#discussion_r4415112674
##########
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 I had in mind is the following:
Each processing strategy should handle tombstones explicitly.
If `ByteRecord.isTombstone()` returns `true`, it'd be clearer if a caller
handles tombstone record explicitly instead of relying on the `value` holding
an empty byte array.
This is not a blocker, and this is not related to the current tombstone skip
logic during RECORD mode.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md:
##########
@@ -175,3 +175,14 @@ Here is an example of FlowFile content that is emitted by
JsonRecordSetWriter wh
}
]
```
+
+### Handling tombstones
+
+A Kafka tombstone is a record whose value is null (which is not the same as
empty byte array).
+
+Handling depends on Processing Strategy:
+
+- FLOW_FILE: The tombstone is represented as a FlowFile with `kafka.tombstone`
attribute set to `true`.
+- DEMARCATOR: The tombstone is included in the bundle records as empty byte
array.
+- RECORD: The tombstone is skipped for every Output Strategy. The processor
increments `Tombstones Skipped for <topic>` metric.
Review Comment:
Nit:
```suggestion
- RECORD: The tombstone is skipped for every Output Strategy. The processor
increments `Tombstones Skipped for <topic>` counter.
```
--
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]
