You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "arunpandianp (via GitHub)" <gi...@apache.org> on 2026/06/11 13:37:18 UTC
[PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
arunpandianp opened a new pull request, #38920:
URL: https://github.com/apache/beam/pull/38920
Drop failed work in BoundedQueueExecutor::pollWork before it reaches the harness threads for execution.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3396344073
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutorTest.java:
##########
@@ -553,4 +553,71 @@ public void testPollWorkWithLinkedBlockingQueue() throws Exception {
blockerStop.countDown();
testExecutor.shutdown();
}
+
+ @Test
+ public void testPollWorkDropsFailedWork() throws Exception {
+ BoundedQueueExecutor testExecutor =
+ new BoundedQueueExecutor(
+ /* initialMaximumPoolSize= */ 1,
+ /* keepAliveTime= */ 60,
+ /* unit= */ TimeUnit.SECONDS,
+ /* maximumElementsOutstanding= */ 100,
+ /* maximumBytesOutstanding= */ 10000000,
+ new ThreadFactoryBuilder().setNameFormat("testStealing-%d").setDaemon(true).build(),
+ useFairMonitor,
+ /*useKeyGroupWorkQueue=*/ true);
+
+ // Create blocker task to occupy the worker thread
+ CountDownLatch blockerStart = new CountDownLatch(1);
+ CountDownLatch blockerStop = new CountDownLatch(1);
+ ExecutableWork blockerWork =
+ createWorkWithCompIdAndKeyGroup(
+ "blockerComp",
+ DEFAULT_KEY_GROUP,
+ ignored -> {
+ blockerStart.countDown();
+ try {
+ blockerStop.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ testExecutor.execute(blockerWork, 0);
+ blockerStart.await();
+
+ Work.KeyGroup keyGroup1 = Work.KeyGroup.create(1, 1);
+
+ // Create executable tasks
+ ExecutableWork work1 = createWorkWithCompIdAndKeyGroup("compA", keyGroup1, ignored -> {});
+ ExecutableWork work2 = createWorkWithCompIdAndKeyGroup("compA", keyGroup1, ignored -> {});
+
+ // Mark work1 as failed
+ work1.work().setFailed();
+
+ // Enqueue tasks
+ testExecutor.execute(work1, 100);
+ testExecutor.execute(work2, 150);
+
+ // Total outstanding elements must be 3 (blocker + work1 + work2)
+ assertEquals(3, testExecutor.elementsOutstanding());
+
+ // Steal work from keyGroup1.
+ // The first work in queue is work1, which is failed.
+ // It should be dropped, its handle closed, and work2 should be returned.
+ try (BoundedQueueExecutorWorkHandleImpl stealHandle = testExecutor.createBudgetHandle(0, 0L)) {
+ ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup1, stealHandle);
+ assertNotNull(stolen);
+ assertEquals(work2, stolen);
+ // blocker (1) + work2 (1) = 2. work1 (1) should have been released.
+ assertEquals(2, testExecutor.elementsOutstanding());
+ }
+ // work2 should also be released now because stealHandle is closed.
+ // blocker (1) = 1.
+ assertEquals(1, testExecutor.elementsOutstanding());
+
+ // Unblock the blocker and shut down
+ blockerStop.countDown();
+ testExecutor.shutdown();
Review Comment:
If any assertion fails during the test execution (for example, `assertEquals(3, testExecutor.elementsOutstanding())`), the `blockerStop.countDown()` and `testExecutor.shutdown()` calls will be skipped. This will cause the blocker thread to hang indefinitely on `blockerStop.await()`, leading to thread leaks and potential test suite hangs.
We should wrap the execution and assertions in a `try-finally` block to ensure that resources are always cleaned up and the blocker thread is unblocked.
```java
try {
// Enqueue tasks
testExecutor.execute(work1, 100);
testExecutor.execute(work2, 150);
// Total outstanding elements must be 3 (blocker + work1 + work2)
assertEquals(3, testExecutor.elementsOutstanding());
// Steal work from keyGroup1.
// The first work in queue is work1, which is failed.
// It should be dropped, its handle closed, and work2 should be returned.
try (BoundedQueueExecutorWorkHandleImpl stealHandle = testExecutor.createBudgetHandle(0, 0L)) {
ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup1, stealHandle);
assertNotNull(stolen);
assertEquals(work2, stolen);
// blocker (1) + work2 (1) = 2. work1 (1) should have been released.
assertEquals(2, testExecutor.elementsOutstanding());
}
// work2 should also be released now because stealHandle is closed.
// blocker (1) = 1.
assertEquals(1, testExecutor.elementsOutstanding());
} finally {
// Unblock the blocker and shut down
blockerStop.countDown();
testExecutor.shutdown();
}
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4831858671
Current status: waiting for other PR to merge to improve tests for this one before merging
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-5045897999
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @Abacn added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4809805938
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @Abacn added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3779476116
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutorTest.java:
##########
@@ -569,10 +598,83 @@ public void testPollWorkWithLinkedBlockingQueue() throws Exception {
ExecutableWork work = createWorkWithCompIdAndKeyGroup("compA", keyGroup, ignored -> {});
testExecutor.execute(work, 100);
- ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup, stealHandle);
+ ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup, stealHandle, ignored -> {});
assertNull(stolen);
blockerStop.countDown();
testExecutor.shutdown();
}
+
+ @Test
+ public void testPollWork_skipsFailedWorkAndCallsOnFailedWorkHandler() throws Exception {
+ BoundedQueueExecutor testExecutor =
+ new BoundedQueueExecutor(
+ 1,
+ 60,
+ TimeUnit.SECONDS,
+ 100,
+ 10000000,
+ new ThreadFactoryBuilder().setNameFormat("testPollWork-%d").setDaemon(true).build(),
+ useFairMonitor,
+ /* useKeyGroupWorkQueue= */ true);
+
+ CountDownLatch blockerStart = new CountDownLatch(1);
+ CountDownLatch blockerStop = new CountDownLatch(1);
+ AtomicReference<BoundedQueueExecutorWorkHandle> blockerHandleRef = new AtomicReference<>();
+ ExecutableWork blockerWork =
+ createWorkWithHandle(
+ "compA",
+ DEFAULT_KEY_GROUP,
+ (work, handle) -> {
+ blockerHandleRef.set(handle);
+ blockerStart.countDown();
+ try {
+ blockerStop.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ testExecutor.execute(blockerWork, 10);
+ blockerStart.await();
+ BoundedQueueExecutorWorkHandleImpl stealHandle =
+ (BoundedQueueExecutorWorkHandleImpl) blockerHandleRef.get();
+ assertNotNull(stealHandle);
+
+ Work.KeyGroup keyGroup = Work.KeyGroup.create(1, 1);
+ FailedWorkHandler onFailedWorkHandler = mock(FailedWorkHandler.class);
+
+ ExecutableWork work1 =
+ createWorkWithCompIdAndKeyGroupAndWorkToken("compA", keyGroup, 101, ignored -> {});
+ ExecutableWork work2 =
+ createWorkWithCompIdAndKeyGroupAndWorkToken("compA", keyGroup, 102, ignored -> {});
+
+ // Enqueue both tasks (they will wait in the queue because the thread is blocked).
+ testExecutor.execute(work1, 100);
+ testExecutor.execute(work2, 150);
+
+ assertEquals(3, testExecutor.elementsOutstanding());
+ assertEquals(260, testExecutor.bytesOutstanding());
+
+ // Mark work1 as failed while waiting in the queue.
+ work1.work().setFailed();
+
+ // pollWork should skip work1, close work1's handle, invoke
+ // onFailedWorkHandler callback, and return work2.
+ ExecutableWork stolen =
+ testExecutor.pollWork("compA", keyGroup, stealHandle, onFailedWorkHandler);
+ assertNotNull(stolen);
+ assertEquals(work2, stolen);
+
+ verify(onFailedWorkHandler).onFailedWork(work1.work());
+
+ // Verify stealHandle merged (blockerWork: 10 bytes, work2: 150 bytes).
+ assertEquals(160, stealHandle.bytes());
+
+ // Polling again should return null since no more tasks exist for keyGroup.
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -361,6 +361,7 @@ public void start(
this.workQueueExecutor = workQueueExecutor;
this.budgetHandle = budgetHandle;
this.keyTransitionListener = keyTransitionListener;
+ this.onFailedWorkHandler = onFailedWorkHandler;
Review Comment:
Added a unconditional checkNotNull here.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContextTest.java:
##########
@@ -577,11 +579,17 @@ public void testAdvance_noMoreWork() throws Exception {
createMockWork(
workItem1, Watermarks.builder().setInputDataWatermark(Instant.EPOCH).build());
- when(mockExecutor.pollWork(eq(COMPUTATION_ID), eq(work1.getKeyGroup()), eq(mockHandle)))
+ when(mockExecutor.pollWork(eq(COMPUTATION_ID), eq(work1.getKeyGroup()), eq(mockHandle), any()))
.thenReturn(null);
executionContext.start(
- work1, workExecutor, mockExecutor, mockHandle, null, (oldWork, newWork) -> {});
+ work1,
+ workExecutor,
+ mockExecutor,
+ mockHandle,
+ null,
+ (oldWork, newWork) -> {},
+ ignored -> {});
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutorTest.java:
##########
@@ -511,7 +540,7 @@ public void testPollWork() throws Exception {
assertEquals(3, testExecutor.elementsOutstanding());
// Steal work2 using pollWork with compA and keyGroup2
- ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup2, stealHandle);
+ ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup2, stealHandle, ignored -> {});
Review Comment:
done.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java:
##########
@@ -1524,13 +1524,144 @@ public void testCompleteCommit_retryableFailureTriggersReExecution() throws Exce
worker.stop();
}
+ @Test
+ public void testMultiKeyCommit_queuedWorkItemFailsAndSubsequentWorkItemPickedUp()
+ throws Exception {
+ if (!streamingEngine) {
+ return;
+ }
+ BlockingKvDoFn.reset();
+ StreamingDataflowWorker worker = makeMultiKeyEnabledWorker(new BlockingKvDoFn());
+ worker.start();
+
+ String batchInputText1 =
+ "work {"
+ + " computation_id: \""
+ + DEFAULT_COMPUTATION_ID
+ + "\""
+ + " input_data_watermark: 0"
+ + " work {"
+ + " key: \"key1\""
+ + " sharding_key: 1"
+ + " work_token: 1"
+ + " cache_token: 2"
+ + " key_group { high: 0 low: 1 }"
+ + " message_bundles {"
+ + " source_computation_id: \""
+ + DEFAULT_SOURCE_COMPUTATION_ID
+ + "\""
+ + " messages {"
+ + " timestamp: 0"
+ + " data: \"data1\""
+ + " }"
+ + " }"
+ + " }"
+ + " work {"
+ + " key: \"key2\""
+ + " sharding_key: 2"
+ + " work_token: 2"
+ + " cache_token: 3"
+ + " key_group { high: 0 low: 1 }"
+ + " message_bundles {"
+ + " source_computation_id: \""
+ + DEFAULT_SOURCE_COMPUTATION_ID
+ + "\""
+ + " messages {"
+ + " timestamp: 0"
+ + " data: \"data2\""
+ + " }"
+ + " }"
+ + " }"
+ + "}";
+
+ String batchInputText2 =
+ "work {"
+ + " computation_id: \""
+ + DEFAULT_COMPUTATION_ID
+ + "\""
+ + " input_data_watermark: 0"
+ + " work {"
+ + " key: \"key2\""
+ + " sharding_key: 2"
+ + " work_token: 3"
+ + " cache_token: 4"
+ + " key_group { high: 0 low: 1 }"
+ + " message_bundles {"
+ + " source_computation_id: \""
+ + DEFAULT_SOURCE_COMPUTATION_ID
+ + "\""
+ + " messages {"
+ + " timestamp: 0"
+ + " data: \"data3\""
+ + " }"
+ + " }"
+ + " }"
+ + "}";
+ Windmill.GetWorkResponse batchInput1 =
+ buildInput(
+ batchInputText1,
+ CoderUtils.encodeToByteArray(
+ CollectionCoder.of(IntervalWindow.getCoder()),
+ Collections.singletonList(DEFAULT_WINDOW)));
+ Windmill.GetWorkResponse batchInput2 =
+ buildInput(
+ batchInputText2,
+ CoderUtils.encodeToByteArray(
+ CollectionCoder.of(IntervalWindow.getCoder()),
+ Collections.singletonList(DEFAULT_WINDOW)));
+
+ server.whenGetDataCalled().answerByDefault(StreamingDataflowWorkerTest::emptyDataResponder);
+
+ server.whenGetWorkCalled().thenReturn(batchInput1).thenReturn(batchInput2);
+ server.waitForEmptyWorkQueue();
+
+ // Wait for key1 to start processing and block on BlockingKvDoFn.
+ BlockingKvDoFn.counter.get().acquire(1);
+
+ // Fail key2 (work token 2) via failed heartbeat while key1 is still processing.
+ ComputationHeartbeatResponse.Builder failedHeartbeat =
+ ComputationHeartbeatResponse.newBuilder();
+ failedHeartbeat
+ .setComputationId(DEFAULT_COMPUTATION_ID)
+ .addHeartbeatResponsesBuilder()
+ .setCacheToken(3)
+ .setWorkToken(2)
+ .setShardingKey(2)
+ .setFailed(true);
+ server.sendFailedHeartbeats(Collections.singletonList(failedHeartbeat.build()));
+
+ // Unblock key1 to allow bundle to poll key2 (token 2 -> failed, skipped) and key2 (token 3).
+ BlockingKvDoFn.blocker.get().countDown();
Review Comment:
The Fake windmill server calls the heartbeat processor synchronously in sendFailedHeartbeats. So this is not racy. verfied by running it 100 times.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-5011246303
Reminder, please take a look at this pr: @Abacn
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4779591232
Reminder, please take a look at this pr: @Abacn
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4925484236
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @Abacn added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-5117992049
Reminder, please take a look at this pr: @Abacn
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3415998150
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -391,12 +391,18 @@ BoundedQueueExecutorWorkHandleImpl createBudgetHandle(int elements, 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;
+ }
+ if (queuedWork.getWork().work().isFailed()) {
+ queuedWork.getHandle().close();
Review Comment:
Thanks for catching it. Will fix it with a test, after https://github.com/apache/beam/pull/38814 is merged
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-5119827795
@arunpandianp I think you can sync and update tests now that error handling is merged
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-5200726285
@scwhittle this is ready for another look. Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3413508642
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -391,12 +391,18 @@ BoundedQueueExecutorWorkHandleImpl createBudgetHandle(int elements, 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;
+ }
+ if (queuedWork.getWork().work().isFailed()) {
+ queuedWork.getHandle().close();
Review Comment:
this is just updating the counters, but it is not updating that this key was scheduled for processing. I think that somehow ActiveWorkState.completeWorkAndGetNextWorkForKey needs to be called so that we note that this key is completed and other work for the key can schedule.
Maybe a unit test at a higher level would help verify this behavior is working properly.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4681168376
R: @scwhittle
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3756490940
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -361,6 +361,7 @@ public void start(
this.workQueueExecutor = workQueueExecutor;
this.budgetHandle = budgetHandle;
this.keyTransitionListener = keyTransitionListener;
+ this.onFailedWorkHandler = onFailedWorkHandler;
Review Comment:
if multikeybundleoptions say it is enabled, verify that failedworkhandler is non-null here?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContextTest.java:
##########
@@ -577,11 +579,17 @@ public void testAdvance_noMoreWork() throws Exception {
createMockWork(
workItem1, Watermarks.builder().setInputDataWatermark(Instant.EPOCH).build());
- when(mockExecutor.pollWork(eq(COMPUTATION_ID), eq(work1.getKeyGroup()), eq(mockHandle)))
+ when(mockExecutor.pollWork(eq(COMPUTATION_ID), eq(work1.getKeyGroup()), eq(mockHandle), any()))
.thenReturn(null);
executionContext.start(
- work1, workExecutor, mockExecutor, mockHandle, null, (oldWork, newWork) -> {});
+ work1,
+ workExecutor,
+ mockExecutor,
+ mockHandle,
+ null,
+ (oldWork, newWork) -> {},
+ ignored -> {});
Review Comment:
change to one that fails if ever called throughout this test? similar with transition?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java:
##########
@@ -1524,13 +1524,144 @@ public void testCompleteCommit_retryableFailureTriggersReExecution() throws Exce
worker.stop();
}
+ @Test
+ public void testMultiKeyCommit_queuedWorkItemFailsAndSubsequentWorkItemPickedUp()
+ throws Exception {
+ if (!streamingEngine) {
+ return;
+ }
+ BlockingKvDoFn.reset();
+ StreamingDataflowWorker worker = makeMultiKeyEnabledWorker(new BlockingKvDoFn());
+ worker.start();
+
+ String batchInputText1 =
+ "work {"
+ + " computation_id: \""
+ + DEFAULT_COMPUTATION_ID
+ + "\""
+ + " input_data_watermark: 0"
+ + " work {"
+ + " key: \"key1\""
+ + " sharding_key: 1"
+ + " work_token: 1"
+ + " cache_token: 2"
+ + " key_group { high: 0 low: 1 }"
+ + " message_bundles {"
+ + " source_computation_id: \""
+ + DEFAULT_SOURCE_COMPUTATION_ID
+ + "\""
+ + " messages {"
+ + " timestamp: 0"
+ + " data: \"data1\""
+ + " }"
+ + " }"
+ + " }"
+ + " work {"
+ + " key: \"key2\""
+ + " sharding_key: 2"
+ + " work_token: 2"
+ + " cache_token: 3"
+ + " key_group { high: 0 low: 1 }"
+ + " message_bundles {"
+ + " source_computation_id: \""
+ + DEFAULT_SOURCE_COMPUTATION_ID
+ + "\""
+ + " messages {"
+ + " timestamp: 0"
+ + " data: \"data2\""
+ + " }"
+ + " }"
+ + " }"
+ + "}";
+
+ String batchInputText2 =
+ "work {"
+ + " computation_id: \""
+ + DEFAULT_COMPUTATION_ID
+ + "\""
+ + " input_data_watermark: 0"
+ + " work {"
+ + " key: \"key2\""
+ + " sharding_key: 2"
+ + " work_token: 3"
+ + " cache_token: 4"
+ + " key_group { high: 0 low: 1 }"
+ + " message_bundles {"
+ + " source_computation_id: \""
+ + DEFAULT_SOURCE_COMPUTATION_ID
+ + "\""
+ + " messages {"
+ + " timestamp: 0"
+ + " data: \"data3\""
+ + " }"
+ + " }"
+ + " }"
+ + "}";
+ Windmill.GetWorkResponse batchInput1 =
+ buildInput(
+ batchInputText1,
+ CoderUtils.encodeToByteArray(
+ CollectionCoder.of(IntervalWindow.getCoder()),
+ Collections.singletonList(DEFAULT_WINDOW)));
+ Windmill.GetWorkResponse batchInput2 =
+ buildInput(
+ batchInputText2,
+ CoderUtils.encodeToByteArray(
+ CollectionCoder.of(IntervalWindow.getCoder()),
+ Collections.singletonList(DEFAULT_WINDOW)));
+
+ server.whenGetDataCalled().answerByDefault(StreamingDataflowWorkerTest::emptyDataResponder);
+
+ server.whenGetWorkCalled().thenReturn(batchInput1).thenReturn(batchInput2);
+ server.waitForEmptyWorkQueue();
+
+ // Wait for key1 to start processing and block on BlockingKvDoFn.
+ BlockingKvDoFn.counter.get().acquire(1);
+
+ // Fail key2 (work token 2) via failed heartbeat while key1 is still processing.
+ ComputationHeartbeatResponse.Builder failedHeartbeat =
+ ComputationHeartbeatResponse.newBuilder();
+ failedHeartbeat
+ .setComputationId(DEFAULT_COMPUTATION_ID)
+ .addHeartbeatResponsesBuilder()
+ .setCacheToken(3)
+ .setWorkToken(2)
+ .setShardingKey(2)
+ .setFailed(true);
+ server.sendFailedHeartbeats(Collections.singletonList(failedHeartbeat.build()));
+
+ // Unblock key1 to allow bundle to poll key2 (token 2 -> failed, skipped) and key2 (token 3).
+ BlockingKvDoFn.blocker.get().countDown();
Review Comment:
is this racy with receiving the heartbeats sent above? will the test be flaky?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutorTest.java:
##########
@@ -511,7 +540,7 @@ public void testPollWork() throws Exception {
assertEquals(3, testExecutor.elementsOutstanding());
// Steal work2 using pollWork with compA and keyGroup2
- ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup2, stealHandle);
+ ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup2, stealHandle, ignored -> {});
Review Comment:
instead of ingnoring, have a failure handler that we can assert is not called
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutorTest.java:
##########
@@ -569,10 +598,83 @@ public void testPollWorkWithLinkedBlockingQueue() throws Exception {
ExecutableWork work = createWorkWithCompIdAndKeyGroup("compA", keyGroup, ignored -> {});
testExecutor.execute(work, 100);
- ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup, stealHandle);
+ ExecutableWork stolen = testExecutor.pollWork("compA", keyGroup, stealHandle, ignored -> {});
assertNull(stolen);
blockerStop.countDown();
testExecutor.shutdown();
}
+
+ @Test
+ public void testPollWork_skipsFailedWorkAndCallsOnFailedWorkHandler() throws Exception {
+ BoundedQueueExecutor testExecutor =
+ new BoundedQueueExecutor(
+ 1,
+ 60,
+ TimeUnit.SECONDS,
+ 100,
+ 10000000,
+ new ThreadFactoryBuilder().setNameFormat("testPollWork-%d").setDaemon(true).build(),
+ useFairMonitor,
+ /* useKeyGroupWorkQueue= */ true);
+
+ CountDownLatch blockerStart = new CountDownLatch(1);
+ CountDownLatch blockerStop = new CountDownLatch(1);
+ AtomicReference<BoundedQueueExecutorWorkHandle> blockerHandleRef = new AtomicReference<>();
+ ExecutableWork blockerWork =
+ createWorkWithHandle(
+ "compA",
+ DEFAULT_KEY_GROUP,
+ (work, handle) -> {
+ blockerHandleRef.set(handle);
+ blockerStart.countDown();
+ try {
+ blockerStop.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ testExecutor.execute(blockerWork, 10);
+ blockerStart.await();
+ BoundedQueueExecutorWorkHandleImpl stealHandle =
+ (BoundedQueueExecutorWorkHandleImpl) blockerHandleRef.get();
+ assertNotNull(stealHandle);
+
+ Work.KeyGroup keyGroup = Work.KeyGroup.create(1, 1);
+ FailedWorkHandler onFailedWorkHandler = mock(FailedWorkHandler.class);
+
+ ExecutableWork work1 =
+ createWorkWithCompIdAndKeyGroupAndWorkToken("compA", keyGroup, 101, ignored -> {});
+ ExecutableWork work2 =
+ createWorkWithCompIdAndKeyGroupAndWorkToken("compA", keyGroup, 102, ignored -> {});
+
+ // Enqueue both tasks (they will wait in the queue because the thread is blocked).
+ testExecutor.execute(work1, 100);
+ testExecutor.execute(work2, 150);
+
+ assertEquals(3, testExecutor.elementsOutstanding());
+ assertEquals(260, testExecutor.bytesOutstanding());
+
+ // Mark work1 as failed while waiting in the queue.
+ work1.work().setFailed();
+
+ // pollWork should skip work1, close work1's handle, invoke
+ // onFailedWorkHandler callback, and return work2.
+ ExecutableWork stolen =
+ testExecutor.pollWork("compA", keyGroup, stealHandle, onFailedWorkHandler);
+ assertNotNull(stolen);
+ assertEquals(work2, stolen);
+
+ verify(onFailedWorkHandler).onFailedWork(work1.work());
+
+ // Verify stealHandle merged (blockerWork: 10 bytes, work2: 150 bytes).
+ assertEquals(160, stealHandle.bytes());
+
+ // Polling again should return null since no more tasks exist for keyGroup.
Review Comment:
how about a failed work3 to cover the case where we clear the failed stuff and there is not another item available?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-5251079877
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/38920?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 55.96%. Comparing base ([`eab1bce`](https://app.codecov.io/gh/apache/beam/commit/eab1bceed4f71d52fded363e0ec595584d5c0d64?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`cce0bb3`](https://app.codecov.io/gh/apache/beam/commit/cce0bb3cced1f89e975cb8c5b19488c1a44eac36?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #38920 +/- ##
============================================
- Coverage 58.80% 55.96% -2.84%
+ Complexity 7638 2276 -5362
============================================
Files 1663 1115 -548
Lines 216332 175831 -40501
Branches 6536 1466 -5070
============================================
- Hits 127219 98408 -28811
+ Misses 84401 74915 -9486
+ Partials 4712 2508 -2204
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/38920/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [java](https://app.codecov.io/gh/apache/beam/pull/38920/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.64% <100.00%> (+3.32%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/beam/pull/38920?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4893342590
Reminder, please take a look at this pr: @Abacn
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4681154683
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request improves the robustness of the BoundedQueueExecutor by ensuring that failed work items are filtered out during the polling process. By identifying and dropping these items before they reach the execution threads, the system avoids unnecessary processing overhead and potential errors associated with failed tasks.
### Highlights
* **Failed Work Handling**: Updated BoundedQueueExecutor::pollWork to automatically identify and drop failed work items before they are processed by harness threads.
* **Queue Management**: Implemented a loop in the polling logic to skip over failed work, ensuring that only valid work is returned for execution.
* **Resource Cleanup**: Added explicit handle closure for failed work items to ensure resources are properly released.
<details>
<summary><b>New Features</b></summary>
🧠 You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38920:
URL: https://github.com/apache/beam/pull/38920#issuecomment-4682100022
Assigning reviewers:
R: @Abacn added as fallback since no labels match configuration
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "arunpandianp (via GitHub)" <gi...@apache.org>.
arunpandianp commented on code in PR #38920:
URL: https://github.com/apache/beam/pull/38920#discussion_r3756373686
##########
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:
Good idea! Updated to send in a callback from StreamingWorkScheduler. Presubmits are still running. Ready for another look.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #38920:
URL: https://github.com/apache/beam/pull/38920
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Streaming] [Multi Key] Drop failed work in BoundedQueueExecutor::pollWork [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org