exception for error message

Signed-off-by: dentiny <[email protected]>

python/ray/_private/runtime_env/agent/runtime_env_agent.py

@@ -117,7 +117,7 @@ def _increase_reference_for_runtime_env(self, serialized_env: str):
         self._runtime_env_reference[serialized_env] += 1
 
     def _decrease_reference_for_runtime_env(self, serialized_env: str):
-        """Decrease reference count for the given [serialized_env]. Return whether decrement succeeds or not."""
+        """Decrease reference count for the given [serialized_env]. Throw exception if we cannot decrement reference."""
         default_logger.debug(f"Decrease reference for runtime env {serialized_env}.")
         unused = False
         if self._runtime_env_reference[serialized_env] > 0:
@@ -127,11 +127,12 @@ def _decrease_reference_for_runtime_env(self, serialized_env: str):
                 del self._runtime_env_reference[serialized_env]
         else:
             default_logger.warning(f"Runtime env {serialized_env} does not exist.")
-            return False
+            raise ValueError(
+                f"{serialized_env} cannot decrement referenced since the reference count is 0"
+            )
         if unused:
             default_logger.info(f"Unused runtime env {serialized_env}.")
             self._unused_runtime_env_callback(serialized_env)
-        return True
 
     def increase_reference(
         self, runtime_env: RuntimeEnv, serialized_env: str, source_process: str
@@ -144,15 +145,13 @@ def increase_reference(
 
     def decrease_reference(
         self, runtime_env: RuntimeEnv, serialized_env: str, source_process: str
-    ) -> bool:
-        """Decrease reference count for runtime env and uri. Return whether decrement succeeds or not."""
+    ):
+        """Decrease reference count for runtime env and uri. Throw exception if decrement reference count fails."""
         if source_process in self._reference_exclude_sources:
-            return True
-        if not self._decrease_reference_for_runtime_env(serialized_env):
-            return False
+            return
+        self._decrease_reference_for_runtime_env(serialized_env)
         uris = self._uris_parser(runtime_env)
         self._decrease_reference_for_uris(uris)
-        return True
 
     @property
     def runtime_env_refs(self) -> Dict[str, int]:
@@ -548,13 +547,14 @@ async def DeleteRuntimeEnvIfPossible(self, request):
                 ),
             )
 
-        deref_succ = self._reference_table.decrease_reference(
-            runtime_env, request.serialized_runtime_env, request.source_process
-        )
-        if not deref_succ:
+        try:
+            self._reference_table.decrease_reference(
+                runtime_env, request.serialized_runtime_env, request.source_process
+            )
+        except ValueError as e:
             return runtime_env_agent_pb2.DeleteRuntimeEnvIfPossibleReply(
                 status=agent_manager_pb2.AGENT_RPC_STATUS_FAILED,
-                error_message=f"Fails to dereference for runtime env {request.serialized_runtime_env}",
+                error_message=f"Fails to decrement reference for runtime env for {str(e)}",
             )
 
         return runtime_env_agent_pb2.DeleteRuntimeEnvIfPossibleReply(