diff --git a/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/datajobs_service_properties.py b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/datajobs_service_properties.py
index 25d2ab0098..f7316115d3 100644
--- a/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/datajobs_service_properties.py
+++ b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/datajobs_service_properties.py
@@ -10,9 +10,9 @@
 from vdk.api.job_input import IProperties
 from vdk.api.plugin.plugin_input import IPropertiesServiceClient
 
-from ...core.errors import report_and_throw
-from ...core.errors import UserCodeError
+from ...core.errors import ResolvableBy
 from .base_properties_impl import check_valid_property
+from .exception import WritePreProcessPropertiesException
 
 log = logging.getLogger(__name__)
 
@@ -82,15 +82,11 @@ def set_all_properties(self, properties: Dict[str, PropertyValue]) -> None:
                         self._job_name, self._team_name, properties
                     )
                 except Exception as e:
-                    report_and_throw(
-                        UserCodeError(
-                            f"A write pre-processor of properties client {client} had failed.",
-                            f"User Error occurred. Exception was: {e}",
-                            "PROPERTIES_WRITE_PREPROCESS_SEQUENCE was interrupted, and "
-                            "properties won't be written by the PROPERTIES_DEFAULT_TYPE client.",
-                            "Handle the exception raised.",
-                        )
-                    )
+                    raise WritePreProcessPropertiesException(
+                        client=client,
+                        preprocess_sequence=str(self._write_preprocessors),
+                        resolvable_by=ResolvableBy.USER_ERROR,
+                    ) from e
 
         for k, v in list(properties.items()):
             check_valid_property(
diff --git a/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/exception.py b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/exception.py
new file mode 100644
index 0000000000..ebeffdb0c2
--- /dev/null
+++ b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_properties/exception.py
@@ -0,0 +1,36 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Optional
+
+from vdk.internal.core.errors import BaseVdkError
+from vdk.internal.core.errors import ResolvableBy
+
+
+class PropertiesException(BaseVdkError):
+    """
+    Base Exception for all custom exceptions related to the properties.
+    """
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        resolvable_by: Optional[ResolvableBy] = None,
+    ):
+        super().__init__(None, resolvable_by, message)
+
+
+class WritePreProcessPropertiesException(PropertiesException):
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        client="unknown",
+        preprocess_sequence: str = "",
+        resolvable_by: Optional[ResolvableBy] = None,
+    ):
+        if not message:
+            message = (
+                f"Write pre-processor for client {client} failed. "
+                f"preprocess sequence  is {preprocess_sequence}. "
+                f"No properties are updated."
+            )
+        super().__init__(message, resolvable_by)
diff --git a/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/datajobs_service_secrets.py b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/datajobs_service_secrets.py
index 36205db410..64d360710c 100644
--- a/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/datajobs_service_secrets.py
+++ b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/datajobs_service_secrets.py
@@ -11,8 +11,10 @@
 from vdk.api.plugin.plugin_input import ISecretsServiceClient
 
 from ...core.errors import report_and_throw
+from ...core.errors import ResolvableBy
 from ...core.errors import UserCodeError
 from .base_secrets_impl import check_valid_secret
+from .exception import WritePreProcessSecretsException
 
 log = logging.getLogger(__name__)
 
@@ -80,15 +82,11 @@ def set_all_secrets(self, secrets: Dict[str, SecretValue]) -> None:
                         self._job_name, self._team_name, secrets
                     )
                 except Exception as e:
-                    report_and_throw(
-                        UserCodeError(
-                            f"A write pre-processor of secrets client {client} had failed.",
-                            f"User Error occurred. Exception was: {e}",
-                            "SECRETS_WRITE_PREPROCESS_SEQUENCE was interrupted, and "
-                            "secrets won't be written by the SECRETS_DEFAULT_TYPE client.",
-                            "Handle the exception raised.",
-                        )
-                    )
+                    raise WritePreProcessSecretsException(
+                        client=client,
+                        preprocess_sequence=str(self._write_preprocessors),
+                        resolvable_by=ResolvableBy.USER_ERROR,
+                    ) from e
 
         for k, v in list(secrets.items()):
             check_valid_secret(k, v, DataJobsServiceSecrets.__VALID_TYPES)  # throws
diff --git a/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/exception.py b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/exception.py
new file mode 100644
index 0000000000..51af1f850c
--- /dev/null
+++ b/projects/vdk-core/src/vdk/internal/builtin_plugins/job_secrets/exception.py
@@ -0,0 +1,36 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Optional
+
+from vdk.internal.core.errors import BaseVdkError
+from vdk.internal.core.errors import ResolvableBy
+
+
+class SecretsException(BaseVdkError):
+    """
+    Base Exception for all custom exceptions related to the properties.
+    """
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        resolvable_by: Optional[ResolvableBy] = None,
+    ):
+        super().__init__(None, resolvable_by, message)
+
+
+class WritePreProcessSecretsException(SecretsException):
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        client="unknown",
+        preprocess_sequence: str = "",
+        resolvable_by: Optional[ResolvableBy] = None,
+    ):
+        if not message:
+            message = (
+                f"Write pre-processor for client {client} failed. "
+                f"preprocess sequence  is {preprocess_sequence}. "
+                f"No properties are updated."
+            )
+        super().__init__(message, resolvable_by)
diff --git a/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_properties/test_properties_router.py b/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_properties/test_properties_router.py
index 27d0469b7e..af841b4f9b 100644
--- a/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_properties/test_properties_router.py
+++ b/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_properties/test_properties_router.py
@@ -5,10 +5,15 @@
 import pytest
 from vdk.api.plugin.plugin_input import IPropertiesServiceClient
 from vdk.internal.builtin_plugins.config.job_config import JobConfigKeys
+from vdk.internal.builtin_plugins.job_properties.exception import (
+    WritePreProcessPropertiesException,
+)
 from vdk.internal.builtin_plugins.job_properties.properties_router import (
     PropertiesRouter,
 )
+from vdk.internal.core import errors
 from vdk.internal.core.config import Configuration
+from vdk.internal.core.errors import ResolvableBy
 from vdk.internal.core.errors import UserCodeError
 from vdk.internal.core.errors import VdkConfigurationError
 
@@ -157,8 +162,12 @@ def test_preprocessing_sequence_error():
     router.set_properties_factory_method("foo", lambda: foo_mock_client)
     router.set_properties_factory_method("bar", lambda: bar_mock_client)
 
-    with pytest.raises(UserCodeError):
+    with pytest.raises(WritePreProcessPropertiesException) as exc_info:
         router.set_all_properties({"a": "b"})
+        assert (
+            errors.get_exception_resolvable_by(exc_info.value)
+            == ResolvableBy.USER_ERROR
+        )
 
 
 def test_preprocessing_sequence_misconfigured():
diff --git a/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_secrets/test_secrets_router.py b/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_secrets/test_secrets_router.py
index fd24d4b7ed..5c37869129 100644
--- a/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_secrets/test_secrets_router.py
+++ b/projects/vdk-core/tests/vdk/internal/builtin_plugins/job_secrets/test_secrets_router.py
@@ -5,10 +5,15 @@
 import pytest
 from vdk.api.plugin.plugin_input import ISecretsServiceClient
 from vdk.internal.builtin_plugins.config.job_config import JobConfigKeys
+from vdk.internal.builtin_plugins.job_secrets.exception import (
+    WritePreProcessSecretsException,
+)
 from vdk.internal.builtin_plugins.job_secrets.secrets_router import (
     SecretsRouter,
 )
+from vdk.internal.core import errors
 from vdk.internal.core.config import Configuration
+from vdk.internal.core.errors import ResolvableBy
 from vdk.internal.core.errors import UserCodeError
 from vdk.internal.core.errors import VdkConfigurationError
 
@@ -147,8 +152,12 @@ def test_preprocessing_sequence_error():
     router.set_secrets_factory_method("foo", lambda: foo_mock_client)
     router.set_secrets_factory_method("bar", lambda: bar_mock_client)
 
-    with pytest.raises(UserCodeError):
+    with pytest.raises(WritePreProcessSecretsException) as exc_info:
         router.set_all_secrets({"a": "b"})
+        assert (
+            errors.get_exception_resolvable_by(exc_info.value)
+            == ResolvableBy.USER_ERROR
+        )
 
 
 def test_preprocessing_sequence_misconfigured():