scwhittle commented on code in PR #38919:
URL: https://github.com/apache/beam/pull/38919#discussion_r3648592426
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -118,6 +120,7 @@ private Work(
+ Long.toHexString(workItem.getWorkToken());
this.currentState = TimedState.initialState(startTime);
this.isFailed = false;
+ this.getWorkStreamLatencies = getWorkStreamLatencies;
Review Comment:
instead of saving just populate totalDurationPerState here
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java:
##########
@@ -99,28 +98,41 @@ private static boolean isOutOfMemoryError(@Nullable
Throwable t) {
return false;
}
- /**
- * Processes failures caused by thrown exceptions that occur during
execution of {@link Work}. May
- * attempt to retry execution of the {@link Work} or drop it if it is
invalid.
- */
- public void logAndProcessFailure(
+ public void logAndProcessFailureBatch(
String computationId,
- ExecutableWork executableWork,
+ List<ExecutableWork> executableWorks,
Throwable t,
Consumer<Work> onInvalidWork)
throws Throwable {
- switch (evaluateRetry(computationId, executableWork.work(), t)) {
- case DO_NOT_RETRY:
- // Consider the item invalid. It will eventually be retried by
Windmill if it still needs to
- // be processed.
- onInvalidWork.accept(executableWork.work());
- break;
- case RETRY_LOCALLY:
- // Try again after some delay and at the end of the queue to avoid a
tight loop.
- executeWithDelay(retryLocallyDelayMs, executableWork);
- break;
- case RETHROW_THROWABLE:
- throw t;
+ List<ExecutableWork> worksToRetryLocally = new java.util.ArrayList<>();
+
+ for (ExecutableWork executableWork : executableWorks) {
+ switch (evaluateRetry(computationId, executableWork.work(), t)) {
+ case DO_NOT_RETRY:
+ // Consider the item invalid. It will eventually be retried by
Windmill if it still needs
+ // to
Review Comment:
nit, wrap comment
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java:
##########
@@ -398,9 +386,52 @@ private void commitWorkBatch(
ComputationState computationState,
List<Work> workBatch,
List<Windmill.WorkItemCommitRequest> workItemCommits) {
- checkState(workBatch.size() == 1, "Expected single-key work batch, got: "
+ workBatch.size());
- checkState(workBatch.size() == workItemCommits.size());
- commitSingleKeyWork(computationState, workBatch.get(0),
workItemCommits.get(0));
+ if (workBatch.isEmpty()) {
+ return;
+ }
+ if (workBatch.size() > 1 || multiKeyBundleOptions.multiKeyBundleEnabled())
{
Review Comment:
is there a benefit to sending single keys in multi-key format? Otherwise
seems likely more overhead in protos etc
--
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]
