class_docstr: Class that handles the requests arriving to the head and the results extracted from the requests future.
  
  :param metrics_registry: optional metrics registry for prometheus. Used if we need to expose metrics from the executor or from the data request handler
  :param runtime_name: optional runtime_name that will be registered during monitoring
code: def __init__(
      self,
      args: "argparse.Namespace",
      logger: "JinaLogger",
      metrics_registry: Optional["CollectorRegistry"] = None,
      meter=None,
      runtime_name: Optional[str] = None,
      aio_tracing_client_interceptors=None,
      tracing_client_interceptor=None,
      **kwargs,
  ):
      if args.name is None:
          args.name = ""
      self.logger = logger
      self.args = args
      self.meter = meter
      self.metrics_registry = metrics_registry
      self.name = args.name
      self._deployment_name = os.getenv(
          "JINA_DEPLOYMENT_NAME", "worker"
      )
      self.aio_tracing_client_interceptors = (
          aio_tracing_client_interceptors
      )
      self.tracing_client_interceptor = tracing_client_interceptor
      self.connection_pool = GrpcConnectionPool(
          runtime_name=self.name,
          logger=self.logger,
          compression=args.compression,
          metrics_registry=self.metrics_registry,
          meter=self.meter,
          aio_tracing_client_interceptors=self.aio_tracing_client_interceptors,
          tracing_client_interceptor=self.tracing_client_interceptor,
          channel_options=self.args.grpc_channel_options,
      )
      self._retries = self.args.retries
  
      polling = getattr(args, "polling", self.DEFAULT_POLLING.name)
      try:
          # try loading the polling args as json
          endpoint_polling = json.loads(polling)
          # '*' is used a wildcard and will match all endpoints, except /index, /search and explicitly defined endpoins
          default_polling = (
              PollingType.from_string(endpoint_polling["*"])
              if "*" in endpoint_polling
              else self.DEFAULT_POLLING
          )
          self._polling = self._default_polling_dict(default_polling)
          for endpoint in endpoint_polling:
              self._polling[endpoint] = PollingType(
                  endpoint_polling[endpoint]
                  if type(endpoint_polling[endpoint]) == int
                  else PollingType.from_string(
                      endpoint_polling[endpoint]
                  )
              )
      except (ValueError, TypeError):
          # polling args is not a valid json, try interpreting as a polling enum type
          default_polling = (
              polling
              if type(polling) == PollingType
              else PollingType.from_string(polling)
          )
          self._polling = self._default_polling_dict(default_polling)
  
      if hasattr(args, "connection_list") and args.connection_list:
          connection_list = json.loads(args.connection_list)
          for shard_id in connection_list:
              shard_connections = connection_list[shard_id]
              if isinstance(shard_connections, str):
                  self.connection_pool.add_connection(
                      deployment=self._deployment_name,
                      address=shard_connections,
                      shard_id=int(shard_id),
                  )
              else:
                  for connection in shard_connections:
                      self.connection_pool.add_connection(
                          deployment=self._deployment_name,
                          address=connection,
                          shard_id=int(shard_id),
                      )
  
      self.uses_before_address = args.uses_before_address
      self.timeout_send = args.timeout_send
      if self.timeout_send:
          self.timeout_send /= 1e3  # convert ms to seconds
  
      if self.uses_before_address:
          self.connection_pool.add_connection(
              deployment="uses_before", address=self.uses_before_address
          )
      self.uses_after_address = args.uses_after_address
      if self.uses_after_address:
          self.connection_pool.add_connection(
              deployment="uses_after", address=self.uses_after_address
          )
      self._reduce = not args.no_reduce
      super().__init__(metrics_registry, meter, runtime_name)
      self.logger = logger
      self._executor_endpoint_mapping = None
      self._gathering_endpoints = False
      self.runtime_name = runtime_name
      self._pydantic_models_by_endpoint = None
      self.endpoints_discovery_stop_event = asyncio.Event()
      self.endpoints_discovery_task = None
      if docarray_v2:
          self.endpoints_discovery_task = asyncio.create_task(
              self._get_endpoints_from_workers(
                  connection_pool=self.connection_pool,
                  name=self._deployment_name,
                  retries=self._retries,
                  stop_event=self.endpoints_discovery_stop_event,
              )
          )
  
imports: import argparse
  import asyncio
  import json
  import os
  from collections import defaultdict
  from typing import TYPE_CHECKING, AsyncIterator, Dict, List, Optional, Tuple, Any
  import grpc
  from jina.enums import PollingType
  from jina.constants import __default_endpoint__
  from jina.excepts import InternalNetworkError
  from jina.helper import get_full_version
  from jina.proto import jina_pb2
  from jina.serve.networking import GrpcConnectionPool
  from jina.serve.runtimes.monitoring import MonitoringRequestMixin
  from jina.serve.runtimes.worker.request_handling import WorkerRequestHandler
  from jina.types.request.data import DataRequest, Response
  from jina._docarray import docarray_v2
label_desc: Use image node_id 1 to create a new HeaderRequestHandler object from inherited base classes: MonitoringRequestMixin with example: obj = HeaderRequestHandler(args, logger, metrics_registry, meter, runtime_name, aio_tracing_client_interceptors, tracing_client_interceptor)