ESP-171_HttpLookupSource_ErrorCodeSupport - HTTP ErrorCode Status config for Lookup Source (#22)

* ESP-171_HttpLookupSource_ErrorCodeSupport - Adding to Lookup HTTP Source HTTP Error Code configuration from properties.

Signed-off-by: Krzysztof Chmielewski <[email protected]>

* Update CHANGELOG.md

Co-authored-by: Andrzej Swatowski <[email protected]>

Signed-off-by: Krzysztof Chmielewski <[email protected]>
Co-authored-by: Andrzej Swatowski <[email protected]>

Author: kristoffSC

CHANGELOG.md

@@ -6,8 +6,9 @@
 - Fix JavaDoc errors.
 
 ### Added
-- 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.
+- Add new properties `gid.connector.http.sink.error.code`,`gid.connector.http.sink.error.code.exclude`,
+  `gid.connector.http.source.lookup.error.code` and `gid.connector.http.source.lookup.error.code.exclude`
+  to set HTTP status codes that should be interpreted as errors both for HTTP Sink and HTTP Lookup Source.
 - Use Flink's format support to Http Lookup Source. 
 - Add HTTP Lookup source client header configuration via properties.
 

README.md

@@ -184,17 +184,18 @@ CREATE TABLE http (
 )
 ```
 
-#### HTTP status code handler (currently supported only for HTTP Sink)
-Http Sink connector allows defining list of HTTP status codes that should be treated as errors. 
+## HTTP status code handler
+Http Sink and Lookup Source connectors allow defining list of HTTP status codes that should be treated as errors. 
 By default all 400s and 500s response codes will be interpreted as error code.
 
-This behavior can be changed by using below properties in table definition (DDL) or passing it via 
+This behavior can be changed by using below properties in table definition (DDL) for Sink and Lookup Source or passing it via 
 `setProperty' method from Sink's builder. The property names are:
-- `gid.connector.http.sink.error.code` used to defined HTTP status code value that should be treated as error for example 404.
+- `gid.connector.http.sink.error.code` and `gid.connector.http.source.lookup.error.code` used to defined HTTP status code value that should be treated as error for example 404.
 Many status codes can be defined in one value, where each code should be separated with comma, for example:
 `401, 402, 403`. User can use this property also to define a type code mask. In that case, all codes from given HTTP response type will be treated as errors.
 An example of such a mask would be `3XX, 4XX, 5XX`. In this case, all 300s, 400s and 500s status codes will be treated as errors.
-- `gid.connector.http.sink.error.code.exclude` used to exclude a HTTP code from error list. Many status codes can be defined in one value, where each code should be separated with comma, for example:
+- `gid.connector.http.sink.error.code.exclude` and `gid.connector.http.source.lookup.error.code.exclude` used to exclude a HTTP code from error list.
+   Many status codes can be defined in one value, where each code should be separated with comma, for example:
   `401, 402, 403`. In this example, codes 401, 402 and 403 would not be interpreted as error codes.
 
 
@@ -291,21 +292,13 @@ Issue was discussed on Flink's user mailing list - https://lists.apache.org/thre
 Implementation of an HTTP Sink is based on Flink's `AsyncSinkBase` introduced in Flink 1.15 [3, 4].
 
 #### Http Response to Table schema mapping
-The mapping from Http Json Response to SQL table schema is done via Json Paths [5].
-This is achieved thanks to `com.jayway.jsonpath:json-path` library.
-
-If no `root` or `field.#.path` option is defined, the connector will use the column name as json path and will try to look for Json Node with that name in received Json. If no node with a given name is found, the connector will return `null` as value for this field.
-
-If the `field.#.path` option is defined, connector will use given Json path from option's value in order to find Json data that should be used for this column.
-For example `'field.isActive.path' = '$.details.isActive'` - the value for table column `isActive` will be taken from `$.details.isActive` node from received Json.
+The mapping from Http Json Response to SQL table schema is done via Flink's Json Format [5].
 
 ## TODO
 
 ### HTTP TableLookup Source
 - Implement caches.
-- Add support for other Flink types. Currently, STRING type is only fully supported.
 - Think about Retry Policy for Http Request
-- Use Flink Format [7] to parse Json response 
 - Add Configurable Timeout value
 - Check other `//TODO`'s.
 
@@ -321,9 +314,7 @@ For example `'field.isActive.path' = '$.details.isActive'` - the value for table
 </br>
 [4] https://nightlies.apache.org/flink/flink-docs-release-1.15/api/java/org/apache/flink/connector/base/sink/AsyncSinkBase.html
 </br>
-[5] https://support.smartbear.com/alertsite/docs/monitors/api/endpoint/jsonpath.html
+[5] https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/
 </br>
 [6] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sqlclient/
 </br>
-[7] https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/
-</br>

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

@@ -26,10 +26,18 @@ public final class HttpConnectorConfigConstants {
     public static final String LOOKUP_SOURCE_HEADER_PREFIX = GID_CONNECTOR_HTTP
         + "source.lookup.header.";
 
-    // Error code handling configuration.
-    public static final String HTTP_ERROR_CODE_WHITE_LIST =
+    // --------- Error code handling configuration ---------
+    public static final String HTTP_ERROR_SINK_CODE_WHITE_LIST =
         GID_CONNECTOR_HTTP + "sink.error.code.exclude";
 
-    public static final String HTTP_ERROR_CODES_LIST = GID_CONNECTOR_HTTP + "sink.error.code";
+    public static final String HTTP_ERROR_SINK_CODES_LIST = GID_CONNECTOR_HTTP + "sink.error.code";
+
+    public static final String HTTP_ERROR_SOURCE_LOOKUP_CODE_WHITE_LIST =
+        GID_CONNECTOR_HTTP + "source.lookup.error.code.exclude";
+
+    public static final String HTTP_ERROR_SOURCE_LOOKUP_CODES_LIST =
+        GID_CONNECTOR_HTTP + "source.lookup.error.code";
+
+    // -----------------------------------------------------
 
 }

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClient.java

@@ -20,9 +20,11 @@
 
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
+import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
-import com.getindata.connectors.http.internal.sink.httpclient.status.ComposeHttpStatusCodeChecker;
-import com.getindata.connectors.http.internal.sink.httpclient.status.HttpStatusCodeChecker;
+import com.getindata.connectors.http.internal.status.ComposeHttpStatusCodeChecker;
+import com.getindata.connectors.http.internal.status.ComposeHttpStatusCodeChecker.ComposeHttpStatusCodeCheckerConfig;
+import com.getindata.connectors.http.internal.status.HttpStatusCodeChecker;
 import com.getindata.connectors.http.internal.utils.ConfigUtils;
 import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.SINK_HEADER_PREFIX;
 
@@ -51,7 +53,14 @@ public JavaNetSinkHttpClient(Properties properties) {
 
         // TODO Inject this via constructor when implementing a response processor.
         //  Processor will be injected and it will wrap statusChecker implementation.
-        this.statusCodeChecker = new ComposeHttpStatusCodeChecker(properties);
+        ComposeHttpStatusCodeCheckerConfig checkerConfig =
+            ComposeHttpStatusCodeCheckerConfig.builder()
+                .properties(properties)
+                .whiteListPrefix(HttpConnectorConfigConstants.HTTP_ERROR_SINK_CODE_WHITE_LIST)
+                .errorCodePrefix(HttpConnectorConfigConstants.HTTP_ERROR_SINK_CODES_LIST)
+                .build();
+
+        this.statusCodeChecker = new ComposeHttpStatusCodeChecker(checkerConfig);
     }
 
     @Override

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/status/ComposeHttpStatusCodeChecker.java renamed to src/main/java/com/getindata/connectors/http/internal/status/ComposeHttpStatusCodeChecker.java

@@ -1,11 +1,15 @@
-package com.getindata.connectors.http.internal.sink.httpclient.status;
+package com.getindata.connectors.http.internal.status;
 
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import lombok.AccessLevel;
+import lombok.Builder;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.StringUtils;
 
@@ -36,9 +40,9 @@ public class ComposeHttpStatusCodeChecker implements HttpStatusCodeChecker {
      */
     private final Set<HttpStatusCodeChecker> errorCodes;
 
-    public ComposeHttpStatusCodeChecker(Properties properties) {
-        excludedCodes = prepareWhiteList(properties);
-        errorCodes = prepareErrorCodes(properties);
+    public ComposeHttpStatusCodeChecker(ComposeHttpStatusCodeCheckerConfig config) {
+        excludedCodes = prepareWhiteList(config);
+        errorCodes = prepareErrorCodes(config);
     }
 
     /**
@@ -68,14 +72,19 @@ public boolean isErrorCode(int statusCode) {
                 .anyMatch(httpStatusCodeChecker -> httpStatusCodeChecker.isErrorCode(statusCode));
     }
 
-    private Set<HttpStatusCodeChecker> prepareErrorCodes(Properties properties) {
-        String sErrorCodes =
-            properties.getProperty(HttpConnectorConfigConstants.HTTP_ERROR_CODES_LIST, "");
+    private Set<HttpStatusCodeChecker> prepareErrorCodes(
+            ComposeHttpStatusCodeCheckerConfig config) {
 
-        if (StringUtils.isNullOrWhitespaceOnly(sErrorCodes)) {
+        Properties properties = config.getProperties();
+        String errorCodePrefix = config.getErrorCodePrefix();
+
+        String errorCodes =
+            properties.getProperty(errorCodePrefix, "");
+
+        if (StringUtils.isNullOrWhitespaceOnly(errorCodes)) {
             return DEFAULT_ERROR_CODES;
         } else {
-            String[] splitCodes = sErrorCodes.split(HttpConnectorConfigConstants.ERROR_CODE_DELIM);
+            String[] splitCodes = errorCodes.split(HttpConnectorConfigConstants.ERROR_CODE_DELIM);
             return prepareErrorCodes(splitCodes);
         }
     }
@@ -110,9 +119,14 @@ private Set<HttpStatusCodeChecker> prepareErrorCodes(String[] statusCodes) {
         return (errorCodes.isEmpty()) ? DEFAULT_ERROR_CODES : errorCodes;
     }
 
-    private Set<WhiteListHttpStatusCodeChecker> prepareWhiteList(Properties properties) {
+    private Set<WhiteListHttpStatusCodeChecker> prepareWhiteList(
+            ComposeHttpStatusCodeCheckerConfig config) {
+
+        Properties properties = config.getProperties();
+        String whiteListPrefix = config.getWhiteListPrefix();
+
         return Arrays.stream(
-                properties.getProperty(HttpConnectorConfigConstants.HTTP_ERROR_CODE_WHITE_LIST, "")
+                properties.getProperty(whiteListPrefix, "")
                     .split(HttpConnectorConfigConstants.ERROR_CODE_DELIM))
             .filter(sCode -> !StringUtils.isNullOrWhitespaceOnly(sCode))
             .map(String::trim)
@@ -132,4 +146,16 @@ private Set<WhiteListHttpStatusCodeChecker> prepareWhiteList(Properties properti
     private boolean isTypeCode(final String code) {
         return code.charAt(1) == 'X' && code.charAt(2) == 'X';
     }
+
+    @Data
+    @Builder
+    @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
+    public static class ComposeHttpStatusCodeCheckerConfig {
+
+        private final String whiteListPrefix;
+
+        private final String errorCodePrefix;
+
+        private final Properties properties;
+    }
 }

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/status/HttpResponseCodeType.java renamed to src/main/java/com/getindata/connectors/http/internal/status/HttpResponseCodeType.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http.internal.sink.httpclient.status;
+package com.getindata.connectors.http.internal.status;
 
 import java.util.HashMap;
 import java.util.Map;

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/status/HttpStatusCodeChecker.java renamed to src/main/java/com/getindata/connectors/http/internal/status/HttpStatusCodeChecker.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http.internal.sink.httpclient.status;
+package com.getindata.connectors.http.internal.status;
 
 /**
  * Base interface for all classes that would validate HTTP status

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/status/SingleValueHttpStatusCodeChecker.java renamed to src/main/java/com/getindata/connectors/http/internal/status/SingleValueHttpStatusCodeChecker.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http.internal.sink.httpclient.status;
+package com.getindata.connectors.http.internal.status;
 
 import lombok.EqualsAndHashCode;
 import lombok.RequiredArgsConstructor;

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/status/TypeStatusCodeChecker.java renamed to src/main/java/com/getindata/connectors/http/internal/status/TypeStatusCodeChecker.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http.internal.sink.httpclient.status;
+package com.getindata.connectors.http.internal.status;
 
 import lombok.EqualsAndHashCode;
 

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/status/WhiteListHttpStatusCodeChecker.java renamed to src/main/java/com/getindata/connectors/http/internal/status/WhiteListHttpStatusCodeChecker.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http.internal.sink.httpclient.status;
+package com.getindata.connectors.http.internal.status;
 
 import lombok.EqualsAndHashCode;
 import lombok.RequiredArgsConstructor;