For non retryable API call failures on a lookup join, allow the error to be in the output in the stream

[davidradl]

Requirement
The original requirement was to expose error information include error messages and error http status codes into the flow. We then extended the scope to include HTTP Status codes, headers which could be useful for non-error cases. When we prototyped that, we found that we needed another metadata column completion state, to indicate whether the http call was successful or not, as the http status code could have values in success and failure scenarios.
A side effect of this feature is that it will be useful in debugging.

Proposed design

A new config option CONTINUE_ON_ERROR which is a boolean that defaults to false. When true the job will not end, but if requests fails then the job will continue with extra information in any metadata columns that have been defined. The lookup join values from the lookup table (the table defined with the HTTP connector) will be null for nullable files, or the default for non nullable fields.

Metadata keys (read only so they cannot be inserted into)
ERROR-MESSAGE string - message from the client side exception or the response body for failed HTTP requests. This only has a value in failure situations
HTTP-HEADERS - map<STRING, LIST<>> http headers
HTTP-STATUS-CODE - integer - http status code
HTTP-COMPLETION-STATE - information about how the last http call completed. The string will have values from an enum:
SUCCESS - HTTP call succeeded
HTTP_ERROR_STATUS - HTTP request failed with a http status code
EXCEPTION - there was an Exception and no HTTP status code.

Notes:
If you are using the Table API TableResult and have an await with a timeout, this Timeout exception will cause the job to terminate, even if there are metadata columns defined.
The config flag determines the behaviour.
The metadata columns are populated when there is something meaningful to put in them

A new java bean is added

public class HttpRowDataWrapper {
    private final Collection<RowData> data;
    private final String errorMessage;
    private final Map<String, List<String>> httpHeadersMap;
    private final Integer httpStatusCode;
    private final HttpCompletionState httpCompletionState;
} 

Allowing the lower HTTP layers to communicate up the http content to the lookup code.
Here is a picture of its lifecycle.

Rejected other designs:

1. Have a configuration boolean fail_on_error, that defaults to true, when true HTTP errors (potentially after retry processing) result in exceptions that end the job. When false. New metadata columns will be populated,
   ERROR_MESSAGE string
   HEADERS
   ERROR_STATUS_CODE
2. We then thought if this is just happening in error cases , then we do not need the config flag, as we can populate the metadata columns if they are there
3. We then extended the scope to include non error cases, so the metadata fields are:
   ERROR-MESSAGE string
   HTTP-HEADERS
   HTTP-STATUS-CODE
   And realised that we needed a configuration option again.
   This also involved amending the lower level pull method to return not just the Collection but also information from the HTTP Response.
4. Added metadata key COMPLETION_STATE for more information
   Metadata keys (read only so they cannot be inserted into)
   ERROR-MESSAGE string - message from the client side exception or the response body for failed HTTP requests. This only has a value in failure situations
   HTTP-HEADERS - map<STRING, LIST<>> http headers
   HTTP-STATUS-CODE - integer - http status code
   HTTP-COMPLETION-STATE information about how the http call ended The string will have values from an enum:
   HTTP_ERROR_STATUS - the http call failed after retrying
   EXCEPTION - error occurred but there wa no http status code
   CLIENT_SIDE_EXCEPTION - no HTTP call completed, and there was an Exception

The problem with this is that we cannot determine HTTP_FAILED_AFTER_RETRY as this happens in the HTTP Client with a Retry utility.
Also CONTINUE_ON_ERROR_INFO is dependent on the CONTINUE_ON_ERROR config being set. Ut would seem more intuitive to just have the http completion state and include SUCCESS
Also if we fail while deserialising the http bad response , we get an IOException - I am not sure if we think of this as a client side Exception - I think Exception is clearer for this field.

Background- this was the original text with which this issue was raised:

We are thinking of a solution like this :

create table api_lookup ( 
  `customer_id` STRING NOT NULL,
  `customer_records_from_backend` STRING,
  `errorMessage` STRING METADATA FROM `error-_string`,
  `responseCode` INT METADATA FROM `error_code`)
WITH (   
  `connector` = 'rest_lookup',
  ...
)

The sample above shows us defining two METADATA fields 'error_string' and 'error_code' which provide a way to surface this data from the connector back to SQL. Allowing the flow to be able to act on this error content driving the appropriate error logic without having to come out of the stream to look at logs, open telemetry etc.

The request values would be in the output, but the response columns (from the lookup table) would be nulls.

The docs describe only how errors can cause retries up to a limit before failing the job. A new gid.connector.http.source.lookup. might be required in order to enable the 'fail and produce an error message' behaviour as opposed to the currently documented 'fail and retry or crash'. When there is an error of this type then the response columns would be null.

Investigation to see if this is feasible

Would the lookup join code path call SupportsReadingMetadata (which is what Kafka uses).
flink has

 public interface SupportsReadingMetadata {
    Map<String, DataType> listReadableMetadata();

    void applyReadableMetadata(List<String> var1, DataType var2);

    default boolean supportsMetadataProjection() {
        return true;
    }

If this was driven at runtime in the lookup join case then the metadata should come through.

[davidradl]

@grzegorz8 @kzarzycki WDYT?

[grzegorz8]

@davidradl I think adding metadata fields is a great idea — thanks for proposing it!

We could even go a step further and include fields that provide more detailed context, such as:

• url
• method
• request_headers
• status_code (instead of error_code)
• error_message (instead of error_string) - set when request failed
• response_headers

[davidradl]

@grzegorz8 we were keeping the fields non http specific so that client side errors could be handled as well. Our use case is around surfacing errors in the flow so the flow can handle them. Do you have a use case to surface the http info? At the same time we are open to extensions to this idea if we need to.

[grzegorz8]

we were keeping the fields non http specific so that client side errors could be handled as well.

I see. We can start with only these two metadata fields. But I suggest to rename error_code to status_code.

Please correct me if I'm wrong:

• if request failed due to client/server error ->status_code = 4XX or 5XX, error_string = body of the response.
• If request failed due to other error (e.g. timeout, dns resolution error) -> status_code is empty, error_string = root cause error message?

Do you have a use case to surface the http info?

Currently, I do not. I just think it may be helpful when debugging. You can quickly see url or request headers of the request. I tried to follow kafka connector approach which returns topic, partition, offset, etc.

[davidradl]

I have a prototype .

I am adding the metadata columns in the lookup function if we get back an error status code, it works for synchronous lookups (for success , server side error and client side error), but it is failing for asynchronous server side errors. They fail in the Flink table planner as the runtime async processing is not aware of the extra metadata columns (that I added in the lookup function) so there is an index out of bounds error.

@grzegorz8 @kristoffSC any thoughts? This seems to be a blocker.

[grzegorz8]

@davidradl Could you provide more details, for example stacktrace? Does it fail on specific unit or integration test?

[davidradl]

I have got round this issue now . Working through updating and adding unit tests now

[davidradl]

Hi , we are now considering exposing the headers and status code for non error cases. This means updating public interface PollingClient so that the pull will return a wrapper object containing the collection of RowData and the httpresponse object.
like this

@Data
public class RowDataWithHttpResponse {
    private final Collection<RowData> data;
    private final HttpResponse<String> response;
}

The lookup function can then get the headers out of the http response.

public interface PollingClient<T> {

    /**
     * Gets enrichment data from external component using provided lookup arguments.
     * @param lookupRow A {@link RowData} containing request parameters.
     * @return an optional result of data lookup.
     */
    Collection<T> pull(RowData lookupRow);

    /**
     * Initialize the client.
     * @param ctx function context
     */
    void open(FunctionContext ctx);
}

would change to

public interface PollingClient {

    /**
     * Gets enrichment data from external component using provided lookup arguments.
     * @param lookupRow A {@link RowData} containing request parameters.
     * @return an optional result of data lookup.
     */
    RowDataWithHttpResponse pull(RowData lookupRow);

    /**
     * Initialize the client.
     * @param ctx function context
     */
    void open(FunctionContext ctx);
}

@kristoffSC and @grzegorz8 are you ok with this ?

[davidradl]

Thinking about it, we could include header information as a map and the message and status as string and int, if we did not want to expose the http response out side of this layer.

Also we could add an overloaded pull method if you think anyone is using the existing method.

But will leave as is, if you do not care about these considerations

[grzegorz8]

Hey @davidradl ! I think your proposal to change pull() return type looks fine 👍
I'm just wondering:

• does this change propagate to any higher level interfaces?
• At some point we need to convert RowDataWithHttpResponse to RowData which contains metadata fields (LookupFunction#lookup() returns Collection<RowData>). Do you see any major challenges with this?

Also we could add an overloaded pull method if you think anyone is using the existing method.

I think this is an internal interface, in contrast to e.g. LookupQueryCreator, so we can modify it without any backward compatibility.

Thinking about it, we could include header information as a map and the message and status as string and int, if we did not want to expose the http response out side of this layer.

I don’t have a strong opinion on this, but I’m leaning toward returning our own object instead of an HttpResponse.

[davidradl]

@grzegorz8 great . I will go ahead with :

@Data
public class HttpRowDataWrapper {
    private final Collection<RowData> data;
    private final Map<String, List<String>> http_headers_map;
    private final String http_error_message;
    private final int http_status_code; 
}

Let me know if you need it adjusting.

[grzegorz8]

@davidradl Except fields naming convention (camelCase preferred), HttpRowDataWrapper looks good.

[davidradl]

@grzegorz8 .
I am thinking we should have all the new fields as optional. There is a case to say we could have an optional data as well, but the existing code handles this with an empty list - so I think this is sufficient.

public class HttpRowDataWrapper {
    private final Collection<RowData> data;
    private final Optional<Map<String, List<String>>> httpHeadersMap;
    private final Optional<Integer> httpStatusCode;
}

The error message is from the Exception so not from the RowData.

[kristoffSC]

@grzegorz8 . I am thinking we should have all the new fields as optional. There is a case to say we could have an optional data as well, but the existing code handles this with an empty list - so I think this is sufficient.

public class HttpRowDataWrapper {
    private final Collection<RowData> data;
    private final Optional<Map<String, List<String>>> httpHeadersMap;
    private final Optional<Integer> httpStatusCode;
}

The error message is from the Exception so not from the RowData.

Hi there,
IMO using Java's Optional type as class level fields or method arguments its s code smell.

Optionals were ment for return type, instead returning null (in short).

So instead declaring field as optional, I would recommended getter that is returning an Optional.