scwhittle commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3727247296
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -395,12 +395,21 @@ BoundedQueueExecutorWorkHandleImpl
createBudgetHandle(Work work, long bytes) {
if (keyGroupWorkQueue == null) {
return null;
}
- @Nullable QueuedWork queuedWork =
keyGroupWorkQueue.pollWork(computationId, keyGroup);
- if (queuedWork == null) {
- return null;
+ while (true) {
+ @Nullable QueuedWork queuedWork =
keyGroupWorkQueue.pollWork(computationId, keyGroup);
+ if (queuedWork == null) {
+ return null;
+ }
+ Work work = queuedWork.getWork().work();
+ if (work.isFailed()) {
+ queuedWork.getHandle().close();
+ work.getComputationState()
Review Comment:
I'm wondering if there is someway to make the control flow clearer for these
work items. One idea would be to not handle the failure here but pass the
failed work items back to the caller of poll. Then
StreamingWorkExecutionContext could possibly have a abortedworkitemhhandler
similar to the KeyTransitionListener. Then we can have consistent error
handling to StreamingWorkScheduler.handleProcessWorkFailure by passing it in
instead of duplicating what it does.
Or perhaps we could collect the failed work in StreamingWorkExecutionContext
and then handle it later after execution completes instead of using an injected
function.
--
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]
