Yicong-Huang commented on code in PR #6892:
URL: https://github.com/apache/texera/pull/6892#discussion_r3661381292
##########
amber/src/main/python/core/architecture/handlers/control/end_worker_handler.py:
##########
@@ -37,19 +39,45 @@ async def end_worker(self, req: EmptyRequest) ->
EmptyReturn:
has finished not only the data processing logic, but also the
processing
of all the control messages.
"""
- # Ensure this is really the last message. Read the queued count once
(InternalQueue
- # exposes size(); the base IQueue interface does not) and branch on it.
+ # Ensure this is really the last message that carries work. Read the
+ # queued count once (InternalQueue exposes size(); the base IQueue
+ # interface does not) and branch on it.
input_queue: InternalQueue = self.context.input_queue
- queued_count = input_queue.size()
- if queued_count > 0:
+ if input_queue.size() == 0:
+ # Now we can safely acknowledge that this worker can be terminated.
+ return EmptyReturn()
+
+ # RPC acks (ReturnInvocations for this worker's own fire-and-forget
+ # calls, e.g. worker_execution_completed) race with EndWorker by
+ # design: the coordinator decides to end the worker while its acks are
+ # still in flight. The worker never awaits those acks, so an ack-only
+ # backlog is safe to drop at termination. Anything else fails loudly
+ # AND is put back on the queue so the coordinator's retried EndWorker
+ # finds it processed (mirrors the Scala EndHandler).
+ pending = []
+ while not input_queue.is_empty():
+ pending.append(input_queue.get())
+
+ def is_ack(element) -> bool:
+ return isinstance(element, DCMElement) and isinstance(
+ get_one_of(element.payload, sealed=False), ReturnInvocation
+ )
+
+ if all(is_ack(element) for element in pending):
logger.warning(
- f"Received EndWorker before all {queued_count} queued "
- f"message(s) were processed; failing the RPC so a later "
- f"coordinator retry succeeds once the queue has drained."
+ f"Received EndWorker with only RPC acks left in the queue; "
+ f"proceeding with termination. Pending acks: {pending}"
)
- # Fail this RPC (the counterpart of the Scala EndHandler's
- # Future.exception) so a later coordinator retry succeeds once
- # the queue has drained, instead of dropping the pending message.
- raise RuntimeError("worker still has unprocessed messages")
- # Now we can safely acknowledge that this worker can be terminated.
- return EmptyReturn()
+ return EmptyReturn()
+
+ for element in pending:
Review Comment:
this would change the thread to be a producer (by putting elements into the
queue). can we avoid it?
--
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]
