udayaw opened a new pull request, #40176:
URL: https://github.com/apache/beam/pull/40176
Might be related to #20908
## Problem
`CustomTimestampPolicyWithLimitedDelay.getWatermark` advances an idle
partition
only when three conditions hold:
| gate | condition |
| --- | --- |
| 1 | `ctx.getMessageBacklog() == 0` |
| 2 | `ctx.getBacklogCheckTime().minus(maxDelay).isAfter(maxEventTimestamp)`
|
| 3 | `maxEventTimestamp.getMillis() > 0` — *"Read at least one record with
a positive timestamp."* |
`maxEventTimestamp` is initialised to
`previousWatermark.orElse(TIMESTAMP_MIN_VALUE).plus(maxDelay)` and is **only
ever
raised by `getTimestampForRecord`**, which runs once per delivered record.
So on a partition which delivers nothing after the job starts:
```
maxEventTimestamp = TIMESTAMP_MIN_VALUE + maxDelay // still hugely
negative, gate 3 fails
getWatermark() → maxEventTimestamp.minus(maxDelay)
= TIMESTAMP_MIN_VALUE // the +maxDelay and
-maxDelay cancel
```
which is why affected partitions report *exactly*
`-290308-12-21T19:59:05.225Z`
rather than a value `maxDelay` away from it. Because a stage's watermark is
the
minimum over its partitions, one such partition pins the whole stage at the
floor, for the life of the job or until that partition's first record
arrives.
**The asymmetry is the defect.** `getWatermark` is driven by the backlog
check —
a timer, roughly every 2s — and is therefore independent of data. Gate 3
makes
that data-independent report conditional on a variable only data can move,
so a
partition with no data holds a live reporting path and a dead source for the
value it reports.
**Two conditions have to coincide, and they do different work.** *Never
delivered* is what leaves `maxEventTimestamp` negative. *Caught up* is what
makes
blocking wrong — had the partition not been caught up, holding at the floor
would
be defensible, since there is then unread data whose timestamps are unknown.
**So it is launch-dependent, not purely producer-dependent.** A partition is
affected when it has zero backlog at launch (a previously drained pipeline
had
fully ingested it, so it resumes at the log end) *and* receives nothing
after.
Two jobs launched minutes apart against the same topic get different affected
sets.
**And it is sticky rather than recurring.** `createTimestampPolicy` is
called per
bundle, so policy instances are rebuilt continuously, but `previousWatermark`
carries the state across — a healthy partition stays healthy and a pinned one
stays pinned.
## Impact
The consequence depends on how the pipeline triggers. Where windowing is on
event time, a pinned partition means no window ever closes, so aggregations
never
fire and output is silently withheld while records accumulate in state — the
symptom described in #20908. Where triggers are processing-time based,
output is
unaffected, but the reported watermark is meaningless: `TIMESTAMP_MIN_VALUE`
is
Beam's sentinel for "no watermark", so a runner's data-freshness metric
cannot
distinguish a caught-up stage from a stalled one, and monitoring keyed on it
is
blind to real lag.
Either way one quiet partition is enough, and it stays pinned until its next
record — which for config, lookup or CDC topics can be hours or days.
## Change
**Gate 3 removed.** It is redundant with gate 1: at `getMessageBacklog() ==
0`
the reader has a position and knows it is at the log end, so there is no
unread
data which could arrive late, whether or not a record has ever been seen.
Having
seen a record adds no safety once the backlog is zero.
**A monotonicity clamp added.** `getWatermark` has two ways to answer — the
idle
path, `backlogCheckTime - maxDelay`, which is roughly now; and the fallback
path,
`maxEventTimestamp - maxDelay`, taken whenever the partition is not caught
up.
The idle path never writes `maxEventTimestamp`, so the two can disagree and
the
answer can move backwards:
```
backlog 0, nothing read yet → idle path → now - maxDelay
first record arrives, backlog 1 → fallback path → maxEventTimestamp -
maxDelay
```
Today gate 3 keeps that regression to seconds, because the idle path cannot
run
until `maxEventTimestamp` holds a real record timestamp. Without the gate
`maxEventTimestamp` may still be the floor, so the watermark would collapse
at
the exact moment a quiet partition finally receives data. `getWatermark` now
remembers the last value it returned and never returns anything earlier.
Note the clamp matters most on the legacy read path, where
`KafkaUnboundedReader.updateAndGetWatermark` assigns the policy's answer
straight
to `lastWatermark` with no max. On the SDF path a regression would be masked
by
`WatermarkEstimators.Manual`, which is already monotonic. Clamping in the
policy
makes the class correct on its own terms rather than relying on its caller.
Watermark semantics are otherwise unchanged:
`Min(now, maxEventTimestamp) - maxDelay`, advanced to
`backlogCheckTime - maxDelay` while caught up.
## Validation
The equivalent of this patch — the same two changes applied as a local
subclass
of the policy — was run on two long-running streaming jobs reading several
hundred Kafka partitions each, instrumented to report per-partition watermark
progress.
Before: 95% and 87% of partitions pinned at launch, still 81% and 60% pinned
47
hours later, the count decaying only as partitions received their first
post-launch record. After: zero pinned partitions on either job, held for 23
consecutive hours including a full night — the quiet period the symptom
requires.
## Tests
- `testWatermarkAdvancesWhenIdleWithoutAnyRecord` — a caught-up partition
which
has delivered no record advances to `backlogCheckTime - maxDelay` instead
of
returning `TIMESTAMP_MIN_VALUE`. Fails before this change.
- `testWatermarkDoesNotRegressWhenFirstRecordArrives` — after advancing while
idle, a first record carrying a much older timestamp does not drag the
watermark back. Fails without the clamp.
- `testCustomTimestampPolicyWithLimitedDelay` — unchanged and still passing,
including its `TIMESTAMP_MIN_VALUE` assertion, which holds a non-zero
backlog
and so exercises the fallback 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]
