Paweł Rzepiński created NIFI-16566:
--------------------------------------
Summary: ConsumeKafka DEMARCATOR commits the wrong Kafka offset
Key: NIFI-16566
URL: https://issues.apache.org/jira/browse/NIFI-16566
Project: Apache NiFi
Issue Type: Bug
Affects Versions: 2.9.0
Reporter: Paweł Rzepiński
h2. Summary
With {{Processing Strategy = DEMARCATOR}}, {{ConsumeKafka}} commits a consumer
group offset that is one past the *first* offset in each bundle instead of one
past the *highest* offset consumed. The committed offset is therefore too low,
and any consumer that resumes from the group offset re-reads records that were
already delivered.
h2. Component
* Module: {{nifi-kafka-processors}} ({{ConsumeKafka}})
* Kafka client service: {{Kafka3ConsumerService}} / {{Kafka3ConnectionService}}
({{nifi-kafka-3-service}})
* Affected strategy: {{DEMARCATOR}} only
h2. Environment
* Apache NiFi, {{main}} branch at the time of investigation
* Kafka client: Kafka 3.x ({{Kafka3ConsumerService}})
h2. Steps to Reproduce
# Produce six records with offsets 0 through 5 to a topic with a single
partition.
# Configure {{ConsumeKafka}} with {{Processing Strategy = DEMARCATOR}} and
{{Commit Offsets = true}}.
# Let a single poll return offsets 0 through 5 and emit them as one FlowFile.
# Inspect the committed offset for the consumer group on the broker (for
example with {{kafka-consumer-groups.sh --describe}}).
h2. Expected Behavior
The committed offset for the partition is {{6}} (one past offset 5, the highest
offset consumed). This matches {{FLOW_FILE}} and {{RECORD}}, which commit {{6}}
for the same six records.
h2. Actual Behavior
The committed offset is {{1}} (one past offset 0, the first offset of the
bundle).
The in-memory consumer position is not affected, because
{{KafkaConsumer.poll()}} has already advanced past offset 5. Only the group
offset stored on the broker is short.
h2. Impact
* *Duplicates after restart.* A 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|Kafka3ConsumerService.java]). A later
rebalance has nothing to commit, so the broker keeps the short offset and the
partition is replayed on its new owner.
* *Duplicates without a 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 that were already delivered.
* *Backlog never reports caught up.* {{getBacklog}} computes lag from the
committed offset via {{Kafka3ConnectionService.getCommittedOffsetLag}}
({{Kafka3ConnectionService.java:379-396}}). With DEMARCATOR, the reported lag
does not reach 0 when the final bundle holds more than one record, so
{{lastCaughtUp}} is never set.
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 to 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.
Replacement is also a problem when several bundles are emitted for the same
partition. With {{Separate By Key}} or header-based grouping, bundles are
emitted in {{HashMap}} iteration order ({{ByteRecordBundler.java:52, 60-61}}).
With interleaved keys (A@0, B@1, A@2, B@3), replacing with "last offset per
bundle" commits 3 or 4 depending on that order. Only tracking the maximum
offset across all records gives the correct result.
{{RECORD}} is not affected by the bundler.
{{AbstractRecordStreamKafkaMessageConverter}} ({{:110, :116}}) calls {{update}}
per raw Kafka record on both the success and parse-failure branches.
{{MergeSchemaGrouping}} and {{CreateNewFlowFileGrouping}} compute only the
{{kafka.offset}} and {{kafka.max.offset}} attributes and never feed the commit.
h2. History
* [NIFI-11259|https://issues.apache.org/jira/browse/NIFI-11259], commit
{{24925ed9b5}} (author date 2023-03-28, committed 2024-06-24, PR
[#8463|https://github.com/apache/nifi/pull/8463]) introduced the bug. The
replaced {{ConsumeKafka_2_6}} committed {{maxOffset + 1}} computed from the raw
poll for every strategy, using the first offset only for the {{kafka.offset}}
attribute. The redesigned bundler collapsed these two values into one field, so
DEMARCATOR bundles committed {{0}}.
* [NIFI-14598|https://issues.apache.org/jira/browse/NIFI-14598], commit
{{eaca86ab6f}} (2025-05-23, PR
[#9971|https://github.com/apache/nifi/pull/9971]) added the {{+ 1}} to fix
{{FLOW_FILE}} replaying its last record after restart, and changed
{{ConsumeKafkaOffsetsIT}} from {{values.length - 1}} to {{values.length}}. The
DEMARCATOR commit became {{1}}.
* [NIFI-13787|https://issues.apache.org/jira/browse/NIFI-13787] and
[NIFI-13801|https://issues.apache.org/jira/browse/NIFI-13801] changed the
bundler for unrelated reasons (rollback handling, record counter) and did not
change the offset logic.
No Jira ticket describes the DEMARCATOR commit value, and no test asserts it.
Nothing in {{nifi-kafka-processors/src/test}} calls {{getPollingSummary}} or
captures {{commit}}.
h2. Existing Test Coverage
* {{ConsumeKafkaDemarcatorIT}} asserts FlowFile attributes only:
{{kafka.offset}} is the bundle's first offset and {{kafka.max.offset}} is the
last. It never reads the committed group offset.
* {{ConsumeKafkaOffsetsIT.testVerifyOffsetsCommitted}} is the only test that
reads the committed offset. It uses the default {{FLOW_FILE}} strategy and
expects {{values.length}}. Its wait loop counts one FlowFile per record, so it
cannot be reused as-is for a demarcated batch.
* {{ConsumeKafkaIT.testConsumesAllRecordsWithoutDuplicates}} covers the restart
case, explicitly on {{FLOW_FILE}}.
* {{ConsumeKafkaRebalanceIT}} drives {{Kafka3ConsumerService}} directly. In
production, {{ConsumeKafka}} always registers a rebalance callback
({{ConsumeKafka.java:866}}), so the revoked-partition commit runs inside
{{onPartitionsRevoked}} ({{Kafka3ConsumerService.java:117-139}}). That path
uses {{uncommittedOffsets}}, which tracks {{record.offset() + 1}} as a maximum
({{Kafka3ConsumerService.java:370}}), so the rebalance path is already correct.
h2. Suggested Fix
Track the highest consumed offset per partition at the point where the
processor receives each raw polled record, before any bundling or grouping. In
{{ConsumeKafka.TopicPartitionScanningIterator.next}}
({{ConsumeKafka.java:1018-1022}}), pass {{record.getOffset()}} to
{{offsetTracker.recordConsumed(...)}}, which keeps a per-partition maximum.
{{OffsetTracker.update}} then stops touching offsets.
{{Kafka3ConsumerService.getOffsets}} remains unchanged and still commits
{{storedOffset + 1}}.
An alternative is to carry a {{maxOffset}} on {{ByteRecord}} in
{{nifi-kafka-service-api}}, set by {{ByteRecordBundler}} from the bundle's last
offset, and have {{OffsetTracker.update}} store the maximum. This changes a
public service API class and is less robust against future transforms that
might forget to set the field.
Both approaches produce the same committed offsets. Under either,
{{kafka.offset}} stays as the bundle's first offset and {{kafka.max.offset}}
stays as the last.
h3. Expected Results After Fix
||Strategy||Current commit||Expected commit||
|DEMARCATOR, one bundle of offsets 0-5|1|6|
|DEMARCATOR, several bundles per partition|1 + first offset of last-written
bundle|1 + highest offset in poll|
|FLOW_FILE|6|6|
|RECORD|6|6|
|Rebalance path|6|6|
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
