docstr: Initializes a thread-safe datapath over a Ray Client gRPC channel.
  
  Args:
      client_worker: The Ray Client worker that manages this client
      client_id: the generated ID representing this client
      metadata: metadata to pass to gRPC requests
code: def __init__(
      self, client_worker: "Worker", client_id: str, metadata: list
  ):
      """Initializes a thread-safe datapath over a Ray Client gRPC channel.
  
      Args:
          client_worker: The Ray Client worker that manages this client
          client_id: the generated ID representing this client
          metadata: metadata to pass to gRPC requests
      """
      self.client_worker = client_worker
      self._client_id = client_id
      self._metadata = metadata
      self.data_thread = self._start_datathread()
  
      # Track outstanding requests to resend in case of disconnection
      self.outstanding_requests: Dict[int, Any] = OrderedDict()
  
      # Serialize access to all mutable internal states: self.request_queue,
      # self.ready_data, self.asyncio_waiting_data,
      # self._in_shutdown, self._req_id, self.outstanding_requests and
      # calling self._next_id()
      self.lock = threading.Lock()
  
      # Waiting for response or shutdown.
      self.cv = threading.Condition(lock=self.lock)
  
      self.request_queue = self._create_queue()
      self.ready_data: Dict[int, Any] = {}
      # NOTE: Dictionary insertion is guaranteed to complete before lookup
      # and/or removal because of synchronization via the request_queue.
      self.asyncio_waiting_data: Dict[int, ResponseCallable] = {}
      self._in_shutdown = False
      self._req_id = 0
      self._last_exception = None
      self._acknowledge_counter = 0
  
      self.data_thread.start()
  
imports: import math
  import logging
  import queue
  import threading
  import warnings
  import grpc
  from collections import OrderedDict
  from typing import Any, Callable, Dict, TYPE_CHECKING, Optional, Union
  import ray.core.generated.ray_client_pb2
  import ray.core.generated.ray_client_pb2_grpc
  from ray.util.client.common import INT32_MAX, OBJECT_TRANSFER_CHUNK_SIZE, OBJECT_TRANSFER_WARNING_SIZE
  from ray.util.debug import log_once
label_desc: Use image node_id 1 to create a new DataClient object  with example: obj = DataClient(client_worker, client_id, metadata)