udayaw opened a new pull request, #40016:
URL: https://github.com/apache/beam/pull/40016
**Please** add a meaningful description for your change here
`ReadFromKafkaDoFn` reports a fully caught-up partition as having a backlog
of 1, never 0.
The backlog handed to the `TimestampPolicy` comes from the restriction
tracker:
```java
new TimestampPolicyContext(
(long) ((HasProgress) tracker).getProgress().getWorkRemaining(),
Instant.now());
```
`GrowableOffsetRangeTracker.getProgress()` computes that as:
```java
final long completedEnd = lastAttemptedOffset == null ? range.getFrom() :
lastAttemptedOffset;
final long remainingEnd = Math.max(completedEnd,
rangeEndEstimator.estimate());
return Progress.from(completedEnd - range.getFrom(), remainingEnd -
completedEnd);
RangeEndEstimator.estimate() is documented as an exclusive end offset, and
in KafkaIO it is consumer.position() + consumer.currentLag().
lastAttemptedOffset, however, is the last inclusive offset claimed. Once at
least one record has been claimed, workRemaining == realBacklog + 1.
```
`RangeEndEstimator.estimate()` is documented as an exclusive end offset, and
in KafkaIO it is `consumer.position() + consumer.currentLag()`.
lastAttemptedOffset, however, is the last inclusive offset claimed. Once at
least one record has been claimed, `workRemaining == realBacklog + 1`.
**Change**
updateWatermarkManually now takes the exclusive next offset and the latest
end
offset estimate directly and computes
`max(estimatedEndOffset, nextOffset) - nextOffset`, matching the gauge. Each
call site already had expectedOffset in hand, maintained as
`rawRecord.offset() + 1 and consumer.position(topicPartition)`.
TimestampPolicy.PartitionContext#getMessageBacklog's javadoc previously read
"latest offset of the partition - last processed record offset", which
describes
the buggy inclusive arithmetic. Reworded to state that both offsets are
exclusive and that zero means fully caught up.
Long.MIN_VALUE, the sentinel latestOffsetEstimator holds when the position is
unknown, still yields a backlog of 0 via the max, matching today's behaviour
on
that path.
------------------------
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example:
`addresses #123`), if applicable. This will automatically add a link to the
pull request in the issue. If you would like the issue to automatically close
on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more
tips on [how to make review process
smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit
[https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more
information about GitHub Actions CI or the [workflows
README](https://github.com/apache/beam/blob/master/.github/workflows/README.md)
to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
