Paweł Rzepiński created NIFI-16518:
--------------------------------------
Summary: ConsumeKafka DEMARCATOR Processing Strategy commits the
wrong Kafka offset
Key: NIFI-16518
URL: https://issues.apache.org/jira/browse/NIFI-16518
Project: Apache NiFi
Issue Type: Bug
Reporter: Paweł Rzepiński
h2. Symptom
With {{Processing Strategy = DEMARCATOR}}, the offset committed to the Kafka
consumer group is one past the *first* offset in the bundle instead of one past
the *last* offset consumed.
A single poll of offsets 0 through 5, bundled into one FlowFile, commits offset
{{1}}. {{FLOW_FILE}} and {{RECORD}} commit {{6}} for the same six records.
The in-memory consumer position is unaffected, because {{KafkaConsumer.poll()}}
already advanced it past offset 5. Only the group offset stored on the broker
is short. That surfaces as:
* *Duplicates after restart.* The new consumer starts from the committed offset
and re-reads offsets 1 through 5.
* *Duplicates after rebalance.* {{commit()}} removes the partition from
{{uncommittedOffsets}} ({{Kafka3ConsumerService.java:161}}), so a later
rebalance has nothing to commit and the broker keeps the short offset. Moving
the partition to another node replays it.
* *Duplicates without any restart.* {{Kafka3ConsumerService.rollback}} seeks to
{{consumer.committed()}} ({{Kafka3ConsumerService.java:182-192}}). Any later
failure in the same running processor (an exception in {{onTrigger}}, a failed
session commit, a failed offset commit) rewinds to the short offset and
re-emits bundles already delivered.
* *Backlog never reports caught up.* {{getBacklog}} subtracts the committed
offset from the end offset via
{{Kafka3ConnectionService.getCommittedOffsetLag}}
({{Kafka3ConnectionService.java:379-396}}). With DEMARCATOR the reported lag
never reaches 0 when the final bundle holds more than one record, so
{{lastCaughtUp}} is never stamped.
h2. Root cause
Three pieces combine:
# {{ByteRecordBundler.toByteRecord}} ({{ByteRecordBundler.java:66-69}}) builds
one {{ByteRecord}} per bundle and passes {{BundleValue.getFirstOffset()}} as
that record's offset. The last offset is written only into a
{{kafka.max.offset}} header, which becomes a FlowFile attribute and is never
read by the commit path.
# {{OffsetTracker.update}} ({{OffsetTracker.java:39-41}}) stores
{{consumerRecord.getOffset()}} for the partition and *replaces* any previous
value.
# {{Kafka3ConsumerService.getOffsets}} ({{Kafka3ConsumerService.java:326}})
commits {{offsetSummary.getOffset() + 1}}.
For a bundle, step 1 feeds the first offset into step 2, and step 3 turns 0
into 1.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
