Added support for http proxy (#28)

* Added support to use proxy configs for HTTP connection

* Added tets case for porxy support

* Removed unnecessary functions

* Refactored test case to use mock extension port as proxy config

* Corrected log input

* 1. PORT parsing set to fail fast in case not integer value passed to connector configuration
2. Remove unused variable
3. AEPSinkConnectorTest to verify call to baseUrl
4. Update readMe for proxy configuration

* Review Comments :
1. Example to export http proxy configuration
2. Add reference to oracle documentation on proxy configuration

Co-authored-by: Swapnil Desai <[email protected]>
Co-authored-by: vivetiwa <[email protected]>

Author: 3 people

DEVELOPER_GUIDE.md

@@ -34,6 +34,10 @@ AEP Sink connector configurations can be supplied in the call register the conne
 | key.converter.schemas.enable      | enables conversion of schemas                   | false                                                   | no       |                         |
 | value.converter.schemas.enable    | enables conversion of schemas                   | false                                                   | no       |                         |
 | aep.endpoint                      | aep streaming endpoint url                      |                                                         | yes      |                         |
+| aep.connection.proxy.host         | address of the proxy host to connect through    |                                                         | no       |                         |
+| aep.connection.proxy.port         | port of the proxy host to connect through       | 443                                                     | no       |                         |
+| aep.connection.proxy.user         | username for the proxy host                     |                                                         | no       |                         |
+| aep.connection.proxy.password     | password for the proxy host                     |                                                         | no       |                         |
 | aep.connection.auth.enabled       | required for authenticated streaming endpoint   | false                                                   | no       |                         |
 | aep.connection.auth.token.type    | always set to access_token                      | access_token                                            | no       |                         |
 | aep.connection.auth.client.id     | IMS client id                                   |                                                         | no       |                         |
@@ -281,6 +285,27 @@ curl -s -X POST \
 }' http://localhost:8083/connectors
 ```
 
+#### Poxy host configuration
+There are 2 ways to route request to aep endpoint through proxy server :
+1. **Using Environment Variable** : Export poxyHost and proxyPort on each kafka node, then restart kafka connect node.
+   
+    For HTTPS use following :
+    ```
+    export KAFKA_OPTS="-Dhttps.proxyHost=127.0.0.1 -Dhttps.proxyPort=8085 -Dhttps.proxyUser=proxyUsername -Dhttps.proxyPassword=proxyPassword"
+    ``` 
+    For HTTP use following:
+    ```
+    export KAFKA_OPTS="-Dhttp.proxyHost=127.0.0.1 -Dhttp.proxyPort=8085 -Dhttp.proxyUser=proxyUsername -Dhttp.proxyPassword=proxyPassword"
+    ``` 
+2. **Using Connector Properties** : While creating connector set following properties, default values mentioned in [connect configurations](#configuration-options).
+    ```
+    aep.connection.proxy.host                                                          
+    aep.connection.proxy.port
+    aep.connection.proxy.user
+    aep.connection.proxy.password
+    ```
+For reference, more details are in oracle documentation on configuring proxy settings in java : https://docs.oracle.com/javase/8/docs/technotes/guides/net/proxies.html
+
 #### Use the Kafka Topics UI to view your topics
 
 The docker setup comes with Topics UI to view the topic and messages within.

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpConnection.java

@@ -14,15 +14,21 @@
 
 import com.adobe.platform.streaming.auth.AuthException;
 import com.adobe.platform.streaming.auth.AuthProvider;
+
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.Authenticator;
 import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
 import java.net.MalformedURLException;
+import java.net.PasswordAuthentication;
+import java.net.Proxy;
 import java.net.URL;
 import java.util.HashMap;
 import java.util.Map;
@@ -42,6 +48,12 @@ public class HttpConnection {
 
   private String endpoint;
   private String url;
+
+  private String proxyHost;
+  private int proxyPort;
+  private String proxyUser;
+  private String proxyPassword;
+
   private Map<String, String> headers;
   private byte[] postData;
   private int maxRetries;
@@ -69,9 +81,29 @@ HttpURLConnection connect() throws HttpException {
 
     while (retries++ < maxRetries) {
       try {
-        URL request = new URL(new URL(endpoint), url);
+        final URL request = new URL(new URL(endpoint), url);
         LOG.debug("opening connection for: {}", request);
-        conn = (HttpURLConnection) request.openConnection();
+
+        if (isBasicProxyConfigured()) {
+          if (isProxyWithAuthenticationConfigured()) {
+            LOG.debug("proxyUser: {}, proxyPassword: {}", proxyUser, proxyPassword);
+            Authenticator.setDefault(
+              new Authenticator() {
+                @Override
+                protected PasswordAuthentication getPasswordAuthentication() {
+                  return new PasswordAuthentication(proxyUser, proxyPassword.toCharArray());
+                }
+              }
+            );
+          }
+
+          LOG.debug("proxyHost: {}, proxyPort: {}", proxyHost, proxyPort);
+          final Proxy proxy = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(proxyHost, proxyPort));
+          conn = (HttpURLConnection) request.openConnection(proxy);
+        } else {
+          conn = (HttpURLConnection) request.openConnection();
+        }
+
         conn.setDoInput(true);
         conn.setUseCaches(false);
         conn.setConnectTimeout(connectTimeout);
@@ -176,6 +208,15 @@ void close() {
     }
   }
 
+  public boolean isBasicProxyConfigured() {
+    return StringUtils.isNotEmpty(proxyHost);
+  }
+
+  public boolean isProxyWithAuthenticationConfigured() {
+    return StringUtils.isNotEmpty(proxyUser) &&
+           StringUtils.isNotEmpty(proxyPassword);
+  }
+
   public InputStream getInputStream() throws HttpException {
     try {
       if (isGzip()) {
@@ -205,6 +246,26 @@ HttpConnectionBuilder withUrl(String url) {
       return this;
     }
 
+    HttpConnectionBuilder withProxyHost(String proxyHost) {
+      instance.proxyHost = proxyHost;
+      return this;
+    }
+
+    HttpConnectionBuilder withProxyPort(int proxyPort) {
+      instance.proxyPort = proxyPort;
+      return this;
+    }
+
+    HttpConnectionBuilder withProxyUser(String proxyUser) {
+      instance.proxyUser = proxyUser;
+      return this;
+    }
+
+    HttpConnectionBuilder withProxyPassword(String proxyPassword) {
+      instance.proxyPassword = proxyPassword;
+      return this;
+    }
+
     HttpConnectionBuilder withHeaders(Map<String, String> headers) {
       instance.headers.putAll(headers);
       return this;

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpProducer.java

@@ -30,6 +30,12 @@ public class HttpProducer implements Serializable {
   private static final String CONTENT_TYPE = "Content-type";
 
   private final String endpoint;
+
+  private String proxyHost;
+  private int proxyPort;
+  private String proxyUser;
+  private String proxyPassword;
+
   private final boolean enableGzip;
   private int connectTimeout;
   private transient AuthProvider auth;
@@ -75,6 +81,10 @@ private <T> T post(String url, byte[] postData, Map<String, String> headers, Con
   private HttpConnection.HttpConnectionBuilder newConnectionBuilder() {
     return new HttpConnection.HttpConnectionBuilder()
       .withEndpoint(endpoint)
+      .withProxyHost(proxyHost)
+      .withProxyPort(proxyPort)
+      .withProxyUser(proxyUser)
+      .withProxyPassword(proxyPassword)
       .withConnectTimeout(connectTimeout)
       .withAuth(auth)
       .withReadTimeout(readTimeout)
@@ -96,6 +106,26 @@ public static class HttpProducerBuilder {
       this.instance = instance;
     }
 
+    public HttpProducerBuilder withProxyHost(String proxyHost) {
+      instance.proxyHost = proxyHost;
+      return this;
+    }
+
+    public HttpProducerBuilder withProxyPort(int proxyPort) {
+      instance.proxyPort = proxyPort;
+      return this;
+    }
+
+    public HttpProducerBuilder withProxyUser(String proxyUser) {
+      instance.proxyUser = proxyUser;
+      return this;
+    }
+
+    public HttpProducerBuilder withProxyPassword(String proxyPassword) {
+      instance.proxyPassword = proxyPassword;
+      return this;
+    }
+
     public HttpProducerBuilder withReadTimeout(int readTimeout) {
       instance.readTimeout = readTimeout;
       return this;

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/AbstractAEPPublisher.java

@@ -38,6 +38,12 @@ public abstract class AbstractAEPPublisher implements DataPublisher {
   private static final Logger LOG = LoggerFactory.getLogger(AbstractAEPPublisher.class);
 
   private static final String AEP_ENDPOINT = "aep.endpoint";
+
+  private static final String AEP_CONNECTION_PROXY_HOST = "aep.connection.proxy.host";
+  private static final String AEP_CONNECTION_PROXY_PORT = "aep.connection.proxy.port";
+  private static final String AEP_CONNECTION_PROXY_USER = "aep.connection.proxy.user";
+  private static final String AEP_CONNECTION_PROXY_PASSWORD = "aep.connection.proxy.password";
+
   private static final String AEP_CONNECTION_TIMEOUT = "aep.connection.timeout";
   private static final String AEP_CONNECTION_MAX_RETRIES = "aep.connection.maxRetries";
   private static final String AEP_CONNECTION_MAX_RETRIES_BACKOFF = "aep.connection.retryBackoff";
@@ -62,6 +68,10 @@ public abstract class AbstractAEPPublisher implements DataPublisher {
 
   protected HttpProducer getHttpProducer(Map<String, String> props) throws AEPStreamingException {
     return HttpProducer.newBuilder(getAepEndpoint(props.get(AEP_ENDPOINT)))
+      .withProxyHost(SinkUtils.getProperty(props, AEP_CONNECTION_PROXY_HOST, null))
+      .withProxyPort(SinkUtils.getProperty(props, AEP_CONNECTION_PROXY_PORT, 443))
+      .withProxyUser(SinkUtils.getProperty(props, AEP_CONNECTION_PROXY_USER, null))
+      .withProxyPassword(SinkUtils.getProperty(props, AEP_CONNECTION_PROXY_PASSWORD, null))
       .withConnectTimeout(SinkUtils.getProperty(props, AEP_CONNECTION_TIMEOUT, 5000))
       .withReadTimeout(SinkUtils.getProperty(props, AEP_CONNECTION_READ_TIMEOUT, 60000))
       .withMaxRetries(SinkUtils.getProperty(props, AEP_CONNECTION_MAX_RETRIES, 3))

streaming-connect-sink/src/test/java/com/adobe/platform/streaming/integration/AEPSinkConnectorTest.java

@@ -42,13 +42,15 @@ public class AEPSinkConnectorTest extends AbstractConnectorTest {
   private static final Logger LOG = LoggerFactory.getLogger(AEPSinkConnectorTest.class);
 
   private static final String AEP_CONNECTOR_CONFIG = "aep-connector.json";
+  private static final String AEP_CONNECTOR_CONFIG_WITH_PROXY = "aep-connector-with-proxy.json";
   private static final String XDM_PAYLOAD_FILE = "xdm-data.json";
 
   @BeforeEach
   @Override
   public void setup() throws JsonProcessingException {
     super.setup();
     inletSuccessfulResponse();
+    inletSuccessfulResponseViaProxy();
   }
 
   @Test
@@ -68,6 +70,25 @@ public void aepSinkConnectorTest() throws HttpException, JsonProcessingException
       .withRequestBody(equalToJson(payloadReceivedXdmData())));
   }
 
+  @Test
+  public void aepSinkConnectorTestViaProxy() throws HttpException, JsonProcessingException, InterruptedException {
+    getConnect().kafka().createTopic(TOPIC_NAME, TOPIC_PARTITION);
+
+    LOG.info("Starting connector cluster with connector : {}", CONNECTOR_NAME);
+    getConnect().configureConnector(CONNECTOR_NAME, connectorConfigWithProxy());
+
+    // Send single XDM data to aep sink connector
+    getConnect().kafka().produce(TOPIC_NAME, xdmData());
+
+    waitForConnectorStart(CONNECTOR_NAME, 1, 1000);
+
+    // Verify inlet endpoint received 1 XDM record
+    getWiremockServerViaProxy().verify(postRequestedFor(urlEqualTo(getRelativeUrl()))
+      .withRequestBody(equalToJson(payloadReceivedXdmData())));
+    // Check if request from proxy server forward to AEP endpoint
+    getWiremockServer().verify(postRequestedFor(urlEqualTo(getRelativeUrl())));
+  }
+
   @AfterEach
   @Override
   public void stop() {
@@ -101,4 +122,17 @@ public Map<String, String> connectorConfig() throws HttpException, JsonProcessin
     return connectorConfig;
   }
 
+  public Map<String, String> connectorConfigWithProxy() throws HttpException, JsonProcessingException {
+    String connectorProperties = String.format(HttpUtil.streamToString(this.getClass().getClassLoader()
+      .getResourceAsStream(AEP_CONNECTOR_CONFIG_WITH_PROXY)),
+      NUMBER_OF_TASKS,
+      getInletUrl(),
+      PORT_VIA_PROXY);
+
+    Map<String, String> connectorConfig = MAPPER.readValue(connectorProperties,
+      new TypeReference<Map<String, String>>() {});
+    connectorConfig.put("name", CONNECTOR_NAME);
+    return connectorConfig;
+  }
+
 }

streaming-connect-sink/src/test/java/com/adobe/platform/streaming/integration/AbstractConnectorTest.java

@@ -50,6 +50,7 @@ public abstract class AbstractConnectorTest {
   protected static final String CONNECTOR_NAME = "aep-sink-connector";
   protected static final String TOPIC_NAME = "connect-test";
   protected static final int PORT = 8089;
+  protected static final int PORT_VIA_PROXY = 8090;
   private EmbeddedConnectCluster connect;
   private int numberOfWorkers = 1;
   private String inletId = "876e1041c16801b8b3038ec86bb4510e8c89356152191b587367b592e79d91d5";
@@ -59,6 +60,9 @@ public abstract class AbstractConnectorTest {
   @RegisterExtension
   public static final WiremockExtension wiremockExtension = new WiremockExtension(PORT);
 
+  @RegisterExtension
+  public static final WiremockExtension wiremockExtensionViaProxy = new WiremockExtension(PORT_VIA_PROXY);
+
   @BeforeEach
   public void setup() throws JsonProcessingException {
     connect = new EmbeddedConnectCluster.Builder()
@@ -109,6 +113,13 @@ public void inletSuccessfulResponse() throws JsonProcessingException {
       .withJsonBody(MAPPER.readTree("{\"payloadReceived\": true}"))));
   }
 
+  public void inletSuccessfulResponseViaProxy() {
+    wiremockExtensionViaProxy.getWireMockServer()
+      .stubFor(WireMock
+      .post(WireMock.urlEqualTo(getRelativeUrl()))
+      .willReturn(ResponseDefinitionBuilder.responseDefinition().proxiedFrom(baseUrl)));
+  }
+
   public void inletFailedResponse() {
     wiremockExtension.getWireMockServer()
       .stubFor(WireMock
@@ -120,6 +131,10 @@ protected WireMockServer getWiremockServer() {
     return wiremockExtension.getWireMockServer();
   }
 
+  protected WireMockServer getWiremockServerViaProxy() {
+    return wiremockExtensionViaProxy.getWireMockServer();
+  }
+
   protected String getRelativeUrl() {
     return relativePath.replace("/collection/", "/collection/batch/");
   }

streaming-connect-sink/src/test/resources/aep-connector-with-proxy.json

@@ -0,0 +1,12 @@
+{
+  "topics": "connect-test",
+  "tasks.max": "%s",
+  "aep.flush.interval.seconds": "1",
+  "aep.flush.bytes.kb": "4",
+  "connector.class": "com.adobe.platform.streaming.sink.impl.AEPSinkConnector",
+  "key.converter.schemas.enable": "false",
+  "value.converter.schemas.enable": "false",
+  "aep.endpoint": "%s",
+  "aep.connection.proxy.host": "localhost",
+  "aep.connection.proxy.port": "%s"
+}