Drop Dependency for Apache HttpClient (#17)

* Drop_apacheHttpClient_Dependency - drop dependency

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

* Drop_apacheHttpClient_Dependency - Licence update for httpcomponents-client code.

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

* Drop_apacheHttpClient_Dependency - Update README.md

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

* Drop_apacheHttpClient_Dependency - Changes after code review.

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Co-authored-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Author: kristoffSC

CHANGELOG.md

@@ -9,6 +9,12 @@
 - Add new properties `gid.connector.http.sink.error.code` and `gid.connector.http.sink.error.code.exclude`
   to set HTTP status code that should be interpreted as errors.
 
+### Changed
+- Change dependency scope for `org.apache.flink.flink-connector-base` from `compile` to `provided`.
+
+### Removed
+- Remove dependency on `org.apache.httpcomponents.httpclient`from production code. Dependency is only for test scope.
+
 ## [0.3.0] - 2022-07-21
 
 -   Package refactoring. Hide internal classes that does not have to be used by API users under "internal" package.

README.md

@@ -18,6 +18,13 @@ Currently, HTTP TableLookup connector supports only Lookup Joins [1] and expects
 * Maven 3
 * Flink 1.15+
 
+## Runtime dependencies
+This connector has few Flink's runtime dependencies, that are expected to be provided.
+* `org.apache.flink.flink-java`
+* `org.apache.flink.flink-clients`
+* `org.apache.flink.flink-connector-base`
+* `org.apache.flink.flink-java`
+
 ## Installation
 
 In order to use the `flink-http-connector` the following dependencies are required for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles. For build automation tool reference, look into Maven Central: [https://mvnrepository.com/artifact/com.getindata/flink-http-connector](https://mvnrepository.com/artifact/com.getindata/flink-http-connector).

pom.xml

@@ -148,6 +148,7 @@ under the License.
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-connector-base</artifactId>
       <version>${flink.version}</version>
+      <scope>provided</scope>
     </dependency>
 
     <!--===== Needed Dep ====-->
@@ -157,12 +158,6 @@ under the License.
       <version>${json.path.version}</version>
     </dependency>
 
-    <dependency>
-      <groupId>org.apache.httpcomponents</groupId>
-      <artifactId>httpclient</artifactId>
-      <version>4.5.13</version>
-    </dependency>
-
     <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
@@ -171,6 +166,13 @@ under the License.
     </dependency>
 
     <!--TEST-->
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpclient</artifactId>
+      <version>4.5.13</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-table-common</artifactId>

src/main/java/com/getindata/connectors/http/internal/JsonResultTableConverter.java

@@ -1,6 +1,5 @@
 package com.getindata.connectors.http.internal;
 
-
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;

src/main/java/com/getindata/connectors/http/internal/config/ConfigException.java

@@ -16,6 +16,13 @@ public ConfigException(String message, Throwable t) {
         super(message, t);
     }
 
+    /**
+     * Creates an exception object using predefined exception message template:
+     * {@code Invalid value + (value) + for configuration + (property name) + (additional message) }
+     * @param name configuration property name.
+     * @param value configuration property value.
+     * @param message custom message appended to the end of exception message.
+     */
     public ConfigException(String name, Object value, String message) {
         super("Invalid value " + value + " for configuration " + name + (message == null ? ""
             : ": " + message));

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkWriter.java

@@ -27,35 +27,25 @@
 public class HttpSinkWriter<InputT> extends AsyncSinkWriter<InputT, HttpSinkRequestEntry> {
 
     private final String endpointUrl;
+
     private final SinkHttpClient sinkHttpClient;
-    private final Counter numRecordsSendErrorsCounter;
 
-    public HttpSinkWriter(
-        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter, Sink.InitContext context,
-        int maxBatchSize,
-        int maxInFlightRequests, int maxBufferedRequests, long maxBatchSizeInBytes,
-        long maxTimeInBufferMS,
-        long maxRecordSizeInBytes, String endpointUrl, SinkHttpClient sinkHttpClient
-    ) {
-        this(elementConverter, context, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
-            maxBatchSizeInBytes,
-            maxTimeInBufferMS, maxRecordSizeInBytes, endpointUrl, sinkHttpClient,
-            Collections.emptyList()
-        );
-    }
+    private final Counter numRecordsSendErrorsCounter;
 
     public HttpSinkWriter(
-        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter, Sink.InitContext context,
-        int maxBatchSize,
-        int maxInFlightRequests, int maxBufferedRequests, long maxBatchSizeInBytes,
-        long maxTimeInBufferMS,
-        long maxRecordSizeInBytes, String endpointUrl, SinkHttpClient sinkHttpClient,
-        Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates
-    ) {
+            ElementConverter<InputT, HttpSinkRequestEntry> elementConverter,
+            Sink.InitContext context,
+            int maxBatchSize,
+            int maxInFlightRequests,
+            int maxBufferedRequests,
+            long maxBatchSizeInBytes,
+            long maxTimeInBufferMS,
+            long maxRecordSizeInBytes,
+            String endpointUrl,
+            SinkHttpClient sinkHttpClient,
+            Collection<BufferedRequestState<HttpSinkRequestEntry>> bufferedRequestStates) {
         super(elementConverter, context, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
-            maxBatchSizeInBytes,
-            maxTimeInBufferMS, maxRecordSizeInBytes, bufferedRequestStates
-        );
+            maxBatchSizeInBytes, maxTimeInBufferMS, maxRecordSizeInBytes, bufferedRequestStates);
         this.endpointUrl = endpointUrl;
         this.sinkHttpClient = sinkHttpClient;
 
@@ -67,13 +57,13 @@ public HttpSinkWriter(
     @Override
     protected void submitRequestEntries(
         List<HttpSinkRequestEntry> requestEntries,
-        Consumer<List<HttpSinkRequestEntry>> requestResult
-    ) {
+        Consumer<List<HttpSinkRequestEntry>> requestResult) {
         var future = sinkHttpClient.putRequests(requestEntries, endpointUrl);
         future.whenComplete((response, err) -> {
             if (err != null) {
-                var failedRequestsNumber = requestEntries.size();
-                log.error("Http Sink fatally failed to write all {} requests",
+                int failedRequestsNumber = requestEntries.size();
+                log.error(
+                    "Http Sink fatally failed to write all {} requests",
                     failedRequestsNumber);
                 numRecordsSendErrorsCounter.inc(failedRequestsNumber);
 
@@ -83,7 +73,7 @@ protected void submitRequestEntries(
                 //  a clear image how we want to do it, so it would be both efficient and correct.
                 //requestResult.accept(requestEntries);
             } else if (response.getFailedRequests().size() > 0) {
-                var failedRequestsNumber = response.getFailedRequests().size();
+                int failedRequestsNumber = response.getFailedRequests().size();
                 log.error("Http Sink failed to write and will retry {} requests",
                     failedRequestsNumber);
                 numRecordsSendErrorsCounter.inc(failedRequestsNumber);

src/main/java/com/getindata/connectors/http/internal/table/lookup/RestTablePollingClient.java

@@ -12,10 +12,10 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.table.data.RowData;
-import org.apache.http.client.utils.URIBuilder;
 
 import com.getindata.connectors.http.internal.JsonResultTableConverter;
 import com.getindata.connectors.http.internal.PollingClient;
+import com.getindata.connectors.http.internal.utils.uri.URIBuilder;
 
 @Slf4j
 @RequiredArgsConstructor

src/main/java/com/getindata/connectors/http/internal/utils/uri/CharArrayBuffer.java

@@ -0,0 +1,129 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ * ============================= NOTE =================================
+ * This code has been copied from
+ * https://github.com/apache/httpcomponents-client/tree/rel/v4.5.13
+ * and it was changed to use in this project.
+ * ====================================================================
+ */
+
+package com.getindata.connectors.http.internal.utils.uri;
+
+import java.io.Serializable;
+import java.nio.CharBuffer;
+
+import org.apache.flink.util.Preconditions;
+
+/**
+ * A resizable char array.
+ */
+final class CharArrayBuffer implements CharSequence, Serializable {
+
+    private static final long serialVersionUID = -6208952725094867135L;
+
+    private char[] buffer;
+
+    private int len;
+
+    /**
+     * Creates an instance of {@link CharArrayBuffer} with the given initial capacity.
+     *
+     * @param capacity the capacity
+     */
+    CharArrayBuffer(final int capacity) {
+        super();
+        Preconditions.checkArgument(capacity > 0, "Buffer capacity must be bigger than 0.");
+        this.buffer = new char[capacity];
+    }
+
+    /**
+     * Appends chars of the given string to this buffer. The capacity of the buffer is increased, if
+     * necessary, to accommodate all chars.
+     *
+     * @param str the string.
+     */
+    void append(final String str) {
+        final String s = str != null ? str : "null";
+        final int strLen = s.length();
+        final int newLen = this.len + strLen;
+        if (newLen > this.buffer.length) {
+            expand(newLen);
+        }
+        s.getChars(0, strLen, this.buffer, this.len);
+        this.len = newLen;
+    }
+
+    /**
+     * Returns the {@code char} value in this buffer at the specified index. The index argument must
+     * be greater than or equal to {@code 0}, and less than the length of this buffer.
+     *
+     * @param i the index of the desired char value.
+     * @return the char value at the specified index.
+     * @throws IndexOutOfBoundsException if {@code index} is negative or greater than or equal to
+     *                                   {@link #length()}.
+     */
+    @Override
+    public char charAt(final int i) {
+        return this.buffer[i];
+    }
+
+    /**
+     * Returns the length of the buffer (char count).
+     *
+     * @return the length of the buffer
+     */
+    @Override
+    public int length() {
+        return this.len;
+    }
+
+    @Override
+    public CharSequence subSequence(final int beginIndex, final int endIndex) {
+        if (beginIndex < 0) {
+            throw new IndexOutOfBoundsException("Negative beginIndex: " + beginIndex);
+        }
+        if (endIndex > this.len) {
+            throw new IndexOutOfBoundsException("endIndex: " + endIndex + " > length: " + this.len);
+        }
+        if (beginIndex > endIndex) {
+            throw new IndexOutOfBoundsException(
+                "beginIndex: " + beginIndex + " > endIndex: " + endIndex);
+        }
+        return CharBuffer.wrap(this.buffer, beginIndex, endIndex);
+    }
+
+    private void expand(final int newLen) {
+        final char[] newBuffer = new char[Math.max(this.buffer.length << 1, newLen)];
+        System.arraycopy(this.buffer, 0, newBuffer, 0, this.len);
+        this.buffer = newBuffer;
+    }
+
+    @Override
+    public String toString() {
+        return new String(this.buffer, 0, this.len);
+    }
+
+}

src/main/java/com/getindata/connectors/http/internal/utils/uri/NameValuePair.java

@@ -0,0 +1,56 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ * ============================= NOTE =================================
+ * This code has been copied from
+ * https://github.com/apache/httpcomponents-client/tree/rel/v4.5.13
+ * and it was changed to use in this project.
+ * ====================================================================
+ */
+
+package com.getindata.connectors.http.internal.utils.uri;
+
+import lombok.Data;
+import org.apache.flink.util.Preconditions;
+
+@Data
+class NameValuePair {
+
+    private final String name;
+
+    private final String value;
+
+    /**
+     * Default Constructor taking a name and a value. The value may be null.
+     *
+     * @param name The name.
+     * @param value The value.
+     */
+    NameValuePair(final String name, final String value) {
+        super();
+        this.name = Preconditions.checkNotNull(name, "Name may not be null");
+        this.value = value;
+    }
+}