HTTP client proxy

Author: castorm

README.md

@@ -274,6 +274,26 @@ Uses a [OkHttp](https://square.github.io/okhttp/) client.
 > Time to live for the connection
 > *   Type: `Long`
 > *   Default: `300000`
+> 
+> ##### `http.client.proxy.host`
+> Hostname of the HTTP Proxy
+> *   Type: `String`
+> *   Default: ``
+> 
+> ##### `http.client.proxy.port`
+> Port of the HTTP Proxy
+> *   Type: `Integer`
+> *   Default: `3128`
+> 
+> ##### `http.client.proxy.username`
+> Username of the HTTP Proxy
+> *   Type: `String`
+> *   Default: ``
+> 
+> ##### `http.client.proxy.password`
+> Password of the HTTP Proxy
+> *   Type: `String`
+> *   Default: ``
 ---
 <a name="auth"/>
 

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/client/okhttp/OkHttpClient.java

@@ -26,6 +26,7 @@
 import com.github.castorm.kafka.connect.http.model.HttpResponse;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import okhttp3.Authenticator;
 import okhttp3.Call;
 import okhttp3.ConnectionPool;
 import okhttp3.HttpUrl;
@@ -35,18 +36,24 @@
 import okhttp3.logging.HttpLoggingInterceptor;
 
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Proxy;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
+import static java.net.Proxy.NO_PROXY;
+import static java.net.Proxy.Type.HTTP;
 import static java.util.Optional.empty;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static javax.ws.rs.core.HttpHeaders.AUTHORIZATION;
+import static okhttp3.Credentials.basic;
 import static okhttp3.HttpUrl.parse;
 import static okhttp3.RequestBody.create;
 import static okhttp3.logging.HttpLoggingInterceptor.Level.BASIC;
 import static okhttp3.logging.HttpLoggingInterceptor.Level.BODY;
 import static okhttp3.logging.HttpLoggingInterceptor.Level.NONE;
+import static org.apache.commons.lang.StringUtils.isEmpty;
 
 @Slf4j
 public class OkHttpClient implements HttpClient {
@@ -69,6 +76,8 @@ public void configure(Map<String, ?> configs) {
                 .addInterceptor(createLoggingInterceptor())
                 .addInterceptor(chain -> chain.proceed(authorize(chain.request())))
                 .authenticator((route, response) -> authorize(response.request()))
+                .proxy(resolveProxy(config.getProxyHost(), config.getProxyPort()))
+                .proxyAuthenticator(resolveProxyAuthenticator(config.getProxyUsername(), config.getProxyPassword()))
                 .build();
     }
 
@@ -88,6 +97,17 @@ private static HttpLoggingInterceptor createLoggingInterceptor() {
         }
     }
 
+    private static Proxy resolveProxy(String host, Integer port) {
+        return isEmpty(host) ? NO_PROXY : new Proxy(HTTP, new InetSocketAddress(host, port));
+    }
+
+    private static Authenticator resolveProxyAuthenticator(String username, String password) {
+        return isEmpty(username) ? Authenticator.NONE :
+                (route, response) -> response.request().newBuilder()
+                        .header("Proxy-Authorization", basic(username, password))
+                        .build();
+    }
+
     @Override
     @SneakyThrows(IOException.class)
     public HttpResponse execute(HttpRequest httpRequest) {

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/client/okhttp/OkHttpClientConfig.java

@@ -33,6 +33,7 @@
 import static org.apache.kafka.common.config.ConfigDef.Type.CLASS;
 import static org.apache.kafka.common.config.ConfigDef.Type.INT;
 import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
 
 @Getter
 public class OkHttpClientConfig extends AbstractConfig {
@@ -42,12 +43,20 @@ public class OkHttpClientConfig extends AbstractConfig {
     private static final String CONNECTION_KEEP_ALIVE_DURATION_MILLIS = "http.client.ttl.millis";
     private static final String CONNECTION_MAX_IDLE = "http.client.max-idle";
     private static final String AUTHENTICATOR = "http.auth";
+    private static final String PROXY_HOST = "http.client.proxy.host";
+    private static final String PROXY_PORT = "http.client.proxy.port";
+    private static final String PROXY_USERNAME = "http.client.proxy.username";
+    private static final String PROXY_PASSWORD = "http.client.proxy.password";
 
     private final Long connectionTimeoutMillis;
     private final Long readTimeoutMillis;
     private final Long keepAliveDuration;
     private final Integer maxIdleConnections;
     private final HttpAuthenticator authenticator;
+    private final String proxyHost;
+    private final Integer proxyPort;
+    private final String proxyUsername;
+    private final String proxyPassword;
 
     OkHttpClientConfig(Map<String, ?> originals) {
         super(config(), originals);
@@ -56,6 +65,10 @@ public class OkHttpClientConfig extends AbstractConfig {
         keepAliveDuration = getLong(CONNECTION_KEEP_ALIVE_DURATION_MILLIS);
         maxIdleConnections = getInt(CONNECTION_MAX_IDLE);
         authenticator = getConfiguredInstance(AUTHENTICATOR, HttpAuthenticator.class);
+        proxyHost = getString(PROXY_HOST);
+        proxyPort = getInt(PROXY_PORT);
+        proxyUsername = getString(PROXY_USERNAME);
+        proxyPassword = getString(PROXY_PASSWORD);
     }
 
     public static ConfigDef config() {
@@ -64,6 +77,11 @@ public static ConfigDef config() {
                 .define(READ_TIMEOUT_MILLIS, LONG, 2000, HIGH, "Read Timeout Millis")
                 .define(CONNECTION_KEEP_ALIVE_DURATION_MILLIS, LONG, 300000, HIGH, "Keep Alive Duration Millis")
                 .define(CONNECTION_MAX_IDLE, INT, 1, HIGH, "Max Idle Connections")
-                .define(AUTHENTICATOR, CLASS, ConfigurableHttpAuthenticator.class, MEDIUM, "Custom Authenticator");
+                .define(AUTHENTICATOR, CLASS, ConfigurableHttpAuthenticator.class, MEDIUM, "Custom Authenticator")
+                .define(PROXY_HOST, STRING, "", MEDIUM, "Proxy host")
+                .define(PROXY_PORT, INT, 3128, MEDIUM, "Proxy port")
+                .define(PROXY_USERNAME, STRING, "", MEDIUM, "Proxy username")
+                .define(PROXY_PASSWORD, STRING, "", MEDIUM, "Proxy password")
+                ;
     }
 }

kafka-connect-http/src/test/java/com/github/castorm/kafka/connect/http/client/okhttp/OkHttpClientConfigTest.java

@@ -82,6 +82,46 @@ void whenAuthenticator_thenInitialized() {
         assertThat(config(ImmutableMap.of("http.auth", "com.github.castorm.kafka.connect.http.auth.BasicHttpAuthenticator")).getAuthenticator()).isInstanceOf(BasicHttpAuthenticator.class);
     }
 
+    @Test
+    void whenNoProxyHost_thenDefault() {
+        assertThat(config(emptyMap()).getProxyHost()).isEqualTo("");
+    }
+
+    @Test
+    void whenProxyHost_thenInitialized() {
+        assertThat(config(ImmutableMap.of("http.client.proxy.host", "host")).getProxyHost()).isEqualTo("host");
+    }
+
+    @Test
+    void whenNoProxyPort_thenDefault() {
+        assertThat(config(emptyMap()).getProxyPort()).isEqualTo(3128);
+    }
+
+    @Test
+    void whenProxyPort_thenInitialized() {
+        assertThat(config(ImmutableMap.of("http.client.proxy.port", "8080")).getProxyPort()).isEqualTo(8080);
+    }
+
+    @Test
+    void whenNoProxyUsername_thenDefault() {
+        assertThat(config(emptyMap()).getProxyUsername()).isEqualTo("");
+    }
+
+    @Test
+    void whenProxyUsername_thenInitialized() {
+        assertThat(config(ImmutableMap.of("http.client.proxy.username", "user")).getProxyUsername()).isEqualTo("user");
+    }
+
+    @Test
+    void whenNoProxyPassword_thenDefault() {
+        assertThat(config(emptyMap()).getProxyPassword()).isEqualTo("");
+    }
+
+    @Test
+    void whenProxyPassword_thenInitialized() {
+        assertThat(config(ImmutableMap.of("http.client.proxy.password", "pass")).getProxyPassword()).isEqualTo("pass");
+    }
+
     private static OkHttpClientConfig config(Map<String, String> config) {
         return new OkHttpClientConfig(config);
     }