add options to client argument and e2e test

Signed-off-by: Filinto Duran <1373693+filintod@users.noreply.github.com>

Author: filintod

durabletask/aio/client.py

@@ -27,7 +27,9 @@ def __init__(self, *,
                  log_handler: Optional[logging.Handler] = None,
                  log_formatter: Optional[logging.Formatter] = None,
                  secure_channel: bool = False,
-                 interceptors: Optional[Sequence[ClientInterceptor]] = None):
+                 interceptors: Optional[Sequence[ClientInterceptor]] = None,
+                 channel_options: Optional[Sequence[tuple[str, Any]]] = None,
+                 ):
 
         if interceptors is not None:
             interceptors = list(interceptors)
@@ -41,7 +43,8 @@ def __init__(self, *,
         channel = get_grpc_aio_channel(
             host_address=host_address,
             secure_channel=secure_channel,
-            interceptors=interceptors
+            interceptors=interceptors,
+            options=channel_options,
         )
         self._channel = channel
         self._stub = stubs.TaskHubSidecarServiceStub(channel)

durabletask/client.py

@@ -98,7 +98,8 @@ def __init__(self, *,
                  log_handler: Optional[logging.Handler] = None,
                  log_formatter: Optional[logging.Formatter] = None,
                  secure_channel: bool = False,
-                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None):
+                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None,
+                 channel_options: Optional[Sequence[tuple[str, Any]]] = None):
 
         # If the caller provided metadata, we need to create a new interceptor for it and
         # add it to the list of interceptors.
@@ -114,7 +115,8 @@ def __init__(self, *,
         channel = shared.get_grpc_channel(
             host_address=host_address,
             secure_channel=secure_channel,
-            interceptors=interceptors
+            interceptors=interceptors,
+            options=channel_options,
         )
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger("client", log_handler, log_formatter)

tests/durabletask/test_orchestration_e2e.py

@@ -5,6 +5,7 @@
 import threading
 import time
 from datetime import timedelta
+import os
 
 import pytest
 
@@ -28,7 +29,12 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
         w.add_orchestrator(empty_orchestrator)
         w.start()
 
-        c = client.TaskHubGrpcClient()
+        # Enable retry policy via environment and set a custom max send length option
+        os.environ["DAPR_GRPC_RETRY_ENABLED"] = "true"
+        os.environ["DAPR_GRPC_RETRY_MAX_ATTEMPTS"] = "3"
+        c = client.TaskHubGrpcClient(channel_options=[
+            ("grpc.max_send_message_length", 1024 * 1024),  # 1MB
+        ])
         id = c.schedule_new_orchestration(empty_orchestrator)
         state = c.wait_for_orchestration_completion(id, timeout=30)