embed_links: Interrupt consume() function on worker timeout.

This fixes a bug introduced in 95b4654
which made the worker simply log a warning about the timeout and then
continue consume()ing the event that should have also been interrupted.

The idea here is to introduce an exception which can be used to
interrupt the consume() process without triggering the regular handling
of exceptions that happens in _handle_consume_exception.

Author: mateuszmandera

zerver/worker/queue_processors.py

@@ -115,6 +115,16 @@ def __str__(self) -> str:
         return f"Timed out after {self.limit * self.event_count} seconds processing {self.event_count} events"
 
 
+class InterruptConsumeException(Exception):
+    """
+    This exception is to be thrown inside event consume function
+    if the intention is to simply interrupt the processing
+    of the current event and normally continue the work of the queue.
+    """
+
+    pass
+
+
 class WorkerDeclarationException(Exception):
     pass
 
@@ -341,6 +351,11 @@ def timer_expired(
         raise WorkerTimeoutException(limit, len(events))
 
     def _handle_consume_exception(self, events: List[Dict[str, Any]], exception: Exception) -> None:
+        if isinstance(exception, InterruptConsumeException):
+            # The exception signals that no further error handling
+            # is needed and the worker can proceed.
+            return
+
         with configure_scope() as scope:
             scope.set_context(
                 "events",
@@ -775,6 +790,7 @@ def timer_expired(
             event["message_id"],
             event["urls"],
         )
+        raise InterruptConsumeException
 
 
 @assign_queue("outgoing_webhooks")