kennknowles commented on code in PR #40020:
URL: https://github.com/apache/beam/pull/40020#discussion_r3936385492
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkItemStatusClient.java:
##########
@@ -188,14 +208,37 @@ public synchronized void setWorker(
return null;
}
- WorkItemStatus status = createStatusUpdate(false);
+ WorkItemStatus status = createStatusUpdate(false, includeCounters);
status.setRequestedLeaseDuration(TimeUtil.toCloudDuration(requestedLeaseDuration));
populateProgress(status);
populateSplitResult(status, dynamicSplitResult);
return execute(status);
}
+ /**
+ * Sends a lightweight lease renewal heartbeat without counter extraction or
progress sampling.
+ *
+ * <p>This can be used during bundle finalization, teardown, or when the
worker is otherwise busy,
+ * to keep the lease active on the Dataflow service without incurring
serialization overhead.
+ */
+ public @Nullable WorkItemServiceState reportLeasePing(Duration
requestedLeaseDuration)
+ throws Exception {
+ checkState(worker != null, "setWorker should be called before
reportLeasePing");
+ if (finalStateSent) {
+ return null;
+ }
+ checkArgument(requestedLeaseDuration != null, "requestLeaseDuration must
be non-null");
Review Comment:
Use `checkArgumentNotNull`
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkItemStatusClient.java:
##########
@@ -188,14 +208,37 @@ public synchronized void setWorker(
return null;
}
- WorkItemStatus status = createStatusUpdate(false);
+ WorkItemStatus status = createStatusUpdate(false, includeCounters);
status.setRequestedLeaseDuration(TimeUtil.toCloudDuration(requestedLeaseDuration));
populateProgress(status);
populateSplitResult(status, dynamicSplitResult);
return execute(status);
}
+ /**
+ * Sends a lightweight lease renewal heartbeat without counter extraction or
progress sampling.
+ *
+ * <p>This can be used during bundle finalization, teardown, or when the
worker is otherwise busy,
+ * to keep the lease active on the Dataflow service without incurring
serialization overhead.
+ */
+ public @Nullable WorkItemServiceState reportLeasePing(Duration
requestedLeaseDuration)
+ throws Exception {
+ checkState(worker != null, "setWorker should be called before
reportLeasePing");
Review Comment:
Use `checkStateNotNull`
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/BatchDataflowWorker.java:
##########
@@ -324,6 +324,34 @@ void executeWork(DataflowWorkExecutor worker,
DataflowWorkProgressUpdater progre
}
}
+ /**
+ * Executes the work, reports any unreported dynamic split, and reports
final success while
+ * keeping progress reporting (and lease renewal) active.
+ *
+ * <p>By keeping the {@link DataflowWorkProgressUpdater} active until {@link
+ * WorkItemStatusClient#reportSuccess} completes, lease renewal heartbeats
continue even if bundle
+ * completion or status reporting encounters delays, preventing lease
expiration.
+ */
+ void executeWork(
Review Comment:
Don't use method overrides. They are a human-oriented stylistic choice that
obscures readability. Give the method a meaningful name that indicates why it
takes more parameters.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkItemStatusClient.java:
##########
@@ -218,6 +261,15 @@ private static boolean isReadLoopAbortedError(Throwable t)
{
private synchronized @Nullable WorkItemServiceState execute(WorkItemStatus
status)
throws IOException {
+ if (finalStateSent && !Boolean.TRUE.equals(status.getCompleted())) {
Review Comment:
Do not use Boolean.TRUE.equals. `getCompleted()` is already a boolean.
--
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]
