Merge pull request #8 from getindata/Package_refactoring

Package_refactoring - add internal package and expose public API

Author: kristoffSC

CHANGELOG.md

@@ -2,14 +2,18 @@
 
 ## [Unreleased]
 
+- Package refactoring. Hide internal classes that does not have to be used by API users under "internal" package.
+  Methods defined in classes located outside "internal" package are considered "public API".
+  Any changes to those methods should be communicated as "not backward compatible" and should be avoided.
+
 ## [0.2.0] - 2022-07-06
 
--   Implement [HttpSink](src/main/java/com/getindata/connectors/http/sink/HttpSink.java) deriving from [AsyncSinkBase](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) introduced in Flink 1.15.
--   Add support for Table API in HttpSink in the form of [HttpDynamicSink](src/main/java/com/getindata/connectors/http/table/HttpDynamicSink.java). 
+- Implement [HttpSink](src/main/java/com/getindata/connectors/http/HttpSink.java) deriving from [AsyncSinkBase](https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink) introduced in Flink 1.15.
+- Add support for Table API in HttpSink in the form of [HttpDynamicSink](src/main/java/com/getindata/connectors/http/internal/table/sink/HttpDynamicSink.java). 
 
 ## [0.1.0] - 2022-05-26
 
--   Implement basic support for Http connector for Flink SQL
+- Implement basic support for Http connector for Flink SQL
 
 [Unreleased]: https://github.com/getindata/flink-http-connector/compare/0.2.0...HEAD
 

README.md

@@ -7,7 +7,7 @@ The goal for HTTP TableLookup connector was to use it in Flink SQL statement as
 Currently, HTTP TableLookup connector supports only Lookup Joins [1] and expects JSON as a response body. It also supports only the STRING types.
 
 #### HTTP Sink
-`HttpSink` supports both Streaming API (when using [HttpSink](src/main/java/com/getindata/connectors/http/sink/HttpSink.java) built using [HttpSinkBuilder](src/main/java/com/getindata/connectors/http/sink/HttpSinkBuilder.java)) and the Table API (using connector created in [HttpDynamicTableSinkFactory](src/main/java/com/getindata/connectors/http/table/HttpDynamicTableSinkFactory.java)). 
+`HttpSink` supports both Streaming API (when using [HttpSink](src/main/java/com/getindata/connectors/http/internal/sink/HttpSink.java) built using [HttpSinkBuilder](src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkBuilder.java)) and the Table API (using connector created in [HttpDynamicTableSinkFactory](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicTableSinkFactory.java)). 
 
 ## Prerequisites
 * Java 11
@@ -53,7 +53,7 @@ http://localhost:8080/client/service?id=1&uuid=2
 ``
 
 ### HTTP Sink
-The following example shows the minimum Table API example to create a [HttpDynamicSink](src/main/java/com/getindata/connectors/http/table/HttpDynamicSink.java) that writes JSON values to an HTTP endpoint using POST method, assuming Flink has JAR of [JSON serializer](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/json/) installed:
+The following example shows the minimum Table API example to create a [HttpDynamicSink](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicSink.java) that writes JSON values to an HTTP endpoint using POST method, assuming Flink has JAR of [JSON serializer](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/json/) installed:
 
 ```roomsql
 CREATE TABLE http (

src/main/java/com/getindata/connectors/http/HttpSink.java

@@ -0,0 +1,59 @@
+package com.getindata.connectors.http;
+
+import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
+import com.getindata.connectors.http.internal.sink.HttpSinkInternal;
+import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+/**
+ * A public implementation for {@code HttpSink} that performs async requests against a specified
+ * HTTP endpoint using the buffering protocol specified in
+ * {@link org.apache.flink.connector.base.sink.AsyncSinkBase}.
+ *
+ * <p>
+ * To create a new instance  of this class use {@link HttpSinkBuilder}. An example would be:
+ * <pre>
+ * HttpSink<String> httpSink =
+ *     HttpSink.<String>builder()
+ *             .setEndpointUrl("http://example.com/myendpoint")
+ *             .setElementConverter(
+ *                 (s, _context) -> new HttpSinkRequestEntry("POST", "text/plain", s.getBytes(StandardCharsets.UTF_8)))
+ *             .build();
+ * </pre>
+ *
+ * @param <InputT> type of the elements that should be sent through HTTP request.
+ */
+public class HttpSink<InputT> extends HttpSinkInternal<InputT> {
+
+    HttpSink(
+            ElementConverter<InputT, HttpSinkRequestEntry> elementConverter,
+            int maxBatchSize,
+            int maxInFlightRequests,
+            int maxBufferedRequests,
+            long maxBatchSizeInBytes,
+            long maxTimeInBufferMS,
+            long maxRecordSizeInBytes,
+            String endpointUrl,
+            SinkHttpClientBuilder sinkHttpClientBuilder) {
+        super(elementConverter,
+            maxBatchSize,
+            maxInFlightRequests,
+            maxBufferedRequests,
+            maxBatchSizeInBytes,
+            maxTimeInBufferMS,
+            maxRecordSizeInBytes,
+            endpointUrl,
+            sinkHttpClientBuilder
+        );
+    }
+
+    /**
+     * Create a {@link HttpSinkBuilder} constructing a new {@link HttpSink}.
+     *
+     * @param <InputT> type of the elements that should be sent through HTTP request
+     * @return {@link HttpSinkBuilder}
+     */
+    public static <InputT> HttpSinkBuilder<InputT> builder() {
+        return new HttpSinkBuilder<>();
+    }
+}

src/main/java/com/getindata/connectors/http/sink/HttpSinkBuilder.java renamed to src/main/java/com/getindata/connectors/http/HttpSinkBuilder.java

@@ -1,7 +1,8 @@
-package com.getindata.connectors.http.sink;
+package com.getindata.connectors.http;
 
-import com.getindata.connectors.http.SinkHttpClient;
-import com.getindata.connectors.http.SinkHttpClientBuilder;
+import com.getindata.connectors.http.internal.SinkHttpClient;
+import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
+import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 import org.apache.flink.connector.base.sink.AsyncSinkBaseBuilder;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
 
@@ -13,14 +14,14 @@
  * <p>The following example shows the minimum setup to create a {@link HttpSink} that writes String
  * values to an HTTP endpoint using POST method.
  *
- * <pre>{@code
+ * <pre>
  * HttpSink<String> httpSink =
  *     HttpSink.<String>builder()
  *             .setEndpointUrl("http://example.com/myendpoint")
  *             .setElementConverter(
  *                 (s, _context) -> new HttpSinkRequestEntry("POST", "text/plain", s.getBytes(StandardCharsets.UTF_8)))
  *             .build();
- * }</pre>
+ * </pre>
  *
  * <p>If the following parameters are not set in this builder, the following defaults will be used:
  * <ul>
@@ -60,7 +61,8 @@ public HttpSinkBuilder<InputT> setEndpointUrl(String endpointUrl) {
   }
 
   /**
-   * @param sinkHttpClientBuilder builder for an implementation of {@link SinkHttpClient} that will be used by {@link HttpSink}
+   * @param sinkHttpClientBuilder builder for an implementation of {@link SinkHttpClient} that
+   *                              will be used by {@link HttpSink}
    * @return {@link HttpSinkBuilder} itself
    */
   public HttpSinkBuilder<InputT> setSinkHttpClientBuilder(SinkHttpClientBuilder sinkHttpClientBuilder) {

src/main/java/com/getindata/connectors/http/PollingClient.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http;
+package com.getindata.connectors.http.internal;
 
 import java.io.Serializable;
 import org.apache.flink.table.data.RowData;

src/main/java/com/getindata/connectors/http/HttpResultConverter.java renamed to src/main/java/com/getindata/connectors/http/internal/HttpResultConverter.java

@@ -1,10 +1,9 @@
-package com.getindata.connectors.http;
+package com.getindata.connectors.http.internal;
 
-import static com.getindata.connectors.http.table.TableSourceHelper.buildEmptyRow;
-import static com.getindata.connectors.http.table.TableSourceHelper.buildGenericRowData;
+import static com.getindata.connectors.http.internal.table.lookup.TableSourceHelper.buildEmptyRow;
+import static com.getindata.connectors.http.internal.table.lookup.TableSourceHelper.buildGenericRowData;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
-import com.getindata.connectors.http.table.TableSourceHelper;
 import com.jayway.jsonpath.DocumentContext;
 import com.jayway.jsonpath.JsonPath;
 import com.jayway.jsonpath.PathNotFoundException;

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

@@ -0,0 +1,9 @@
+package com.getindata.connectors.http.internal;
+
+import com.getindata.connectors.http.internal.table.lookup.LookupArg;
+import java.util.List;
+
+public interface PollingClient<T> {
+
+  T pull(List<LookupArg> params);
+}

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

@@ -1,6 +1,6 @@
-package com.getindata.connectors.http;
+package com.getindata.connectors.http.internal;
 
-import com.getindata.connectors.http.table.HttpLookupConfig;
+import com.getindata.connectors.http.internal.table.lookup.HttpLookupConfig;
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.table.functions.FunctionContext;
 

src/main/java/com/getindata/connectors/http/PollingClientFactory.java renamed to src/main/java/com/getindata/connectors/http/internal/PollingClientFactory.java

@@ -1,13 +1,15 @@
-package com.getindata.connectors.http;
+package com.getindata.connectors.http.internal;
 
-import com.getindata.connectors.http.sink.HttpSinkRequestEntry;
+import com.getindata.connectors.http.internal.sink.HttpSinkWriter;
+import com.getindata.connectors.http.internal.sink.HttpSinkInternal;
+import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 /**
- * An HTTP client that is used by {@link com.getindata.connectors.http.sink.HttpSinkWriter}
- * to send HTTP requests processed by {@link com.getindata.connectors.http.sink.HttpSink}.
+ * An HTTP client that is used by {@link HttpSinkWriter}
+ * to send HTTP requests processed by {@link HttpSinkInternal}.
  */
 public interface SinkHttpClient {
   /**