fix: add the error to the global result queue#315
Merged
Conversation
Signed-off-by: Vigith Maurice <[email protected]>
Contributor
|
To reproduce this bug we injected a runtime error in the pynumaflow code: def _process_requests(
self,
context: NumaflowServicerContext,
request_iterator: Iterator[map_pb2.MapRequest],
result_queue: SyncIterator,
):
try:
# read through all incoming requests and submit to the
# threadpool for invocation
for request in request_iterator:
_ = self.executor.submit(self._invoke_map, context, request, result_queue)
if should_we_panic(): # <--- inject raising error
raise RuntimeError("random runtime error in MapFn during reading requests!!")
# wait for all tasks to finish after all requests exhausted
self.executor.shutdown(wait=True)
# Indicate to the result queue that no more messages left to process
result_queue.put(STREAM_EOF)
except BaseException:
_LOGGER.critical("MapFn Error, re-raising the error", exc_info=True)BeforeWithout the fix, the container gets stuck on exception as the main thread is still listening for events from the request_queue. UDF logs: Numa logs: After def _process_requests(
self,
context: NumaflowServicerContext,
request_iterator: Iterator[map_pb2.MapRequest],
result_queue: SyncIterator,
):
try:
# read through all incoming requests and submit to the
# threadpool for invocation
for request in request_iterator:
_ = self.executor.submit(self._invoke_map, context, request, result_queue)
if should_we_panic():
raise RuntimeError("random runtime error in MapFn during reading requests!!")
# wait for all tasks to finish after all requests exhausted
self.executor.shutdown(wait=True)
# Indicate to the result queue that no more messages left to process
result_queue.put(STREAM_EOF)
except BaseException as e:
_LOGGER.critical("MapFn Error, re-raising the error", exc_info=True)
result_queue.put(e) # <------- fixWith the fix in place, the udf is able to restart along with the numa container exiting gracefully: container restarts: udf logs: numa logs: PendingReproduce scenario where we get the following error: |
Contributor
|
Further issue reproduction attempt details. We tried sending SIGTERM to the numa container after the UDF gets stuck with the manually added runtime exception in the |
Contributor
|
Further issue reproduction attempt details. Simply added shutdown of the executor followed by submitting another job to it: for request in request_iterator:
if should_we_panic() and not hasattr(self, '_shutdown_once'):
self._shutdown_once = True
self.executor.shutdown(wait=False)
_ = self.executor.submit(self._invoke_map, context, request, result_queue)Can see the same error getting raised in the UDF udf: Tried the fix and the error remains the same, but the containers restart: |
vaibhavtiwari33
approved these changes
Feb 11, 2026
Contributor
|
I was able to reproduce the error (similar to what @vaibhavtiwari33 did). The only possible scenario I could come up with is that the stream ends normally and gets recreated. def _process_requests(
self,
context: NumaflowServicerContext,
request_iterator: Iterable[map_pb2.MapRequest],
result_queue: SyncIterator,
):
_LOGGER.info("Starting _process_requests")
try:
# read through all incoming requests and submit to the
# threadpool for invocation
count = 0
for request in request_iterator:
count += 1
_ = self.executor.submit(self._invoke_map, context, request, result_queue)
if count > 5:
_LOGGER.info("simulating stream end after %d requests", count)
break
# wait for all tasks to finish after all requests exhausted
self.executor.shutdown(wait=True)
# Indicate to the result queue that no more messages left to process
result_queue.put(STREAM_EOF)Error: |
kohlisid
approved these changes
Feb 11, 2026
Contributor
|
Tested with current change. It is causing UDF to restart |
BulkBeing
approved these changes
Feb 11, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes numaproj/numaflow#3216
The top level error handler for map and source-transformer was missing pushing to global result queue. This means, UDF errors will be handled, but not the top level service errors.