ifndef-SleePy commented on PR #8401:
URL: https://github.com/apache/paimon/pull/8401#issuecomment-4900749636
> The coordinator committer currently loses watermarks. The old
`CommitterOperator` updates `currentWatermark` in `processWatermark` and passes
it to `committer.combine`, but `CommittingWriteOperatorCoordinator` always uses
`NO_WATERMARK` and has a TODO for watermark handling. This changes committed
snapshot watermarks and can break watermark-based features such as partition
mark-done. Could we either forward per-subtask watermarks through operator
events and combine them consistently, or explicitly reject watermark-dependent
configurations while coordinator commit is enabled?
@JingsongLi Thanks for reviewing. That's indeed a problem. I looked into the
current implementation, and here's what I found:
1. I couldn't find an easy way to reject watermark-dependent configurations
while coordinator commit is enabled — the sink can't reliably tell at planning
time whether the pipeline depends on watermarks.
2. I've added a new commit that uses a SimpleWatermarkValve to emulate
Flink's watermark handling: writers report their watermarks, and the
coordinator aggregates them into the committed watermark. But two problems
remain:
1. It doesn't fully match Flink's behavior — idle watermarks are not
handled. Supporting them would likely require the operator to override
processWatermarkStatus , along with a StatusWatermarkValve -like implementation
on the coordinator side.
2. SimpleWatermarkValve currently reuses HeapPriorityQueue from
flink-runtime . I've verified it works across all Flink versions Paimon
currently supports (1.16–2.2), but since it's a Flink runtime internal, I think
it's cleaner to fork such Flink implementations into Paimon eventually. There's
quite a bit of code to fork though, so it may need some simplification first.
Do you think the SimpleWatermarkValve -based implementation is enough for
the current PR, and we resolve the two problems above in a follow-up PR?
--
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]
