code: def test_batching_client_dropped_streaming(serve_instance):
      """Test streaming batching with clients that drops the connection.
  
      After requests are dropped. The next request should succeed.
      """
  
      @serve.deployment
      class ModelStreaming:
          @serve.batch(max_batch_size=3)
          async def handle_batch(self, requests):
              await asyncio.sleep(0.05)
              for i in range(10):
                  yield [str(i) for _ in range(len(requests))]
  
          async def __call__(self, request):
              return StreamingResponse(self.handle_batch(request))
  
      serve.run(ModelStreaming.bind())
  
      url = "http://localhost:8000/"
  
      # Sending requests with clients that drops the connection.
      for _ in range(3):
          with pytest.raises(
              (
                  requests.exceptions.ReadTimeout,
                  requests.exceptions.ConnectionError,
              )
          ):
              requests.get(url, timeout=0.005)
  
      # The following request should succeed.
      resp = requests.get(url, timeout=1)
      assert resp.status_code == 200
      assert resp.text == "0123456789"
  
file_path: ray/python/ray/serve/tests/test_batching.py
imports: import asyncio
  from concurrent.futures.thread import ThreadPoolExecutor
  from functools import partial
  from typing import List
  import pytest
  import requests
  from starlette.responses import StreamingResponse
  import ray
  from ray import serve
label_desc: Use image node_id 4 for calling a global function with example usage: test_batching_client_dropped_streaming(serve_instance) and returns: StreamingResponse