ESP-99_ErrorCodes - Error Code configuration from properties. (#16)

* ESP-99_ErrorCodes - Error Code configuration from properties.

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

* ESP-99_ErrorCodes - Error Code configuration from properties.

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

* ESP-99_ErrorCodes - Changes after code review.

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

* ESP-99_ErrorCodes - Changes after code review.

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

Author: kristoffSC

CHANGELOG.md

@@ -5,6 +5,10 @@
 ### Fixed
 - Fix JavaDoc errors.
 
+### Added
+- 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.
+
 ## [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

@@ -8,12 +8,9 @@ The HTTP TableLookup connector that allows for pulling data from external system
 **Note**: The `main` branch may be in an *unstable or even broken state* during development.
 Please use [releases](https://github.com/getindata/flink-http-connector/releases) instead of the `main` branch in order to get a stable set of binaries.
 
-#### HTTP TableLookup Source
 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.
-
-#### HTTP Sink
 `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
@@ -133,21 +130,19 @@ CREATE TABLE http (
 )
 ```
 
-## Implementation
-Implementation of an HTTP source connector is based on Flink's `TableFunction` and `AsyncTableFunction` classes.  
-To be more specific we are using a `LookupTableSource`. Unfortunately Flink's new unified source interface [2] cannot be used for this type of source.
-Issue was discussed on Flink's user mailing list - https://lists.apache.org/thread/tx2w1m15zt5qnvt924mmbvr7s8rlyjmw
+#### 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. 
+By default all 400s and 500s response codes will be interpreted as error code.
 
-Implementation of an HTTP Sink is based on Flink's `AsyncSinkBase` introduced in Flink 1.15 [3, 4].
+This behavior can be changed by using below properties in table definition (DDL) 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.
+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:
+  `401, 402, 403`. In this example, codes 401, 402 and 403 would not be interpreted as error codes.
 
-## 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.
 
 ## Table API Connector Options
 ### HTTP TableLookup Source
@@ -225,6 +220,23 @@ As a result, you should see a table with joined records like so:
 
 The `msg` column shows parameters used with REST call for given JOIN record.
 
+## Implementation
+### HTTP Source
+Implementation of an HTTP source connector is based on Flink's `TableFunction` and `AsyncTableFunction` classes.  
+To be more specific we are using a `LookupTableSource`. Unfortunately Flink's new unified source interface [2] cannot be used for this type of source.
+Issue was discussed on Flink's user mailing list - https://lists.apache.org/thread/tx2w1m15zt5qnvt924mmbvr7s8rlyjmw
+
+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.
+
 ## TODO
 
 ### HTTP TableLookup Source

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

@@ -11,6 +11,8 @@
 @NoArgsConstructor(access = AccessLevel.NONE)
 public final class HttpConnectorConfigConstants {
 
+    public static final String ERROR_CODE_DELIM = ",";
+
     /**
      * A property prefix for http connector.
      */
@@ -21,4 +23,10 @@ public final class HttpConnectorConfigConstants {
      */
     public static final String SINK_HEADER_PREFIX = GID_CONNECTOR_HTTP + "sink.header.";
 
+    // Error code handling configuration.
+    public static final String HTTP_ERROR_CODE_WHITE_LIST =
+        GID_CONNECTOR_HTTP + "sink.error.code.exclude";
+
+    public static final String HTTP_ERROR_CODES_LIST = GID_CONNECTOR_HTTP + "sink.error.code";
+
 }

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

@@ -21,6 +21,8 @@
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
 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.utils.ConfigUtils;
 import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.SINK_HEADER_PREFIX;
 
@@ -35,6 +37,8 @@ public class JavaNetSinkHttpClient implements SinkHttpClient {
 
     private final String[] headersAndValues;
 
+    private final HttpStatusCodeChecker statusCodeChecker;
+
     public JavaNetSinkHttpClient(Properties properties) {
         this.httpClient = HttpClient.newBuilder()
             .followRedirects(HttpClient.Redirect.NORMAL)
@@ -44,6 +48,10 @@ public JavaNetSinkHttpClient(Properties properties) {
             ConfigUtils.propertiesToMap(properties, SINK_HEADER_PREFIX, String.class);
 
         this.headersAndValues = ConfigUtils.toHeaderAndValueArray(headerMap);
+
+        // 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);
     }
 
     @Override
@@ -99,8 +107,9 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse(
 
         for (var response : responses) {
             var sinkRequestEntry = response.getSinkRequestEntry();
+            // TODO Add response processor here and orchestrate it with statusCodeChecker.
             if (response.getResponse().isEmpty()
-                || response.getResponse().get().statusCode() >= 400) {
+                || statusCodeChecker.isErrorCode(response.getResponse().get().statusCode())) {
                 failedResponses.add(sinkRequestEntry);
             } else {
                 successfulResponses.add(sinkRequestEntry);

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

@@ -0,0 +1,135 @@
+package com.getindata.connectors.http.internal.sink.httpclient.status;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+
+import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
+
+/**
+ * An implementation of {@link HttpStatusCodeChecker} that checks Http Status code against
+ * white list, concrete value or {@link HttpResponseCodeType}
+ */
+public class ComposeHttpStatusCodeChecker implements HttpStatusCodeChecker {
+
+    private static final Set<HttpStatusCodeChecker> DEFAULT_ERROR_CODES =
+        Set.of(
+            new TypeStatusCodeChecker(HttpResponseCodeType.CLIENT_ERROR),
+            new TypeStatusCodeChecker(HttpResponseCodeType.SERVER_ERROR)
+        );
+
+    private static final int MIN_HTTP_STATUS_CODE = 100;
+
+    /**
+     * Set of {@link HttpStatusCodeChecker} for white listed status codes.
+     */
+    private final Set<WhiteListHttpStatusCodeChecker> excludedCodes;
+
+    /**
+     * Set of {@link HttpStatusCodeChecker} that check status code againts value match or {@link
+     * HttpResponseCodeType} match.
+     */
+    private final Set<HttpStatusCodeChecker> errorCodes;
+
+    public ComposeHttpStatusCodeChecker(Properties properties) {
+        excludedCodes = prepareWhiteList(properties);
+        errorCodes = prepareErrorCodes(properties);
+    }
+
+    /**
+     * Checks whether given status code is considered as a error code.
+     * This implementation checks if status code matches any single value mask like "404"
+     * or http type mask such as "4XX". Code that matches one of those masks and is not on a
+     * white list will be considered as error code.
+     * @param statusCode http status code to assess.
+     * @return true if status code is considered as error or false if not.
+     */
+    public boolean isErrorCode(int statusCode) {
+
+        Preconditions.checkArgument(
+            statusCode >= MIN_HTTP_STATUS_CODE,
+            String.format(
+                "Provided invalid Http status code %s,"
+                    + " status code should be equal or bigger than %d.",
+                statusCode,
+                MIN_HTTP_STATUS_CODE)
+        );
+
+        boolean isWhiteListed = excludedCodes.stream()
+            .anyMatch(check -> check.isWhiteListed(statusCode));
+
+        return !isWhiteListed
+            && errorCodes.stream()
+                .anyMatch(httpStatusCodeChecker -> httpStatusCodeChecker.isErrorCode(statusCode));
+    }
+
+    private Set<HttpStatusCodeChecker> prepareErrorCodes(Properties properties) {
+        String sErrorCodes =
+            properties.getProperty(HttpConnectorConfigConstants.HTTP_ERROR_CODES_LIST, "");
+
+        if (StringUtils.isNullOrWhitespaceOnly(sErrorCodes)) {
+            return DEFAULT_ERROR_CODES;
+        } else {
+            String[] splitCodes = sErrorCodes.split(HttpConnectorConfigConstants.ERROR_CODE_DELIM);
+            return prepareErrorCodes(splitCodes);
+        }
+    }
+
+    /**
+     * Process given array of status codes and assign them to
+     * {@link SingleValueHttpStatusCodeChecker} for full codes such as 100, 404 etc. or to
+     * {@link TypeStatusCodeChecker} for codes that were constructed with "XX" mask
+     */
+    private Set<HttpStatusCodeChecker> prepareErrorCodes(String[] statusCodes) {
+
+        Set<HttpStatusCodeChecker> errorCodes = new HashSet<>();
+        for (String sCode : statusCodes) {
+            if (!StringUtils.isNullOrWhitespaceOnly(sCode)) {
+                String trimCode = sCode.toUpperCase().trim();
+                Preconditions.checkArgument(
+                    trimCode.length() == 3,
+                    "Status code should contain three characters. Provided [%s]",
+                    trimCode);
+
+                // at this point we have trim, upper case 3 character status code.
+                if (isTypeCode(trimCode)) {
+                    int code = Integer.parseInt(trimCode.replace("X", ""));
+                    errorCodes.add(new TypeStatusCodeChecker(HttpResponseCodeType.getByCode(code)));
+                } else {
+                    errorCodes.add(
+                        new SingleValueHttpStatusCodeChecker(Integer.parseInt(trimCode))
+                    );
+                }
+            }
+        }
+        return (errorCodes.isEmpty()) ? DEFAULT_ERROR_CODES : errorCodes;
+    }
+
+    private Set<WhiteListHttpStatusCodeChecker> prepareWhiteList(Properties properties) {
+        return Arrays.stream(
+                properties.getProperty(HttpConnectorConfigConstants.HTTP_ERROR_CODE_WHITE_LIST, "")
+                    .split(HttpConnectorConfigConstants.ERROR_CODE_DELIM))
+            .filter(sCode -> !StringUtils.isNullOrWhitespaceOnly(sCode))
+            .map(String::trim)
+            .mapToInt(Integer::parseInt)
+            .mapToObj(WhiteListHttpStatusCodeChecker::new)
+            .collect(Collectors.toSet());
+    }
+
+    /**
+     * This method checks if "code" param matches "digit + XX" mask. This method expects that
+     * provided string will be 3 elements long, trim and upper case.
+     *
+     * @param code to check if it contains XX on second ant third position. Parameter is expected to
+     *             be 3 characters long, trim and uppercase.
+     * @return true if string matches "anything + XX" and false if not.
+     */
+    private boolean isTypeCode(final String code) {
+        return code.charAt(1) == 'X' && code.charAt(2) == 'X';
+    }
+}

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

@@ -0,0 +1,49 @@
+package com.getindata.connectors.http.internal.sink.httpclient.status;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * This enum represents HTTP response code types, grouped by "hundreds" digit.
+ */
+public enum HttpResponseCodeType {
+
+    INFO(1),
+    SUCCESS(2),
+    REDIRECTION(3),
+    CLIENT_ERROR(4),
+    SERVER_ERROR(5);
+
+    private static final Map<Integer, HttpResponseCodeType> map;
+
+    static {
+        map = new HashMap<>();
+        for (HttpResponseCodeType httpResponseCodeType : HttpResponseCodeType.values()) {
+            map.put(httpResponseCodeType.httpTypeCode, httpResponseCodeType);
+        }
+    }
+
+    private final int httpTypeCode;
+
+    HttpResponseCodeType(int httpTypeCode) {
+        this.httpTypeCode = httpTypeCode;
+    }
+
+    /**
+     * @param statusCode Http status code to get the {@link HttpResponseCodeType} instance for.
+     * @return a {@link HttpResponseCodeType} instance based on http type code, for example {@code
+     * HttpResponseCodeType.getByCode(1)} will return {@link HttpResponseCodeType#INFO} type.
+     */
+    public static HttpResponseCodeType getByCode(int statusCode) {
+        return map.get(statusCode);
+    }
+
+    /**
+     * @return a "hundreds" digit that represents given {@link HttpResponseCodeType} instance.
+     * For example {@code HttpResponseCodeType.INFO.getHttpTypeCode()} will return 1 since HTTP
+     * information repossess have status codes in range 100 - 199.
+     */
+    public int getHttpTypeCode() {
+        return this.httpTypeCode;
+    }
+}

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

@@ -0,0 +1,16 @@
+package com.getindata.connectors.http.internal.sink.httpclient.status;
+
+/**
+ * Base interface for all classes that would validate HTTP status
+ * code whether it is an error or not.
+ */
+public interface HttpStatusCodeChecker {
+
+    /**
+     * Validates http status code wheter it is considered as error code. The logic for
+     * what status codes are considered as "errors" depends on the concreted implementation
+     * @param statusCode http status code to assess.
+     * @return true if statusCode is considered as Error and false if not.
+     */
+    boolean isErrorCode(int statusCode);
+}

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

@@ -0,0 +1,28 @@
+package com.getindata.connectors.http.internal.sink.httpclient.status;
+
+import lombok.EqualsAndHashCode;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * An implementation of {@link HttpStatusCodeChecker} that validates status code against
+ * constant value.
+ */
+@RequiredArgsConstructor
+@EqualsAndHashCode
+public class SingleValueHttpStatusCodeChecker implements HttpStatusCodeChecker {
+
+    /**
+     * A reference http status code to compare with.
+     */
+    private final int errorCode;
+
+    /**
+     * Validates given statusCode against constant value.
+     * @param statusCode http status code to assess.
+     * @return true if status code is considered as error or false if not.
+     */
+    @Override
+    public boolean isErrorCode(int statusCode) {
+        return errorCode == statusCode;
+    }
+}