[Flink-38864] Add ignore status completion state

docs/content.zh/docs/connectors/table/http.md

@@ -54,7 +54,7 @@ The HTTP source connector supports [Lookup Joins](https://nightlies.apache.org/f
     * [Timeouts](#timeouts)
     * [Source table HTTP status code](#source-table-http-status-code)
     * [Retries and handling errors (Lookup source)](#retries-and-handling-errors-lookup-source)
-        * [Retry strategy](#retry-strategy)
+      * [Retry strategy](#retry-strategy)
       * [Lookup multiple results](#lookup-multiple-results)
   * [Working with HTTP sink tables](#working-with-http-sink-tables)
     * [HTTP Sink](#http-sink)
@@ -71,7 +71,7 @@ The HTTP source connector supports [Lookup Joins](https://nightlies.apache.org/f
     * [Basic Authentication](#basic-authentication)
     * [OIDC Bearer Authentication](#oidc-bearer-authentication)
   * [Logging the HTTP content](#logging-the-http-content)
-      * [Restrictions at this time](#restrictions-at-this-time)
+    * [Restrictions at this time](#restrictions-at-this-time)
 <!-- TOC -->
 ## Dependencies
 
@@ -503,6 +503,10 @@ this means that these columns will be null for nullable columns and hold a defau
 
 When using `http.source.lookup.continue-on-error` as true, consider adding extra metadata columns that will surface information about failures into your stream.
 
+Note that if Metadata columns are specified and the status code is ignored, then a row containing metadata columns will be produced. If
+the status code is ignored and there are no metadata columns defined, then no row will be emitted; this ensures that the expected
+inner join behaviour still occurs.
+
 Metadata columns can be specified and hold http information. They are optional read-only columns that must be declared VIRTUAL to exclude them during an INSERT INTO operation.
 
 | Key                   | Data Type                        | Description                            |
@@ -520,10 +524,14 @@ Metadata columns can be specified and hold http information. They are optional r
 | HTTP_ERROR_STATUS              | HTTP error status code              |
 | EXCEPTION                      | An Exception occurred               |
 | UNABLE_TO_DESERIALIZE_RESPONSE | Unable to deserialize HTTP response |
+| IGNORE_STATUS_CODE             | Status code was ignored             |
 
 If the `error-string` metadata column is defined on the table and the call succeeds then it will have a null value.
 When the HTTP response cannot be deserialized, then the `http-completion-state` will be `UNABLE_TO_DESERIALIZE_RESPONSE`
 and the `error-string` will be the response body.
+When the HTTP status code is in the `http.source.lookup.ignored-response-codes`, then the `http-completion-state` will
+be `IGNORE_STATUS_CODE`and no data is returned; any metadata columns contain information about the API call that
+occurred.
 
 When a HTTP lookup call fails and populates the metadata columns with the error information, the expected enrichment columns from the HTTP call
 are not populated, this means that they will be null for nullable columns and hold a default value for the type for non-nullable columns.

docs/content/docs/connectors/table/http.md

@@ -503,6 +503,10 @@ this means that these columns will be null for nullable columns and hold a defau
 
 When using `http.source.lookup.continue-on-error` as true, consider adding extra metadata columns that will surface information about failures into your stream.
 
+Note that if Metadata columns are specified and the status code is ignored, then a row containing metadata columns will be produced. If
+the status code is ignored and there are no metadata columns defined, then no row will be emitted; this ensures that the expected
+inner join behaviour still occurs.
+
 Metadata columns can be specified and hold http information. They are optional read-only columns that must be declared VIRTUAL to exclude them during an INSERT INTO operation.
 
 | Key                   | Data Type                        | Description                            |
@@ -520,10 +524,14 @@ Metadata columns can be specified and hold http information. They are optional r
 | HTTP_ERROR_STATUS              | HTTP error status code              |
 | EXCEPTION                      | An Exception occurred               |
 | UNABLE_TO_DESERIALIZE_RESPONSE | Unable to deserialize HTTP response |
+| IGNORE_STATUS_CODE             | Status code was ignored             |
 
 If the `error-string` metadata column is defined on the table and the call succeeds then it will have a null value.
 When the HTTP response cannot be deserialized, then the `http-completion-state` will be `UNABLE_TO_DESERIALIZE_RESPONSE`
 and the `error-string` will be the response body.
+When the HTTP status code is in the `http.source.lookup.ignored-response-codes`, then the `http-completion-state` will 
+be `IGNORE_STATUS_CODE`and no data is returned; any metadata columns contain information about the API call that 
+occurred.
 
 When a HTTP lookup call fails and populates the metadata columns with the error information, the expected enrichment columns from the HTTP call
 are not populated, this means that they will be null for nullable columns and hold a default value for the type for non-nullable columns.

flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpCompletionState.java

@@ -22,5 +22,6 @@ public enum HttpCompletionState {
     HTTP_ERROR_STATUS,
     EXCEPTION,
     SUCCESS,
-    UNABLE_TO_DESERIALIZE_RESPONSE
+    UNABLE_TO_DESERIALIZE_RESPONSE,
+    IGNORE_STATUS_CODE
 }

flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java

@@ -104,9 +104,6 @@ public Collection<RowData> lookup(RowData keyRow) {
         int physicalArity = -1;
 
         GenericRowData producedRow = null;
-        if (httpRowDataWrapper.shouldIgnore()) {
-            return Collections.emptyList();
-        }
         // grab the actual data if there is any from the response and populate the producedRow with
         // it
         if (!httpCollector.isEmpty()) {

flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java

@@ -32,6 +32,7 @@
 import org.apache.flink.connector.http.utils.HttpHeaderUtils;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.util.Collector;
 import org.apache.flink.util.ConfigurationException;
 import org.apache.flink.util.StringUtils;
 
@@ -123,6 +124,13 @@ public void open(FunctionContext context) {
     @Override
     public HttpRowDataWrapper pull(RowData lookupRow) {
         if (lookupRow == null) {
+            /*
+             * We are not sure if the following code can be driven. Tested with an equality of booleans (which should
+             * be a filter), but with the latest flink this is rejected by the planner.
+             *
+             * If there is a way for lookupRow to be null here, then the results will not populate any metadata fields
+             * and we should add a new completion state to identify this scenario.
+             */
             return HttpRowDataWrapper.builder()
                     .data(Collections.emptyList())
                     .httpCompletionState(HttpCompletionState.SUCCESS)
@@ -253,7 +261,19 @@ private HttpRowDataWrapper processHttpResponse(
         var responseBody = response.body();
 
         log.debug("Received status code [{}] for RestTableSource request", response.statusCode());
-
+        final boolean ignoreStatusCode = ignoreResponse(response);
+        if (!isError && (StringUtils.isNullOrWhitespaceOnly(responseBody) || ignoreStatusCode)) {
+            return HttpRowDataWrapper.builder()
+                    .data(Collections.emptyList())
+                    .httpCompletionState(HttpCompletionState.SUCCESS)
+                    .httpHeadersMap(response.headers().map())
+                    .httpStatusCode(response.statusCode())
+                    .httpCompletionState(
+                            ignoreStatusCode
+                                    ? HttpCompletionState.IGNORE_STATUS_CODE
+                                    : HttpCompletionState.SUCCESS)
+                    .build();
+        }
         if (this.isSuccessWithNoData(isError, responseBody, response)) {
             return HttpRowDataWrapper.builder()
                     .data(Collections.emptyList())
@@ -326,22 +346,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
         }
     }
 
-    private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
-        return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
-                .map(Collections::singletonList)
-                .orElse(Collections.emptyList());
+    @VisibleForTesting
+    List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+        List<RowData> result = new ArrayList<>();
+        responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
+        return result;
+    }
+
+    @VisibleForTesting
+    Collector<RowData> createRowDataCollector(List<RowData> result) {
+        return new RowDataCollector(result);
+    }
+
+    /** A simple collector implementation that adds RowData records to a list. */
+    @VisibleForTesting
+    static class RowDataCollector implements Collector<RowData> {
+        private final List<RowData> result;
+
+        RowDataCollector(List<RowData> result) {
+            this.result = result;
+        }
+
+        @Override
+        public void collect(RowData record) {
+            result.add(record);
+        }
+
+        @Override
+        public void close() {
+            // No-op - nothing to clean up
+        }
     }
 
-    private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
+    @VisibleForTesting
+    List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
         List<JsonNode> rawObjects = objectMapper.readValue(rawBytes, new TypeReference<>() {});
         List<RowData> result = new ArrayList<>();
         for (JsonNode rawObject : rawObjects) {
             if (!(rawObject instanceof NullNode)) {
-                RowData deserialized =
-                        responseBodyDecoder.deserialize(rawObject.toString().getBytes());
-                // deserialize() returns null if deserialization fails
-                if (deserialized != null) {
-                    result.add(deserialized);
+                List<RowData> deserialized =
+                        deserializeSingleValue(rawObject.toString().getBytes());
+                // deserialize() may return empty list if deserialization fails
+                if (deserialized != null && !deserialized.isEmpty()) {
+                    result.addAll(deserialized);
                 }
             }
         }