github-actions[bot] commented on code in PR #40610:
URL: https://github.com/apache/beam/pull/40610#discussion_r4284339942
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/ReaderInvocationUtil.java:
##########
@@ -62,11 +86,24 @@ public boolean invokeAdvance(ReaderT reader) throws
IOException {
try (Closeable ignored =
MetricsEnvironment.scopedMetricsContainer(container.getMetricsContainer(stepName)))
{
boolean result = reader.advance();
- container.updateMetrics(stepName);
+ maybeUpdateMetrics(result);
return result;
}
} else {
return reader.advance();
}
}
+
+ private void maybeUpdateMetrics(boolean dataAvailable) {
+ long now = nanoTime.getAsLong();
+ // Always update on the first call, so that the metrics are registered in
Flink right away, and
+ // when the reader has no data, so that the final values of a finished
reader are reported.
+ if (!metricsUpdated
+ || !dataAvailable
Review Comment:
**🟠High**: For `UnboundedSource.UnboundedReader` (used in
`FlinkUnboundedSourceReader` and `UnboundedSourceWrapper`), `start()` /
`advance()` returning `false` means no record is currently available rather
than EOF.
Because both `FlinkUnboundedSourceReader#nextReaderWithData` and
`UnboundedSourceWrapper#run` share a single `ReaderInvocationUtil` instance
across all splits assigned to the subtask and poll them round-robin, if even
one split is idle (or is an already-exhausted `BoundedToUnboundedSourceAdapter`
split in `FlinkUnboundedSourceReader`), polling that split returns `false`
before every record read from an active split. With `|| !dataAvailable`, this
bypasses the 1-second throttle and calls `container.updateMetrics(stepName)` on
every single record (and every 50ms per split when completely idle).
Consider either restricting the `!dataAvailable` flush to
`BoundedSource.BoundedReader` (e.g., passing `reader instanceof
BoundedSource.BoundedReader` or checking it in `maybeUpdateMetrics`), or
flushing final metrics explicitly when the source reader/operator closes.
--
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]
