[Serve] Remove shutdown() method from TaskProcessorAdapter (ray-project#59713)

## Description
Fix @task_consumer decorator's __del__ method calling shutdown() which
broadcasts to all Celery workers instead of just the local one. This
kills newly started workers during rolling updates.

## Related issues
None

## Additional information
Removed self._adapter.shutdown() from __del__ - only stop_consumer()
should be called since it targets the specific worker hostname. Also
removed shutdown() implementation & from interface given it is not used
anywhere

---------

Signed-off-by: krisselberg <kselberg@princeton.edu>

Author: krisselberg

python/ray/serve/schema.py

@@ -1582,13 +1582,6 @@ def stop_consumer(self, timeout: float = 10.0):
         """
         pass
 
-    @abstractmethod
-    def shutdown(self):
-        """
-        Shutdown the task processor and clean up resources.
-        """
-        pass
-
     @abstractmethod
     def cancel_task_sync(self, task_id: str):
         """

python/ray/serve/task_consumer.py

@@ -154,7 +154,6 @@ def initialize_callable(self, consumer_concurrency: int):
 
             def __del__(self):
                 self._adapter.stop_consumer()
-                self._adapter.shutdown()
 
                 if hasattr(target_cls, "__del__"):
                     target_cls.__del__(self)

python/ray/serve/task_processor.py

@@ -225,11 +225,6 @@ def stop_consumer(self, timeout: float = 10.0):
 
         self._worker_thread = None
 
-    def shutdown(self):
-        logger.info("Shutting down Celery worker...")
-        self._app.control.shutdown()
-        logger.info("Celery worker shutdown complete...")
-
     def cancel_task_sync(self, task_id):
         """
         Cancels a task synchronously. Only supported for Redis and RabbitMQ brokers by Celery.

python/ray/serve/tests/unit/test_task_consumer.py

@@ -20,7 +20,6 @@ class MockTaskProcessorAdapter(TaskProcessorAdapter):
 
     _start_consumer_received: bool = False
     _stop_consumer_received: bool = False
-    _shutdown_received: bool = False
 
     def __init__(self, config: TaskProcessorConfig):
         self._config = config
@@ -46,9 +45,6 @@ def start_consumer(self, **kwargs):
     def stop_consumer(self, timeout: float = 10.0):
         self._stop_consumer_received = True
 
-    def shutdown(self):
-        self._shutdown_received = True
-
     def cancel_task_sync(self, task_id) -> bool:
         pass