[Processor] Stop processing events on draining and send signal to continue processing #3120
Conversation
There was a problem hiding this comment.
This PR coincides with #3119, which handles the case where a drain signal is received while we wait for an event.
Since it is changing a bits of the drain registering and the same serve_requests function, we should make sure we align everything before merging.
# Conflicts: # pkg/processor/runtime/python/py/_nuclio_wrapper.py
| except BaseException as exc: | ||
| await self._on_handle_event_error(exc) | ||
| # do not handle an event if a worker is drained | ||
| if not self._discard_events: |
There was a problem hiding this comment.
I think it is worth to "debug"-log the discarded events
| self._is_termination_needed = True | ||
| # if serving loop is waiting for an event, unblock this operation to allow the termination callback to be called | ||
| if self._event_message_length_task: | ||
| self._event_message_length_task.cancel() | ||
|
|
||
| def _on_continue_signal(self, signal_name): | ||
| self._logger.debug_with('Received continue signal', signal=signal_name) |
There was a problem hiding this comment.
self._logger.debug_with('Received signal', signal=signal_name)
for consistency with _on_drain_signal and _on_termination_signal
| w.isDrained.Store(isDrained) | ||
| func (w *Worker) Continue() error { | ||
| if err := w.runtime.Continue(); err != nil { | ||
| return err |
There was a problem hiding this comment.
@liranbg we do the same return err in Drain and Terminate. Err will be wrapped on the next step. We have 1-1 relationships between worker and runtime, so maybe we can leave as is?
| if err = k.SignalWorkersToContinue(); err != nil { | ||
| k.Logger.WarnWith("Failed to signal worker to continue event processing", | ||
| "err", err) | ||
| continue |
There was a problem hiding this comment.
if it is not being able to signaling after 2-3 times I would suggest to panic and let the entire function to restart, perhaps a zombie process is there and we dont want to get stuck with kafka stream not being consumed
Jira - https://jira.iguazeng.com/browse/NUC-119
This pull request addresses an issue where the processor sent events to the runtime after draining and before reconnection. Now, the wrapper will skip all events received after draining and wait for the
SIGCONTsignal to resume event processing.These changes also allow us to get rid of
isDrainedchecks on processor side.