[
https://issues.apache.org/jira/browse/NIFI-16350?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132563#comment-18132563
]
Ankur Kotwal. commented on NIFI-16350:
--------------------------------------
I would like to work on this. Before opening a PR, I want to agree on the
intended RECORD-strategy behavior, since it is a user-visible change.
Current state: Kafka3ConsumerService (and RecordIterable) replace a null value
with new byte[0] before building the ByteRecord, so downstream code cannot tell
a tombstone from a non-null empty value. FlowFileStreamKafkaMessageConverter
infers a tombstone from value.length == 0, and
AbstractRecordStreamKafkaMessageConverter passes the empty value to the Record
Reader, which fails for readers that need a header (Protobuf with
ConfluentEncodedSchemaReferenceReader, Avro with a schema-registry encoded
header).
Proposal:
1. Carry the distinction explicitly: add a boolean "tombstone" to ByteRecord
(null value -> tombstone=true, value stays byte[0] so existing callers keep
working), set it in Kafka3ConsumerService and RecordIterable.
2. FLOW_FILE strategy: set kafka.tombstone=true only when the record is a real
tombstone, not for a non-null empty value.
3. RECORD strategy (and the "Parse Records" verification step): do not hand
tombstones to the Record Reader. Options:
a) route each tombstone to the success relationship as a separate empty
FlowFile with kafka.tombstone=true (consistent with FLOW_FILE), or
b) for the USE_WRAPPER output strategy, emit a wrapper record whose value is
null; for USE_VALUE, use (a).
4. Verification: count tombstones separately and report them instead of failing
"Parse Records".
My preference is (a) for USE_VALUE and (b) for USE_WRAPPER, since PublishKafka
already treats a null wrapper value as a tombstone, so a ConsumeKafka ->
PublishKafka round trip would preserve tombstones.
Does this direction work for the maintainers? I'm happy to adjust before I
start on the PR.
> ConsumeKafka RECORD strategy and verification treat Kafka tombstones as parse
> failures
> --------------------------------------------------------------------------------------
>
> Key: NIFI-16350
> URL: https://issues.apache.org/jira/browse/NIFI-16350
> Project: Apache NiFi
> Issue Type: Bug
> Reporter: Wojciech Trefon
> Priority: Major
>
> h2. Summary
> A Kafka tombstone is a record with a non-null key and a null value. When
> ConsumeKafka uses the RECORD processing strategy, tombstones are passed to
> the configured Record Reader as zero-length content.
> With StandardProtobufReader and ConfluentEncodedSchemaReferenceReader, this
> causes tombstones to be treated as malformed Protobuf records. The same
> record also causes ConsumeKafka verification to fail in the Parse Records
> step.
> h2. Steps to reproduce
> # Publish a Kafka record with a non-null key and a null value.
> # Configure ConsumeKafka with Processing Strategy set to RECORD.
> # Configure StandardProtobufReader with ConfluentEncodedSchemaReferenceReader.
> # Stop ConsumeKafka and run Verify.
> # Observe that the Parse Records verification step fails on the tombstone.
> # Start ConsumeKafka normally.
> # Observe that the tombstone is routed as a parse failure while the consumer
> offset advances.
> h2. Actual behavior
> Verify reports an error similar to:
> {code}
> Parse Records
> Failed to parse Record number 9:
> org.apache.nifi.schema.access.SchemaNotFoundException: Failed to read header
> in first 5 bytes from stream
> {code}
> During normal RECORD processing, the zero-length value reaches
> StandardProtobufReader and is routed to the parse failure relationship. If
> that relationship is auto-terminated, the tombstone is silently discarded
> while consumption continues and the offset advances.
> This differs from FLOW_FILE processing, where a tombstone can be represented
> by an empty FlowFile with the kafka.tombstone attribute set to true and
> routed through success.
> h2. Expected behavior
> A valid Kafka tombstone should not be classified as a malformed Protobuf
> record solely because it has no schema header. Verification and normal
> processing should handle the same tombstone consistently and preserve the
> distinction between a null Kafka value and a non-null empty value.
> h2. Code observation
> Kafka3ConsumerService converts a null ConsumerRecord value to a zero-length
> byte array before RECORD processing. At that point, downstream record parsing
> cannot distinguish a tombstone from a non-null empty value.
> h2. Related issues
> * [NIFI-12371|https://issues.apache.org/jira/browse/NIFI-12371] added
> tombstone support for non-record Kafka processors and noted that record-based
> processors might require separate handling.
> * [NIFI-14122|https://issues.apache.org/jira/browse/NIFI-14122] and
> [NIFI-14123|https://issues.apache.org/jira/browse/NIFI-14123] restored
> tombstone support in Kafka publish paths.
> * [NIFI-14247|https://issues.apache.org/jira/browse/NIFI-14247] covers
> ConsumeKafka tombstones in FLOW_FILE processing.
> The related issues do not cover ConsumeKafka RECORD processing or the Verify
> Parse Records step.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
