[
https://issues.apache.org/jira/browse/NIFI-16517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133070#comment-18133070
]
Paweł Rzepiński edited comment on NIFI-16517 at 10/27/26, 4:57 PM:
-------------------------------------------------------------------
In {{kafka-clients}} 4.3.1 (commit
[26b251a|https://github.com/apache/kafka/commit/26b251a451ce941d3d7a55e6487bcb7f16b5ad48],
the commit for tag {{4.3.1}}), a null value and an empty value are different
records. Only the null value is a tombstone.
The current record format writes them with different lengths. A null value is
length {{-1}} and has no payload. An empty buffer is length {{0}} followed by
zero bytes.
[DefaultRecord.writeTo|https://github.com/apache/kafka/blob/26b251a451ce941d3d7a55e6487bcb7f16b5ad48/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java#L191-L197]:
{code:java}
if (value == null) {
ByteUtils.writeVarint(-1, out);
} else {
int valueSize = value.remaining();
ByteUtils.writeVarint(valueSize, out);
Utils.writeTo(out, value, valueSize);
}
{code}
On read, a negative length comes back as null. A length of {{0}} comes back as
an empty buffer.
[Utils.readBytes|https://github.com/apache/kafka/blob/26b251a451ce941d3d7a55e6487bcb7f16b5ad48/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L750-L758]:
{code:java}
public static ByteBuffer readBytes(ByteBuffer srcBuf, int bytesToRead) {
if (bytesToRead < 0)
return null;
final ByteBuffer dstBuf = srcBuf.slice();
dstBuf.limit(bytesToRead);
srcBuf.position(srcBuf.position() + bytesToRead);
return dstBuf;
}
{code}
{{Record.hasValue()}} is true whenever the value is not null, so an empty value
still counts as present. {{DefaultRecord}} checks {{value != null}}.
{{PartialDefaultRecord}} checks {{valueSize >= 0}}. The older message formats
do the same: a null value is stored as the int {{-1}}, and {{hasNullValue()}}
is {{valueSize() < 0}}.
The client marks a record as a tombstone only when that check fails.
[MemoryRecords.filterBatch|https://github.com/apache/kafka/blob/26b251a451ce941d3d7a55e6487bcb7f16b5ad48/clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecords.java#L230-L238]:
{code:java}
if (filter.shouldRetainRecord(batch, record)) {
if (record.offset() > maxOffset)
maxOffset = record.offset();
retainedRecords.add(record);
if (!record.hasValue()) {
containsTombstones = true;
}
{code}
A producer can send both. {{ProducerRecord}} allows a null value. After a round
trip, the consumer sees {{value() == null}} for the tombstone and a non-null
zero-length buffer for the empty message.
was (Author: JIRAUSER315004):
The current record format writes tombstone and empty buffer with different
lengths. A null value is length -1 and has no payload. An empty buffer is
length 0 followed by zero bytes.
DefaultRecord.writeTo
(https://github.com/apache/kafka/blob/26b251a451ce941d3d7a55e6487bcb7f16b5ad48/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java#L191-L197):
if (value == null) {
ByteUtils.writeVarint(-1, out);
} else {
int valueSize = value.remaining();
ByteUtils.writeVarint(valueSize, out);
Utils.writeTo(out, value, valueSize);
}
On read, a negative length comes back as null. A length of 0 comes back as an
empty buffer.
Utils.readBytes
(https://github.com/apache/kafka/blob/26b251a451ce941d3d7a55e6487bcb7f16b5ad48/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L750-L758):
public static ByteBuffer readBytes(ByteBuffer srcBuf, int bytesToRead) {
if (bytesToRead < 0)
return null;
final ByteBuffer dstBuf = srcBuf.slice();
dstBuf.limit(bytesToRead);
srcBuf.position(srcBuf.position() + bytesToRead);
return dstBuf;
}
Record.hasValue() is true whenever the value is not null, so an empty value
still counts as present. DefaultRecord checks value != null.
PartialDefaultRecord checks valueSize >= 0. The older message formats do the
same: a null value is stored as the int -1, and hasNullValue() is valueSize() <
0.
The client marks a record as a tombstone only when that check fails.
MemoryRecords.filterBatch
(https://github.com/apache/kafka/blob/26b251a451ce941d3d7a55e6487bcb7f16b5ad48/clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecords.java#L230-L238):
if (filter.shouldRetainRecord(batch, record)) {
if (record.offset() > maxOffset)
maxOffset = record.offset();
retainedRecords.add(record);
if (!record.hasValue()) {
containsTombstones = true;
}
So a producer can send both. ProducerRecord allows a null value. After a round
trip, the consumer sees value() == null for the tombstone and a non-null
zero-length buffer for the empty message.
> ConsumeKafka sets tombstone attribute to true for empty payload in FLOW_FILE
> Processing Strategy
> ------------------------------------------------------------------------------------------------
>
> Key: NIFI-16517
> URL: https://issues.apache.org/jira/browse/NIFI-16517
> Project: Apache NiFi
> Issue Type: Bug
> Reporter: Paweł Rzepiński
> Priority: Minor
>
> Both null and empty (zero bytes array) payload results in empty flow file
> with tombstone attribute set to "true" in FLOW_FILE Processing Strategy.
> Only null payload should do that.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
