lgbo-ustc opened a new issue, #12341: URL: https://github.com/apache/gluten/issues/12341
### Description # Update Watermark Gauge in Gluten Output Collector ## Background `GlutenOutputCollector` implements `WatermarkGaugeExposingOutput`, but its internal `WatermarkGauge` is not updated when watermarks are emitted. As a result, Flink runtime metrics may not reflect the current watermark even though watermark events are forwarded downstream. This is a low-risk first step toward improving Gluten Flink watermark support. It only fixes watermark observability and does not change data record routing or native watermark generation semantics. ## Scope - Update the Java-side `GlutenOutputCollector` watermark gauge when watermarks are emitted. - Preserve existing watermark routing behavior. - Preserve existing data record routing behavior. - Add focused unit tests for the collector-level behavior. ## Tasks - [ ] Update `GlutenOutputCollector.emitWatermark(...)` to set the current watermark in `WatermarkGauge` before broadcasting the watermark to all outputs. - [ ] Update the `StatefulWatermark` branch in `GlutenOutputCollector.collect(...)` to set the current watermark in `WatermarkGauge` before forwarding the watermark to the output selected by `nodeId`. - [ ] Add a unit test verifying that `emitWatermark(...)` updates the gauge and broadcasts the watermark to all outputs. - [ ] Add a unit test verifying that `collect(StreamRecord<StatefulWatermark>)` updates the gauge and keeps the existing `nodeId`-based routing behavior. ## Non-Goals - Do not change data record routing. - Do not change `StatefulElement.nodeId` to output mapping. - Do not change whether watermarks are broadcast or routed by `nodeId` in existing paths. - Do not add `WatermarkStatus` support. - Do not implement idle source handling. - Do not implement periodic watermark emission. - Do not implement end-of-input or final watermark handling. - Do not change velox4j or native Velox code. ## Acceptance Criteria - `GlutenOutputCollector.getWatermarkGauge().getValue()` advances after `emitWatermark(...)` is called. - `GlutenOutputCollector.getWatermarkGauge().getValue()` advances after a `StatefulWatermark` is passed through `collect(...)`. - `emitWatermark(...)` continues to broadcast watermarks to all outputs. - `collect(StreamRecord<StatefulWatermark>)` continues to route the watermark only to the output matching the watermark's `nodeId`. - Existing data record handling remains unchanged. ### Gluten version None -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
