boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467266336
##########
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:
Are you saying that one transform can report the progress metric per
element/window? For example, saying that the Truncate is on window1 and there
are 3 windows in total, the downstream sdf progress is 0.5. Then the Truncate
should report [(work_completed=1, work_remaining=0), (work_completed=1,
work_remaining=0), (work_completed=0, work_remaining=1)] and the sdf should
report [(work_completed=0.5, work_remaining=0.5)]? The the actual completed
should be 1 + 1 * 0.5?
----------------------------------------------------------------
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:
[email protected]
