Extend HTTPClient Builder to allow setting a proxy server

core/src/main/java/org/apache/iceberg/rest/HTTPClient.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import org.apache.hc.client5.http.auth.CredentialsProvider;
 import org.apache.hc.client5.http.classic.methods.HttpUriRequest;
 import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
 import org.apache.hc.client5.http.config.ConnectionConfig;
@@ -41,6 +42,7 @@
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpHeaders;
+import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.HttpRequestInterceptor;
 import org.apache.hc.core5.http.HttpStatus;
 import org.apache.hc.core5.http.Method;
@@ -92,6 +94,8 @@ public class HTTPClient implements RESTClient {
 
   private HTTPClient(
       String uri,
+      HttpHost proxy,
+      CredentialsProvider proxyCredsProvider,
       Map<String, String> baseHeaders,
       ObjectMapper objectMapper,
       HttpRequestInterceptor requestInterceptor,
@@ -118,6 +122,14 @@ private HTTPClient(
     int maxRetries = PropertyUtil.propertyAsInt(properties, REST_MAX_RETRIES, 5);
     clientBuilder.setRetryStrategy(new ExponentialHttpRequestRetryStrategy(maxRetries));
 
+    if (proxy != null) {
+      if (proxyCredsProvider != null) {
+        clientBuilder.setDefaultCredentialsProvider(proxyCredsProvider);
+      }
+
+      clientBuilder.setProxy(proxy);
+    }
+
     this.httpClient = clientBuilder.build();
   }
 
@@ -496,6 +508,8 @@ public static class Builder {
     private final Map<String, String> baseHeaders = Maps.newHashMap();
     private String uri;
     private ObjectMapper mapper = RESTObjectMapper.mapper();
+    private HttpHost proxy;
+    private CredentialsProvider proxyCredentialsProvider;
 
     private Builder(Map<String, String> properties) {
       this.properties = properties;
@@ -507,6 +521,19 @@ public Builder uri(String path) {
       return this;
     }
 
+    public Builder withProxy(String hostname, int port) {
+      Preconditions.checkNotNull(hostname, "Invalid hostname for http client proxy: null");
+      this.proxy = new HttpHost(hostname, port);
+      return this;
+    }
+
+    public Builder withProxyCredentialsProvider(CredentialsProvider credentialsProvider) {
+      Preconditions.checkNotNull(
+          credentialsProvider, "Invalid credentials provider for http client proxy: null");
+      this.proxyCredentialsProvider = credentialsProvider;
+      return this;
+    }
+
     public Builder withHeader(String key, String value) {
       baseHeaders.put(key, value);
       return this;
@@ -532,8 +559,15 @@ public HTTPClient build() {
         interceptor = loadInterceptorDynamically(SIGV4_REQUEST_INTERCEPTOR_IMPL, properties);
       }
 
+      if (this.proxyCredentialsProvider != null) {
+        Preconditions.checkNotNull(
+            proxy, "Invalid http client proxy for proxy credentials provider: null");
+      }
+
       return new HTTPClient(
           uri,
+          proxy,
+          proxyCredentialsProvider,
           baseHeaders,
           mapper,
           interceptor,

core/src/test/java/org/apache/iceberg/rest/TestHTTPClient.java

@@ -37,10 +37,15 @@
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import org.apache.hc.client5.http.auth.AuthScope;
+import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
 import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
 import org.apache.hc.core5.http.EntityDetails;
 import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.HttpRequestInterceptor;
+import org.apache.hc.core5.http.HttpStatus;
 import org.apache.hc.core5.http.protocol.HttpContext;
 import org.apache.iceberg.IcebergBuild;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -52,9 +57,11 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+import org.mockserver.configuration.Configuration;
 import org.mockserver.integration.ClientAndServer;
 import org.mockserver.model.HttpRequest;
 import org.mockserver.model.HttpResponse;
+import org.mockserver.verify.VerificationTimes;
 
 /**
  * * Exercises the RESTClient interface, specifically over a mocked-server using the actual
@@ -126,6 +133,95 @@ public void testHeadFailure() throws JsonProcessingException {
     testHttpMethodOnFailure(HttpMethod.HEAD);
   }
 
+  @Test
+  public void testProxyServer() throws IOException {
+    int proxyPort = 1070;
+    try (ClientAndServer proxyServer = startClientAndServer(proxyPort);
+        RESTClient clientWithProxy =
+            HTTPClient.builder(ImmutableMap.of())
+                .uri(URI)
+                .withProxy("localhost", proxyPort)
+                .build()) {
+      String path = "v1/config";
+      HttpRequest mockRequest =
+          request("/" + path).withMethod(HttpMethod.HEAD.name().toUpperCase(Locale.ROOT));
+      HttpResponse mockResponse = response().withStatusCode(200);
+      proxyServer.when(mockRequest).respond(mockResponse);
+      clientWithProxy.head(path, ImmutableMap.of(), (onError) -> {});
+      proxyServer.verify(mockRequest, VerificationTimes.exactly(1));
+    }
+  }
+
+  @Test
+  public void testProxyCredentialProviderWithoutProxyServer() {
+    Assertions.assertThatThrownBy(
+            () ->
+                HTTPClient.builder(ImmutableMap.of())
+                    .uri(URI)
+                    .withProxyCredentialsProvider(new BasicCredentialsProvider())
+                    .build())
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("Invalid http client proxy for proxy credentials provider: null");
+  }
+
+  @Test
+  public void testProxyServerWithNullHostname() {
+    Assertions.assertThatThrownBy(
+            () -> HTTPClient.builder(ImmutableMap.of()).uri(URI).withProxy(null, 1070).build())
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("Invalid hostname for http client proxy: null");
+  }
+
+  @Test
+  public void testProxyAuthenticationFailure() throws IOException {
+    int proxyPort = 1050;
+    String proxyHostName = "localhost";
+    String authorizedUsername = "test-username";
+    String authorizedPassword = "test-password";
+    String invalidPassword = "invalid-password";
+
+    HttpHost proxy = new HttpHost(proxyHostName, proxyPort);
+    BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+    credentialsProvider.setCredentials(
+        new AuthScope(proxy),
+        new UsernamePasswordCredentials(authorizedUsername, invalidPassword.toCharArray()));
+
+    try (ClientAndServer proxyServer =
+            startClientAndServer(
+                new Configuration()
+                    .proxyAuthenticationUsername(authorizedUsername)
+                    .proxyAuthenticationPassword(authorizedPassword),
+                proxyPort);
+        RESTClient clientWithProxy =
+            HTTPClient.builder(ImmutableMap.of())
+                .uri(URI)
+                .withProxy(proxyHostName, proxyPort)
+                .withProxyCredentialsProvider(credentialsProvider)
+                .build()) {
+
+      ErrorHandler onError =
+          new ErrorHandler() {
+            @Override
+            public ErrorResponse parseResponse(int code, String responseBody) {
+              return null;
+            }
+
+            @Override
+            public void accept(ErrorResponse errorResponse) {
+              throw new RuntimeException(errorResponse.message() + " - " + errorResponse.code());
+            }
+          };
+
+      Assertions.assertThatThrownBy(
+              () -> clientWithProxy.get("v1/config", Item.class, ImmutableMap.of(), onError))
+          .isInstanceOf(RuntimeException.class)
+          .hasMessage(
+              String.format(
+                  "%s - %s",
+                  "Proxy Authentication Required", HttpStatus.SC_PROXY_AUTHENTICATION_REQUIRED));
+    }
+  }
+
   @Test
   public void testDynamicHttpRequestInterceptorLoading() {
     Map<String, String> properties = ImmutableMap.of("key", "val");