boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465381343
##########
File path:
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() +
progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() +
progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() -
currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double
elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
I think the combination of Truncate observing windows and Process
consuming elements makes progress/split quite complicated. It seems like a
special case of handling split/progress for a non-root SDF.
If we treat the Truncate only as a special case, we can go with option1 and
let the runner to decide how to make the split decision. In this case, it's
natural for Truncate to report the progress combing with the downstream
progress.
For option2, I'm not sure how a runner computes the work remaining based on
the new metric.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
