Repository: funcx-faas/funcx
Issue: MaxResultSizeExceeded should be defined in funcx_sdk and not funcx_endpoint

**Describe the bug**

Currently the `MaxResultSizeExceeded` exception is defined in `funcx_endpoint.executors.high_throughput.funcx_worker`.
This exception is resolved in the `funcx_sdk` and the user shouldn't have to install `funcx_endpoint` to resolve exceptions.
Secondly, this introduces a circular dependency between `funcx_sdk` and the `funcx_endpoint`

Note. Once this is resolved, please update  `~/.github/workflows/smoke_test.yaml` to not install `funcx_endpoint`

**Expected behavior**

Move the exception over the `funcx_sdk` such that you do not need the `funcx_endpoint` installed to run functions.

**Environment**
- All

**Distributed Environment**
- N/A

Changed files:
funcx_sdk/funcx/utils/errors.py, funcx_endpoint/funcx_endpoint/executors/high_throughput/funcx_worker.py, funcx_sdk/funcx/tests/test_result_size.py, smoke_tests/tests/test_s3_indirect.py

diff --git a/funcx_endpoint/funcx_endpoint/executors/high_throughput/funcx_worker.py b/funcx_endpoint/funcx_endpoint/executors/high_throughput/funcx_worker.py
index fd2d00e2..8a483907 100644
--- a/funcx_endpoint/funcx_endpoint/executors/high_throughput/funcx_worker.py
+++ b/funcx_endpoint/funcx_endpoint/executors/high_throughput/funcx_worker.py
@@ -14,28 +14,31 @@ from funcx.serialize import FuncXSerializer
 from funcx_endpoint.executors.high_throughput.messages import Message
 from funcx_endpoint.logging_config import setup_logging
 
+try:
+    from funcx.utils.errors import MaxResultSizeExceeded
+except ImportError:
+    # to-do: Remove this after funcx,funcx-endpoint==0.3.5 is released
+    class MaxResultSizeExceeded(Exception):
+        """Result produced by the function exceeds the maximum supported result size
+        threshold"""
+
+        def __init__(self, result_size: int, result_size_limit: int):
+            self.result_size = result_size
+            self.result_size_limit = result_size_limit
+
+        def __str__(self) -> str:
+            return (
+                f"Task result of {self.result_size}B exceeded current "
+                f"limit of {self.result_size_limit}B"
+            )
+
+
 log = logging.getLogger(__name__)
 
 DEFAULT_RESULT_SIZE_LIMIT_MB = 10
 DEFAULT_RESULT_SIZE_LIMIT_B = DEFAULT_RESULT_SIZE_LIMIT_MB * 1024 * 1024
 
 
-class MaxResultSizeExceeded(Exception):
-    """
-    Result produced by the function exceeds the maximum supported result size
-    threshold"""
-
-    def __init__(self, result_size, result_size_limit):
-        self.result_size = result_size
-        self.result_size_limit = result_size_limit
-
-    def __str__(self):
-        return (
-            f"Task result of {self.result_size}B exceeded current "
-            f"limit of {self.result_size_limit}B"
-        )
-
-
 class FuncXWorker:
     """The FuncX worker
     Parameters
diff --git a/funcx_sdk/funcx/tests/test_result_size.py b/funcx_sdk/funcx/tests/test_result_size.py
index 40a9fbbb..5bdd84ab 100644
--- a/funcx_sdk/funcx/tests/test_result_size.py
+++ b/funcx_sdk/funcx/tests/test_result_size.py
@@ -1,7 +1,6 @@
 import pytest
 
-from funcx.utils.errors import TaskPending
-from funcx_endpoint.executors.high_throughput.funcx_worker import MaxResultSizeExceeded
+from funcx.utils.errors import MaxResultSizeExceeded, TaskPending
 
 
 def large_result_producer(size) -> str:
diff --git a/funcx_sdk/funcx/utils/errors.py b/funcx_sdk/funcx/utils/errors.py
index 267a0c9d..e991d090 100644
--- a/funcx_sdk/funcx/utils/errors.py
+++ b/funcx_sdk/funcx/utils/errors.py
@@ -104,3 +104,18 @@ class TaskPending(FuncxError):
 
     def __repr__(self):
         return f"Task is pending due to {self.reason}"
+
+
+class MaxResultSizeExceeded(Exception):
+    """Result produced by the function exceeds the maximum supported result size
+    threshold"""
+
+    def __init__(self, result_size: int, result_size_limit: int):
+        self.result_size = result_size
+        self.result_size_limit = result_size_limit
+
+    def __str__(self) -> str:
+        return (
+            f"Task result of {self.result_size}B exceeded current "
+            f"limit of {self.result_size_limit}B"
+        )
diff --git a/smoke_tests/tests/test_s3_indirect.py b/smoke_tests/tests/test_s3_indirect.py
index 8b73401d..fd94d88c 100644
--- a/smoke_tests/tests/test_s3_indirect.py
+++ b/smoke_tests/tests/test_s3_indirect.py
@@ -1,6 +1,6 @@
 import pytest
 
-from funcx_endpoint.executors.high_throughput.funcx_worker import MaxResultSizeExceeded
+from funcx.utils.errors import MaxResultSizeExceeded
 
 
 def large_result_producer(size: int) -> str: