gemini-code-assist[bot] commented on code in PR #38921:
URL: https://github.com/apache/beam/pull/38921#discussion_r3396546584
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java:
##########
@@ -216,6 +216,8 @@ public void scheduleWork(
Work.ProcessingContext processingContext,
boolean drainMode,
ImmutableList<LatencyAttribution> getWorkStreamLatencies) {
+ // Before any processing starts, call any pending OnCommit callbacks
+
commitFinalizer.finalizeCommits(workItem.getSourceState().getFinalizeIdsList());
Review Comment:
Moving `commitFinalizer.finalizeCommits(...)` to `scheduleWork` executes it
on the work scheduling/dispatching thread instead of the worker/harness threads.
Since finalizers (such as `CheckpointMark.finalizeCheckpoint()`) can execute
arbitrary user code or perform blocking network I/O (e.g., committing offsets
to Kafka or Pub/Sub), executing them synchronously on the scheduling thread can
block the main dispatching loop. This can lead to severe performance
bottlenecks, increased latency, or thread starvation for the entire worker.
If the goal is to ensure finalizers are executed even if the work item is
cancelled, consider offloading this execution to an executor/thread pool
asynchronously, or handling it within the cancellation/cleanup lifecycle of the
`Work` item itself.
--
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]
