This is an automated email from the ASF dual-hosted git repository.
davidradl pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-http.git
The following commit(s) were added to refs/heads/main by this push:
new 7fa2f25 [FLINK-40940][connector-http] Decide lookup row emission
independently of applyReadableMetadata (#56)
7fa2f25 is described below
commit 7fa2f25fe30f1f7c2330b093da17fe9866cbc62e
Author: Arvid Heise <[email protected]>
AuthorDate: Fri Oct 9 17:38:42 2026 +0200
[FLINK-40940][connector-http] Decide lookup row emission independently of
applyReadableMetadata (#56)
* [FLINK-40940][connector-http] Decide lookup row emission independently of
applyReadableMetadata
HttpTableLookupFunction decides whether a response without data yields a
row based on whether producedDataType is set, which only happens when the
planner calls applyReadableMetadata. For tables without metadata columns the
planner calls it only when it pushes a projection, so a failed call with
http.source.lookup.continue-on-error or an ignored status code drops the row
for queries without a pushed projection but emits an all-null row for queries
with one.
A response without data yields a row exactly when metadata columns are
declared, sized from the physical row type. Without metadata columns a failed
call is a lookup miss, so an inner lookup join drops the row and a LEFT JOIN
pads it with nulls; the docs state this explicitly. The unused producedDataType
is removed.
Generated-by: Claude Code (Opus 5.5)
* [FLINK-40940][connector-http] Retain readable metadata in
HttpLookupTableSource#copy
HttpLookupTableSource#copy creates a source from the constructor arguments
only, so a copy taken after applyReadableMetadata drops the metadata keys. The
lookup function of such a copy emits rows without the metadata columns the
planner expects.
The copy carries the metadata keys over, and the source defines equality
over its physical type, lookup config, decoding format, cache and metadata keys.
Generated-by: Claude Code (Opus 5.5)
---------
Co-authored-by: Arvid Heise <[email protected]>
---
docs/content.zh/docs/connectors/table/http.md | 12 +-
docs/content/docs/connectors/table/http.md | 12 +-
.../http/table/lookup/HttpLookupTableSource.java | 42 +++++--
.../http/table/lookup/HttpTableLookupFunction.java | 31 ++---
.../lookup/HttpLookupTableSourceITCaseTest.java | 75 +++++++++++-
.../table/lookup/HttpLookupTableSourceTest.java | 15 +++
...ttpTableLookupFunctionPopulateJoinKeysTest.java | 4 -
.../table/lookup/HttpTableLookupFunctionTest.java | 127 +++++++++++++++++++++
8 files changed, 265 insertions(+), 53 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/http.md
b/docs/content.zh/docs/connectors/table/http.md
index 33be05f..6c25f63 100644
--- a/docs/content.zh/docs/connectors/table/http.md
+++ b/docs/content.zh/docs/connectors/table/http.md
@@ -717,14 +717,14 @@ The metadata column `http-status-code`, if specified in
the table definition, wi
The metadata column `http-headers-map`, if specified in the table definition,
will get a map of the HTTP headers.
HTTP requests can fail either immediately or after temporary error retries.
The usual behaviour after such failures is to end the job. If you would like to
continue
-processing after these failures then specify
`http.source.lookup.continue-on-error` as true. The lookup join will complete
without content in the expected enrichment columns from the http call,
-this means that these columns will be null for nullable columns and hold a
default value for the type for non-nullable columns.
+processing after these failures then specify
`http.source.lookup.continue-on-error` as true. Without metadata columns, a
failed call is then treated as a lookup miss:
+an inner lookup join drops the row, and a `LEFT JOIN` emits it with null
enrichment columns.
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.
+Note that if metadata columns are specified, then a row containing metadata
columns will be produced for every call, including failed
+calls and ignored status codes. If there are no metadata columns defined, then
no row will be emitted for a call without response data;
+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.
@@ -752,7 +752,7 @@ When the HTTP status code is in the
`http.source.lookup.ignored-response-codes`,
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
+When a HTTP lookup call fails and populates the metadata columns with the
error information, the expected enrichment columns from the HTTP call,
including the lookup join key columns,
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.
If you are using the Table API `TableResult` and have an `await` with a
timeout, this Timeout exception will cause the job to terminate,
diff --git a/docs/content/docs/connectors/table/http.md
b/docs/content/docs/connectors/table/http.md
index bc36a19..85822ea 100644
--- a/docs/content/docs/connectors/table/http.md
+++ b/docs/content/docs/connectors/table/http.md
@@ -718,14 +718,14 @@ The metadata column `http-status-code`, if specified in
the table definition, wi
The metadata column `http-headers-map`, if specified in the table definition,
will get a map of the HTTP headers.
HTTP requests can fail either immediately or after temporary error retries.
The usual behaviour after such failures is to end the job. If you would like to
continue
-processing after these failures then specify
`http.source.lookup.continue-on-error` as true. The lookup join will complete
without content in the expected enrichment columns from the http call,
-this means that these columns will be null for nullable columns and hold a
default value for the type for non-nullable columns.
+processing after these failures then specify
`http.source.lookup.continue-on-error` as true. Without metadata columns, a
failed call is then treated as a lookup miss:
+an inner lookup join drops the row, and a `LEFT JOIN` emits it with null
enrichment columns.
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.
+Note that if metadata columns are specified, then a row containing metadata
columns will be produced for every call, including failed
+calls and ignored status codes. If there are no metadata columns defined, then
no row will be emitted for a call without response data;
+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.
@@ -753,7 +753,7 @@ When the HTTP status code is in the
`http.source.lookup.ignored-response-codes`,
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
+When a HTTP lookup call fails and populates the metadata columns with the
error information, the expected enrichment columns from the HTTP call,
including the lookup join key columns,
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.
If you are using the Table API `TableResult` and have an `await` with a
timeout, this Timeout exception will cause the job to terminate,
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
index 298c6b7..1e570ca 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
@@ -58,6 +58,7 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -86,9 +87,6 @@ public class HttpLookupTableSource
// Mutable attributes
//
--------------------------------------------------------------------------------------------
- /** Data type that describes the final output of the source. */
- protected DataType producedDataType;
-
/** Metadata that is appended at the end of a physical source row. */
protected List<String> metadataKeys;
@@ -157,7 +155,6 @@ public class HttpLookupTableSource
lookupRow,
lookupConfig,
metadataConverters,
- this.producedDataType,
this.physicalRowDataType);
if (lookupConfig.isUseAsync()) {
AsyncLookupFunction asyncLookupFunction =
@@ -182,12 +179,36 @@ public class HttpLookupTableSource
@Override
public DynamicTableSource copy() {
- return new HttpLookupTableSource(
- physicalRowDataType,
- lookupConfig,
- decodingFormat,
- dynamicTableFactoryContext,
- cache);
+ final HttpLookupTableSource copy =
+ new HttpLookupTableSource(
+ physicalRowDataType,
+ lookupConfig,
+ decodingFormat,
+ dynamicTableFactoryContext,
+ cache);
+ copy.metadataKeys = metadataKeys;
+ return copy;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final HttpLookupTableSource that = (HttpLookupTableSource) o;
+ return Objects.equals(physicalRowDataType, that.physicalRowDataType)
+ && Objects.equals(lookupConfig, that.lookupConfig)
+ && Objects.equals(decodingFormat, that.decodingFormat)
+ && Objects.equals(cache, that.cache)
+ && Objects.equals(metadataKeys, that.metadataKeys);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(physicalRowDataType, lookupConfig, decodingFormat,
cache, metadataKeys);
}
@Override
@@ -336,7 +357,6 @@ public class HttpLookupTableSource
decodingFormat.applyReadableMetadata(requestedFormatMetadataKeys);
}
this.metadataKeys = connectorMetadataKeys;
- this.producedDataType = producedDataType;
}
//
--------------------------------------------------------------------------------------------
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
index cc41612..609414b 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
@@ -28,7 +28,6 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.LookupFunction;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.types.RowKind;
import lombok.AccessLevel;
@@ -60,7 +59,6 @@ public class HttpTableLookupFunction extends LookupFunction {
private final HttpLookupConfig options;
private transient AtomicInteger localHttpCallCounter;
- private final DataType producedDataType;
private final DataType physicalRowDataType;
private transient PollingClient client;
private final MetadataConverter[] metadataConverters;
@@ -71,7 +69,6 @@ public class HttpTableLookupFunction extends LookupFunction {
LookupRow lookupRow,
HttpLookupConfig options,
MetadataConverter[] metadataConverters,
- DataType producedDataType,
DataType physicalRowDataType) {
this.pollingClientFactory = pollingClientFactory;
@@ -79,7 +76,6 @@ public class HttpTableLookupFunction extends LookupFunction {
this.lookupRow = lookupRow;
this.options = options;
this.metadataConverters = metadataConverters;
- this.producedDataType = producedDataType;
this.physicalRowDataType = physicalRowDataType;
}
@@ -107,6 +103,10 @@ public class HttpTableLookupFunction extends
LookupFunction {
HttpRowDataWrapper httpRowDataWrapper = client.pull(keyRow);
Collection<RowData> httpCollector = httpRowDataWrapper.getData();
+ if (httpCollector.isEmpty() && metadataArity == 0) {
+ // A response without data is a lookup miss unless the row carries
metadata columns.
+ return Collections.emptyList();
+ }
int physicalArity = -1;
GenericRowData producedRow = null;
@@ -168,29 +168,16 @@ public class HttpTableLookupFunction extends
LookupFunction {
producedRow.setField(pos, value);
}
}
- // if we did not get the physical arity from the http response
physical row then get it from
- // the producedDataType, which is set when we have metadata or when
there's no data
- if (physicalArity == -1) {
- if (producedDataType == null) {
- // If producedDataType is null and we have no data, return the
same way as ignore.
- return Collections.emptyList();
- } else {
- List<LogicalType> childrenLogicalTypes =
- producedDataType.getLogicalType().getChildren();
- physicalArity = childrenLogicalTypes.size() - metadataArity;
- }
- }
// if there was no data, create an empty producedRow
if (producedRow == null) {
+ physicalArity = physicalRowDataType.getChildren().size();
producedRow = new GenericRowData(RowKind.INSERT, physicalArity +
metadataArity);
}
// add any metadata
- if (producedDataType != null) {
- for (int metadataPos = 0; metadataPos < metadataArity;
metadataPos++) {
- producedRow.setField(
- physicalArity + metadataPos,
-
metadataConverters[metadataPos].read(httpRowDataWrapper));
- }
+ for (int metadataPos = 0; metadataPos < metadataArity; metadataPos++) {
+ producedRow.setField(
+ physicalArity + metadataPos,
+ metadataConverters[metadataPos].read(httpRowDataWrapper));
}
outputList.add(producedRow);
return outputList;
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java
index d6e8a3a..9305234 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java
@@ -1125,6 +1125,10 @@ class HttpLookupTableSourceITCaseTest {
}
private SortedSet<Row> testLookupJoin(String lookupTable, int maxRows)
throws Exception {
+ return testLookupJoin(lookupTable, maxRows, "JOIN");
+ }
+
+ private TableResult executeLookupJoin(String lookupTable, int maxRows,
String joinType) {
createLookupAndSourceTables(lookupTable, maxRows);
@@ -1132,11 +1136,18 @@ class HttpLookupTableSourceITCaseTest {
// SQL query that performs JOIN on both tables.
String joinQuery =
"SELECT o.id, o.id2, c.msg, c.uuid, c.isActive, c.balance FROM
Orders AS o "
- + "JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS
c "
+ + joinType
+ + " Customers FOR SYSTEM_TIME AS OF o.proc_time AS c "
+ "ON o.id = c.id "
+ "AND o.id2 = c.id2";
- TableResult result = tEnv.executeSql(joinQuery);
+ return tEnv.executeSql(joinQuery);
+ }
+
+ private SortedSet<Row> testLookupJoin(String lookupTable, int maxRows,
String joinType)
+ throws Exception {
+
+ TableResult result = executeLookupJoin(lookupTable, maxRows, joinType);
result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// Wait for all async HTTP requests to complete and be processed
@@ -1448,7 +1459,13 @@ class HttpLookupTableSourceITCaseTest {
}
private void assertResultsForSpec(TestSpec spec, Collection<Row> rows) {
- if (spec.badStatus) {
+ if (spec.badStatus && !spec.useMetadata && !spec.leftJoin) {
+ assertThat(rows).isEmpty();
+ assertThat(wireMockServer.getAllServeEvents())
+ .hasSizeGreaterThanOrEqualTo(spec.maxRows);
+ } else if (spec.badStatus && !spec.useMetadata) {
+ assertNoEnrichment(rows);
+ } else if (spec.badStatus) {
assertEnrichedRowsNoDataBadStatus(rows);
} else if (spec.deserError) {
if (spec.ignoreParseErrors) {
@@ -1495,6 +1512,22 @@ class HttpLookupTableSourceITCaseTest {
}
}
+ private void assertNoEnrichment(Collection<Row> collectedRows) {
+ assertThat(collectedRows).hasSize(4);
+ int intElement = 0;
+ for (Row row : collectedRows) {
+ intElement++;
+ assertThat(row)
+ .returns(6, Row::getArity)
+ .returns(String.valueOf(intElement), r -> r.getField("id"))
+ .returns(String.valueOf(intElement + 1), r ->
r.getField("id2"))
+ .returns(null, r -> r.getField("msg"))
+ .returns(null, r -> r.getField("uuid"))
+ .returns(null, r -> r.getField("isActive"))
+ .returns(null, r -> r.getField("balance"));
+ }
+ }
+
private void assertEnrichedRowsNoDataBadStatus(Collection<Row>
collectedRows) {
final int rowArity = 10;
@@ -1719,8 +1752,14 @@ class HttpLookupTableSourceITCaseTest {
try {
if (spec.useMetadata) {
rows = testLookupJoinWithMetadata(lookupTable, spec.maxRows);
+ } else if (spec.badStatus && !spec.leftJoin) {
+ // An inner join without matches has no first row to await, so
collect until the
+ // bounded job finishes.
+ rows = getCollectedRows(executeLookupJoin(lookupTable,
spec.maxRows, "JOIN"));
} else {
- rows = testLookupJoin(lookupTable, spec.maxRows);
+ rows =
+ testLookupJoin(
+ lookupTable, spec.maxRows, spec.leftJoin ?
"LEFT JOIN" : "JOIN");
}
// THEN
assertResultsForSpec(spec, rows);
@@ -1793,6 +1832,33 @@ class HttpLookupTableSourceITCaseTest {
}
}
+ // Bad status test cases without metadata columns, so the planner
never calls
+ // applyReadableMetadata on the lookup source
+ for (String method : Arrays.asList("GET", "POST", "PUT")) {
+ for (boolean asyncFlag : Arrays.asList(false, true)) {
+ for (boolean continueOnError : Arrays.asList(false, true)) {
+ for (boolean leftJoin : Arrays.asList(false, true)) {
+ specs.add(
+ TestSpec.builder()
+ .testName(
+ "HTTP Lookup Join Bad Status
continue on error:"
+ + continueOnError
+ + ". asyncFlag:"
+ + asyncFlag
+ + ". leftJoin:"
+ + leftJoin)
+ .methodName(method)
+ .maxRows(4)
+ .useAsync(asyncFlag)
+ .badStatus(true)
+ .continueOnError(continueOnError)
+ .leftJoin(leftJoin)
+ .build());
+ }
+ }
+ }
+ }
+
// Deserialization error test cases
(testHttpLookupJoinWithMetadataDeserException)
for (String method : Arrays.asList("GET", "POST", "PUT")) {
for (boolean asyncFlag : Arrays.asList(false, true)) {
@@ -1854,6 +1920,7 @@ class HttpLookupTableSourceITCaseTest {
final boolean useAsync;
final boolean continueOnError;
final boolean ignoreParseErrors;
+ final boolean leftJoin;
@Override
public String toString() {
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java
index 15e5c07..f5f0d60 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java
@@ -169,6 +169,21 @@ class HttpLookupTableSourceTest {
.isEqualTo(StringData.fromString(testCompletionState.name()));
}
+ @Test
+ void copyRetainsReadableMetadata() {
+ final HttpLookupTableSource tableSource =
+ (HttpLookupTableSource) createTableSource(SCHEMA,
getOptions());
+ tableSource.applyProjection(new int[][] {{0}}, PHYSICAL_ROW_DATA_TYPE);
+ tableSource.applyReadableMetadata(
+ List.of(HTTP_STATUS_CODE.key),
+ row(
+ List.of(
+ DataTypes.FIELD("id",
DataTypes.STRING().notNull()),
+ DataTypes.FIELD("status",
HTTP_STATUS_CODE.dataType))));
+
+ assertThat(tableSource.copy()).isEqualTo(tableSource);
+ }
+
private static Map<String, List<String>> convertGenericMapDataToMap(
GenericMapData genericMapData, ArrayData keys) {
Map<String, List<String>> map = new HashMap<>();
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
index f078bd6..503f669 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
@@ -100,7 +100,6 @@ class HttpTableLookupFunctionPopulateJoinKeysTest {
lookupRow,
httpLookupConfig,
new MetadataConverter[0],
- null,
producedDataType);
// Create keyRow with join key value
@@ -171,7 +170,6 @@ class HttpTableLookupFunctionPopulateJoinKeysTest {
lookupRow,
httpLookupConfig,
new MetadataConverter[0],
- null,
producedDataType);
// Create keyRow with three join key values
@@ -241,7 +239,6 @@ class HttpTableLookupFunctionPopulateJoinKeysTest {
lookupRow,
httpLookupConfig,
new MetadataConverter[0],
- null,
producedDataType);
// Create keyRow with join key value
@@ -307,7 +304,6 @@ class HttpTableLookupFunctionPopulateJoinKeysTest {
lookupRow,
httpLookupConfig,
new MetadataConverter[0],
- null,
producedDataType);
// Create keyRow with join key value
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionTest.java
new file mode 100644
index 0000000..f8823c9
--- /dev/null
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionTest.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.http.table.lookup;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.connector.http.clients.PollingClient;
+import org.apache.flink.connector.http.clients.PollingClientFactory;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.InstanceOfAssertFactories.type;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+/** Tests the rows {@link HttpTableLookupFunction} emits for responses without
data. */
+@ExtendWith(MockitoExtension.class)
+class HttpTableLookupFunctionTest {
+
+ private static final DataType PHYSICAL_ROW_DATA_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.STRING()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("age", DataTypes.INT()));
+
+ private static final StringData KEY = StringData.fromString("1");
+
+ @Mock private PollingClientFactory pollingClientFactory;
+
+ @Mock private PollingClient pollingClient;
+
+ @Mock private DeserializationSchema<RowData> responseSchemaDecoder;
+
+ @Mock private FunctionContext context;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ when(pollingClientFactory.createPollClient(any(),
any())).thenReturn(pollingClient);
+ when(context.getMetricGroup()).thenReturn(new
UnregisteredMetricsGroup());
+ }
+
+ /**
+ * Without metadata columns, a response without data is a lookup miss,
failed calls included.
+ */
+ @ParameterizedTest
+ @EnumSource
+ void noDataWithoutMetadataEmitsNoRow(final HttpCompletionState
completionState)
+ throws Exception {
+ final HttpTableLookupFunction lookupFunction = openLookupFunction();
+
when(pollingClient.pull(any())).thenReturn(noDataResponse(completionState));
+
+ assertThat(lookupFunction.lookup(GenericRowData.of(KEY))).isEmpty();
+ }
+
+ @ParameterizedTest
+ @EnumSource
+ void noDataWithMetadataEmitsMetadataRow(final HttpCompletionState
completionState)
+ throws Exception {
+ final HttpTableLookupFunction lookupFunction =
+ openLookupFunction(new
MetadataConverter.HttpCompletionStateConverter());
+
when(pollingClient.pull(any())).thenReturn(noDataResponse(completionState));
+
+ assertThat(lookupFunction.lookup(GenericRowData.of(KEY)))
+ .singleElement(type(GenericRowData.class))
+ .returns(4, GenericRowData::getArity)
+ .returns(null, row -> row.getField(0))
+ .returns(null, row -> row.getField(1))
+ .returns(null, row -> row.getField(2))
+ .returns(StringData.fromString(completionState.name()), row ->
row.getField(3));
+ }
+
+ private HttpTableLookupFunction openLookupFunction(
+ final MetadataConverter... metadataConverters) throws Exception {
+ final LookupRow lookupRow = new LookupRow();
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "id", RowData.createFieldGetter(new VarCharType(),
0)));
+ final HttpTableLookupFunction lookupFunction =
+ new HttpTableLookupFunction(
+ pollingClientFactory,
+ responseSchemaDecoder,
+ lookupRow,
+ HttpLookupConfig.builder().build(),
+ metadataConverters,
+ PHYSICAL_ROW_DATA_TYPE);
+ lookupFunction.open(context);
+ return lookupFunction;
+ }
+
+ private static HttpRowDataWrapper noDataResponse(final HttpCompletionState
completionState) {
+ return HttpRowDataWrapper.builder()
+ .data(Collections.emptyList())
+ .httpCompletionState(completionState)
+ .build();
+ }
+}
