@Override
  public void fetchSegmentToLocal(URI downloadURI, File dest)
      throws Exception {
    // Create a RoundRobinURIProvider to round robin IP addresses when retry uploading. Otherwise may always try to
    // download from a same broken host as: 1) DNS may not RR the IP addresses 2) OS cache the DNS resolution result.
    RoundRobinURIProvider uriProvider = new RoundRobinURIProvider(List.of(downloadURI), true);

    int retryCount = getRetryCount(uriProvider);

    _logger.info("Retry downloading for {} times. retryCount from pinot server config: {}, number of IP addresses for "
        + "download URI: {}", retryCount, _retryCount, uriProvider.numAddresses());
    RetryPolicies.exponentialBackoffRetryPolicy(retryCount, _retryWaitMs, _retryDelayScaleFactor).attempt(() -> {
      URI uri = uriProvider.next();
      try {
        String hostName = downloadURI.getHost();
        int port = downloadURI.getPort();
        // If the original download address is specified as host name, need add a "HOST" HTTP header to the HTTP
        // request. Otherwise, if the download address is a LB address, when the LB be configured as "disallow direct
        // access by IP address", downloading will fail.
        List<Header> httpHeaders = new LinkedList<>();
        if (!InetAddresses.isInetAddress(hostName)) {
          httpHeaders.add(new BasicHeader(HttpHeaders.HOST, hostName + ":" + port));
        }
        int statusCode = _httpClient.downloadFile(uri, dest, _authProvider, httpHeaders);
        _logger.info("Downloaded segment from: {} to: {} of size: {}; Response status code: {}", uri, dest,
            dest.length(), statusCode);
        return true;
      } catch (HttpErrorStatusException e) {
        int statusCode = e.getStatusCode();
        if (statusCode == HttpStatus.SC_NOT_FOUND || statusCode >= 500) {
          // Temporary exception
          // 404 is treated as a temporary exception, as the downloadURI may be backed by multiple hosts,
          // if singe host is down, can retry with another host.
          _logger.warn("Got temporary error status code: {} while downloading segment from: {} to: {}", statusCode, uri,
              dest, e);
          return false;
        } else {
          // Permanent exception
          _logger.error("Got permanent error status code: {} while downloading segment from: {} to: {}, won't retry",
              statusCode, uri, dest, e);
          throw e;
        }
      } catch (Exception e) {
        _logger.warn("Caught exception while downloading segment from: {} to: {}", uri, dest, e);
        return false;
      }
    });
  }
@Test
  public void testFetchSegmentToLocalSuccessAfterFirstTwoAttemptsFoundNoPeerServers()
      throws Exception {
    FileUploadDownloadClient client = mock(FileUploadDownloadClient.class);
    // The download always succeeds
    when(client.downloadFile(any(), any(), any())).thenReturn(200);
    HttpSegmentFetcher segmentFetcher = getSegmentFetcher(client);
    List<URI> uris = List.of(new URI("http://h1:8080"), new URI("http://h2:8080"));
    // The first two attempts found NO peers hosting the segment, and the last one found two servers
    //noinspection unchecked
    Supplier<List<URI>> uriSupplier = mock(Supplier.class);
    when(uriSupplier.get()).thenReturn(List.of()).thenReturn(List.of()).thenReturn(uris);
    segmentFetcher.fetchSegmentToLocal(SEGMENT_NAME, uriSupplier, SEGMENT_FILE);
  }