ESP-148_UseFlink_JsonFormat_InSource - Use Flink Json Format instead custom implemented Json Format for Source Lookup. (#19)

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

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

Author: kristoffSC

CHANGELOG.md

@@ -6,14 +6,17 @@
 - Fix JavaDoc errors.
 
 ### Added
-- Add new properties `gid.connector.http.sink.error.code` and `gid.connector.http.sink.error.code.exclude`
+- Add to Http Sink a 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.
+- Use Flink's format support to Http Lookup Source. 
 
 ### Changed
 - Change dependency scope for `org.apache.flink.flink-connector-base` from `compile` to `provided`.
+- Changed DDL of `rest-lookup` connector. Dropped `json-path` properties, and add mandatory `format` property.
 
 ### Removed
 - Remove dependency on `org.apache.httpcomponents.httpclient`from production code. Dependency is only for test scope.
+- Removed dependency on `com.jayway.jsonpath.json-path`
 
 ## [0.3.0] - 2022-07-21
 

README.md

@@ -10,7 +10,7 @@ Please use [releases](https://github.com/getindata/flink-http-connector/releases
 
 The goal for HTTP TableLookup connector was to use it in Flink SQL statement as a standard table that can be later joined with other stream using pure SQL Flink.
 
-Currently, HTTP TableLookup connector supports only Lookup Joins [1] and expects JSON as a response body. It also supports only the STRING types.
+Currently, HTTP source connector supports only Lookup Joins (TableLookup) [1] in Table/SQL API.
 `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
@@ -23,7 +23,6 @@ This connector has few Flink's runtime dependencies, that are expected to be pro
 * `org.apache.flink.flink-java`
 * `org.apache.flink.flink-clients`
 * `org.apache.flink.flink-connector-base`
-* `org.apache.flink.flink-java`
 
 ## Installation
 
@@ -38,23 +37,41 @@ You can read the official JavaDoc documentation of the latest release at [https:
 ### HTTP TableLookup Source
 Flink SQL table definition:
 
+Enrichment Lookup Table
 ```roomsql
 CREATE TABLE Customers (
-id STRING,
-id2 STRING,
-msg STRING,
-uuid STRING,
-isActive STRING,
-balance STRING
+	id STRING,
+	id2 STRING,
+	msg STRING,
+	uuid STRING,
+	details ROW<
+	  isActive BOOLEAN,
+	  nestedDetails ROW<
+	    balance STRING
+	  >
+	>
 ) WITH (
 'connector' = 'rest-lookup',
+'format' = 'json',
 'url' = 'http://localhost:8080/client',
-'asyncPolling' = 'true',
-'field.isActive.path' = '$.details.isActive',
-'field.balance.path' = '$.details.nestedDetails.balance'
+'asyncPolling' = 'true'
 )
 ```
-Using _Customers_ table in Flink SQL Lookup Join:
+Data Source Table
+```roomsql
+CREATE TABLE Orders (id STRING, id2 STRING, proc_time AS PROCTIME()
+) WITH (
+'connector' = 'datagen',
+'rows-per-second' = '1',
+'fields.id.kind' = 'sequence',
+'fields.id.start' = '1',
+'fields.id.end' = '120',
+'fields.id2.kind' = 'sequence',
+'fields.id2.start' = '2',
+'fields.id2.end' = '120'
+);
+
+Using _Customers_ table in Flink SQL Lookup Join with Orders table:
 
 ```roomsql
 SELECT o.id, o.id2, c.msg, c.uuid, c.isActive, c.balance FROM Orders AS o 
@@ -153,13 +170,12 @@ An example of such a mask would be `3XX, 4XX, 5XX`. In this case, all 300s, 400s
 
 ## Table API Connector Options
 ### HTTP TableLookup Source
-| Option       | Required | Description/Value                                                                                                                                                                                                                         |
-|--------------|----------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| connector    | required | The Value should be set to _rest-lookup_                                                                                                                                                                                                  |
-| url          | required | The base URL that should be use for GET requests. For example _http://localhost:8080/client_                                                                                                                                              |
-| asyncPolling | optional | true/false - determines whether Async Pooling should be used. Mechanism is based on Flink's Async I/O.                                                                                                                                    |
-| root         | optional | Sets the json root node for entire table. The value should be presented as Json Path [5], for example `$.details`.                                                                                                                        |
-| field.#.path | optional | The Json Path from response model that should be use for given `#` field. If `root` option was defined it will be added to field path. The value must be presented in Json Path format [5], for example `$.details.nestedDetails.balance` |
+| Option       | Required | Description/Value                                                                                        |
+|--------------|----------|----------------------------------------------------------------------------------------------------------|
+| connector    | required | The Value should be set to _rest-lookup_                                                                 |
+| format       | required | Flink's format name that should be used to decode REST response, Use `json` for a typical REST endpoint. |
+| url          | required | The base URL that should be use for GET requests. For example _http://localhost:8080/client_             |
+| asyncPolling | optional | true/false - determines whether Async Pooling should be used. Mechanism is based on Flink's Async I/O.   |
 
 ### HTTP Sink
 | Option                     | Required | Description/Value                                                                                                                                                                                  |
@@ -207,13 +223,22 @@ CREATE TABLE Orders (id STRING, id2 STRING, proc_time AS PROCTIME()
 
 Create Http Connector Lookup Table:
 ```roomsql
-CREATE TABLE Customers (id STRING, id2 STRING, msg STRING, uuid STRING, isActive STRING, balance STRING
+CREATE TABLE Customers (
+	id STRING,
+	id2 STRING,
+	msg STRING,
+	uuid STRING,
+	details ROW<
+	  isActive BOOLEAN,
+	  nestedDetails ROW<
+	    balance STRING
+	  >
+	>
 ) WITH (
-'connector' = 'rest-lookup', 
+'connector' = 'rest-lookup',
+'format' = 'json' 
 'url' = 'http://localhost:8080/client', 
-'asyncPolling' = 'true', 
-'field.isActive.path' = '$.details.isActive', 
-'field.balance.path' = '$.details.nestedDetails.balance'
+'asyncPolling' = 'true'
 );
 ```
 

pom.xml

@@ -77,7 +77,6 @@ under the License.
     <log4j.version>2.17.2</log4j.version>
     <lombok.version>1.18.22</lombok.version>
     <junit.jupiter.version>5.8.1</junit.jupiter.version>
-    <json.path.version>2.6.0</json.path.version>
     <assertj.core.version>3.21.0</assertj.core.version>
     <mockito.version>4.0.0</mockito.version>
     <wiremock.version>2.27.2</wiremock.version>
@@ -151,13 +150,6 @@ under the License.
       <scope>provided</scope>
     </dependency>
 
-    <!--===== Needed Dep ====-->
-    <dependency>
-      <groupId>com.jayway.jsonpath</groupId>
-      <artifactId>json-path</artifactId>
-      <version>${json.path.version}</version>
-    </dependency>
-
     <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>

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

@@ -1,10 +1,19 @@
 package com.getindata.connectors.http.internal;
 
 import java.util.List;
+import java.util.Optional;
 
 import com.getindata.connectors.http.internal.table.lookup.LookupArg;
 
+/**
+ * A client that is used to get enrichment data from external component.
+ */
 public interface PollingClient<T> {
 
-    T pull(List<LookupArg> params);
+    /**
+     * Gets enrichment data from external component using provided lookup arguments.
+     * @param lookupArgs The list of {@link LookupArg} containing request parameters.
+     * @return an optional result of data lookup.
+     */
+    Optional<T> pull(List<LookupArg> lookupArgs);
 }

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

@@ -1,14 +1,15 @@
 package com.getindata.connectors.http.internal;
 
-import org.apache.flink.api.connector.source.SourceReaderContext;
-import org.apache.flink.table.functions.FunctionContext;
+import java.io.Serializable;
 
-import com.getindata.connectors.http.internal.table.lookup.HttpLookupConfig;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
 
-public interface PollingClientFactory<OUT> {
+import com.getindata.connectors.http.internal.table.lookup.HttpLookupConfig;
 
-    PollingClient<OUT> createPollClient(FunctionContext context,
-        HttpLookupConfig options);
+public interface PollingClientFactory<OUT> extends Serializable {
 
-    PollingClient<OUT> createPollClient(SourceReaderContext readerContext);
+    PollingClient<OUT> createPollClient(
+        HttpLookupConfig options,
+        DeserializationSchema<OUT> schemaDecoder
+    );
 }

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

@@ -2,6 +2,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -24,10 +25,20 @@ public class AsyncHttpTableLookupFunction extends AsyncTableFunction<RowData> {
 
     private static final int PUBLISHING_THREAD_POOL_SIZE = 4;
 
+    /**
+     * The {@link org.apache.flink.table.functions.TableFunction} we want to decorate with
+     * async framework.
+     */
     private final HttpTableLookupFunction decorate;
 
+    /**
+     * Thread pool for polling data from Http endpoint.
+     */
     private transient ExecutorService pollingThreadPool;
 
+    /**
+     * Thread pool for publishing data to Flink.
+     */
     private transient ExecutorService publishingThreadPool;
 
     @Override
@@ -50,20 +61,21 @@ public void open(FunctionContext context) throws Exception {
 
     public void eval(CompletableFuture<Collection<RowData>> resultFuture, Object... keys) {
 
-        CompletableFuture<RowData> future = new CompletableFuture<>();
+        CompletableFuture<Optional<RowData>> future = new CompletableFuture<>();
         future.completeAsync(() -> decorate.lookupByKeys(keys), pollingThreadPool);
 
         // We don't want to use ForkJoinPool at all. We are using a different thread pool
         // for publishing here intentionally to avoid thread starvation.
         future.whenCompleteAsync(
-            (result, throwable) -> {
+            (optionalResult, throwable) -> {
                 if (throwable != null) {
                     log.error("Exception while processing Http Async request", throwable);
                     resultFuture.completeExceptionally(
                         new RuntimeException("Exception while processing Http Async request",
                             throwable));
                 } else {
-                    resultFuture.complete(Collections.singleton(result));
+                    optionalResult
+                        .ifPresent(result -> resultFuture.complete(Collections.singleton(result)));
                 }
             },
             publishingThreadPool);