add timeout to ActivateJobs request

pyzeebe/errors/zeebe_errors.py

@@ -15,6 +15,10 @@ class ZeebeInternalError(PyZeebeError):
     pass
 
 
+class ZeebeDeadlineExceeded(PyZeebeError):
+    pass
+
+
 class UnkownGrpcStatusCodeError(PyZeebeError):
     def __init__(self, grpc_error: grpc.aio.AioRpcError):
         super().__init__()

pyzeebe/grpc_internals/zeebe_adapter_base.py

@@ -8,6 +8,7 @@
     ZeebeBackPressureError,
     ZeebeGatewayUnavailableError,
     ZeebeInternalError,
+    ZeebeDeadlineExceeded,
 )
 from pyzeebe.errors.pyzeebe_errors import PyZeebeError
 from pyzeebe.grpc_internals.grpc_utils import is_error_status
@@ -31,7 +32,7 @@ async def _handle_grpc_error(self, grpc_error: grpc.aio.AioRpcError):
         try:
             pyzeebe_error = _create_pyzeebe_error_from_grpc_error(grpc_error)
             raise pyzeebe_error
-        except (ZeebeGatewayUnavailableError, ZeebeInternalError):
+        except (ZeebeGatewayUnavailableError, ZeebeInternalError, ZeebeDeadlineExceeded):
             self._current_connection_retries += 1
             if not self._should_retry():
                 await self._close()
@@ -51,4 +52,6 @@ def _create_pyzeebe_error_from_grpc_error(grpc_error: grpc.aio.AioRpcError) -> P
         return ZeebeGatewayUnavailableError()
     if is_error_status(grpc_error, grpc.StatusCode.INTERNAL):
         return ZeebeInternalError()
+    elif is_error_status(grpc_error, grpc.StatusCode.DEADLINE_EXCEEDED):
+        return ZeebeDeadlineExceeded()
     return UnkownGrpcStatusCodeError(grpc_error)

pyzeebe/grpc_internals/zeebe_job_adapter.py

@@ -36,6 +36,7 @@ async def activate_jobs(
         request_timeout: int,
     ) -> AsyncGenerator[Job, None]:
         try:
+            grpc_request_timeout = request_timeout / 1000 * 2 if request_timeout > 0 else 20
             async for response in self._gateway_stub.ActivateJobs(
                 ActivateJobsRequest(
                     type=task_type,
@@ -44,7 +45,8 @@ async def activate_jobs(
                     maxJobsToActivate=max_jobs_to_activate,
                     fetchVariable=variables_to_fetch,
                     requestTimeout=request_timeout,
-                )
+                ),
+                timeout=grpc_request_timeout,
             ):
                 for raw_job in response.jobs:
                     job = self._create_job_from_raw_job(raw_job)

pyzeebe/worker/job_poller.py

@@ -6,6 +6,7 @@
     ZeebeBackPressureError,
     ZeebeGatewayUnavailableError,
     ZeebeInternalError,
+    ZeebeDeadlineExceeded,
 )
 from pyzeebe.grpc_internals.zeebe_job_adapter import ZeebeJobAdapter
 from pyzeebe.task.task import Task
@@ -65,7 +66,12 @@ async def poll_once(self):
         except ActivateJobsRequestInvalidError:
             logger.warning("Activate job requests was invalid for task %s", self.task.type)
             raise
-        except (ZeebeBackPressureError, ZeebeGatewayUnavailableError, ZeebeInternalError) as error:
+        except (
+            ZeebeBackPressureError,
+            ZeebeGatewayUnavailableError,
+            ZeebeInternalError,
+            ZeebeDeadlineExceeded,
+        ) as error:
             logger.warning(
                 "Failed to activate jobs from the gateway. Exception: %s. Retrying in 5 seconds...",
                 repr(error),