@Override
  public void getClient(Request request, RequestContext requestContext, Callback<TransportClient> clientCallback)
  {
    URI uri = request.getURI();
    debug(_log, "get client for uri: ", uri);

    if (!D2_SCHEME_NAME.equalsIgnoreCase(uri.getScheme()))
    {
      throw new IllegalArgumentException("Unsupported scheme in URI " + uri);
    }

    // get the service for this uri
    String extractedServiceName = LoadBalancerUtil.getServiceNameFromUri(uri);

    listenToServiceAndCluster(extractedServiceName, Callbacks.handle(service -> {
      String serviceName = service.getServiceName();
      String clusterName = service.getClusterName();
      try
      {
        ClusterProperties cluster = getClusterProperties(serviceName, clusterName);

        // Check if we want to override the service URL and bypass choosing among the existing
        // tracker clients. This is useful when the service we want is not announcing itself to
        // the cluster, ie a private service for a set of clients. This mechanism is deprecated;
        // use host override list instead.
        @SuppressWarnings("deprecation")
        URI targetService = LoadBalancerUtil.TargetHints.getRequestContextTargetService(requestContext);

        // Checks if we have a host override list provided in the request context. If present,
        // get the override URI available override for the current cluster and service names.
        HostOverrideList overrides = (HostOverrideList) requestContext.getLocalAttr(HOST_OVERRIDE_LIST);
        URI override = overrides == null ? null : overrides.getOverride(clusterName, serviceName);

        if (targetService == null && override == null)
        {
          LoadBalancerStateItem<UriProperties> uriItem = getUriItem(serviceName, clusterName, cluster);
          UriProperties uris = uriItem.getProperty();

          List<LoadBalancerState.SchemeStrategyPair> orderedStrategies =
              _state.getStrategiesForService(serviceName, service.getPrioritizedSchemes());

          TrackerClient trackerClient = null;

          // Use client provided by CustomURIAffinityRoutingProvider when it's enabled
          CustomAffinityRoutingURIProvider customAffinityRoutingURIProvider =
              (CustomAffinityRoutingURIProvider) requestContext.getLocalAttr(CustomAffinityRoutingURIProvider.CUSTOM_AFFINITY_ROUTING_URI_PROVIDER);

          boolean enableCustomAffinityRouting = isCustomAffinityRoutingEnabled(requestContext, customAffinityRoutingURIProvider);
          if (enableCustomAffinityRouting)
          {
            trackerClient = customAffinityRoutingURIProvider.getTargetHostURI(clusterName)
                  .map(targetHost -> _state.getClient(serviceName, targetHost))
                  .orElse(null);
          }

          if (trackerClient == null)
          {
            trackerClient =
                chooseTrackerClient(request, requestContext, serviceName, clusterName, cluster, uriItem, uris,
                    orderedStrategies, service);

            // set host URI for the cluster. with that next time, for the same inbound request, if downstream request is
            // made to same cluster and custom affinity routing is enabled, then it will go to same box.
            if (enableCustomAffinityRouting)
            {
              customAffinityRoutingURIProvider.setTargetHostURI(clusterName, trackerClient.getUri());
            }
          }

          String clusterAndServiceUriString = trackerClient.getUri() + service.getPath();
          _serviceAvailableStats.inc();
          clientCallback.onSuccess(new RewriteLoadBalancerClient(serviceName,
            URI.create(clusterAndServiceUriString),
            trackerClient));
        }
        else
        {
          URI target = override == null ? targetService : URI.create(override + service.getPath());
          if (targetService != null && override != null)
          {
            _log.warn("Both TargetHints and HostOverrideList are found. HostOverList will take precedence %s.", target);
          }

          if (_log.isDebugEnabled())
          {
            _log.debug("Rewrite URI as specified in the TargetHints/HostOverrideList {} for cluster {} and service {}.",
                target, clusterName, serviceName);
          }
          TransportClient transportClient = _state.getClient(serviceName, target.getScheme());
          if (transportClient == null)
          {
            throw new ServiceUnavailableException(serviceName, String.format(
                    "PEGA_1001. Cannot find transportClient for service %s and scheme %s with URI specified in"
                        + "TargetHints/HostOverrideList %s", serviceName, target.getScheme(), target));
          }

          clientCallback.onSuccess(new RewriteLoadBalancerClient(serviceName, target, transportClient));
        }
      }
      catch (ServiceUnavailableException e)
      {
        clientCallback.onError(e);
      }
    }, clientCallback));
  }
@Test(dataProvider = "customAffinityRoutingSkippedDataProvider")
  public void testCustomAffinityRoutingSkipped(boolean enableTargetHostHint, boolean enableCustomAffinityRouting,
      URI expectedURI) throws Exception
  {
    MockStore<ServiceProperties> serviceRegistry = new MockStore<>();
    MockStore<ClusterProperties> clusterRegistry = new MockStore<>();
    MockStore<UriProperties> uriRegistry = new MockStore<>();
    List<String> prioritizedSchemes = new ArrayList<>();

    SimpleLoadBalancer loadBalancer = setupLoadBalancer(serviceRegistry, clusterRegistry, uriRegistry);

    URI uri1 = URI.create("http://test.qd.com:1234");

    Map<Integer, PartitionData> partitionData = new HashMap<>(1);
    partitionData.put(DEFAULT_PARTITION_ID, new PartitionData(1d));
    Map<URI, Map<Integer, PartitionData>> uriData = new HashMap<>(2);

    uriData.put(uri1, partitionData);

    URI uri2 = URI.create("http://targethosthint.qd.com:1234");
    RequestContext serviceContext = new RequestContext();
    if (enableTargetHostHint) {
      uriData.put(uri2, partitionData);
      KeyMapper.TargetHostHints.setRequestContextTargetHost(serviceContext, uri2);
    }

    prioritizedSchemes.add(PropertyKeys.HTTP_SCHEME);

    Set<URI> bannedSet = new HashSet<>();
    clusterRegistry.put(CLUSTER1_NAME, new ClusterProperties(CLUSTER1_NAME, Collections.emptyList(),
        Collections.emptyMap(), bannedSet, NullPartitionProperties.getInstance()));

    serviceRegistry.put("foo", new ServiceProperties("foo",
        CLUSTER1_NAME,
        "/foo", Collections.singletonList("degrader"),
        Collections.<String,Object>emptyMap(),
        null,
        null,
        prioritizedSchemes,
        null));
    uriRegistry.put(CLUSTER1_NAME, new UriProperties(CLUSTER1_NAME, uriData));

    URIRequest uriRequest = new URIRequest("d2://foo/52");

    serviceContext.putLocalAttr(
        CustomAffinityRoutingURIProvider.CUSTOM_AFFINITY_ROUTING_URI_PROVIDER, new CustomAffinityRoutingURIProvider() {
      private final Map<String, URI> uriMap = new HashMap<>();

      @Override
      public boolean isEnabled() {
        return enableCustomAffinityRouting;
      }

      @Override
      public Optional<URI> getTargetHostURI(String clusterName) {
        return Optional.ofNullable(uriMap.get(clusterName));
      }

      @Override
      public void setTargetHostURI(String clusterName, URI targetHostURI) {
        uriMap.put(clusterName, targetHostURI);
      }
    });

    RewriteLoadBalancerClient client =
        (RewriteLoadBalancerClient) loadBalancer.getClient(uriRequest, serviceContext);
    Assert.assertEquals(client.getUri(), expectedURI);
    CustomAffinityRoutingURIProvider affinityRoutingURIProvider
        = (CustomAffinityRoutingURIProvider) serviceContext.getLocalAttr(CustomAffinityRoutingURIProvider.CUSTOM_AFFINITY_ROUTING_URI_PROVIDER);
    Assert.assertFalse(affinityRoutingURIProvider.getTargetHostURI(CLUSTER1_NAME).isPresent());
  }