[
https://issues.apache.org/jira/browse/NIFI-16566?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Paweł Rzepiński updated NIFI-16566:
-----------------------------------
Description:
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. 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. 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.
was:
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. 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. 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.
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.
> 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
> Priority: Major
>
> 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. 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. 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
