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 4f5f3c6 [FLINK-40149][connector-http] Rename asyncPolling to
async-polling (#53)
4f5f3c6 is described below
commit 4f5f3c6a01b195c0823f3708d6359862e25d3c99
Author: Bharath Reddy Gunapati
<[email protected]>
AuthorDate: Tue Jul 21 20:29:43 2026 +0530
[FLINK-40149][connector-http] Rename asyncPolling to async-polling (#53)
* [FLINK-40149][connector-http] Rename asyncPolling to async-polling
Rename the canonical lookup option key to async-polling to align with Flink
connector kebab-case conventions. The legacy asyncPolling key remains
supported
via withDeprecatedKeys for backward-compatible DDL.
Updates docs, examples, and tests to use the new canonical key. Adds
factory-
and runtime-level parameterized tests for both the canonical key and the
deprecated alias. Documents the alias in the ConfigOption description and
docs.
* [FLINK-40149][connector-http] Address David's review on async-polling
rename
Document async-polling as preferred and asyncPolling as deprecated in both
connector docs, reject DDL that sets both keys, and keep the ConfigOption
description focused on behavior only.
* [FLINK-40149][connector-http] Simplify validation overload, document
deprecated-key WARN, and tighten docs wording.
---
docs/content.zh/docs/connectors/table/http.md | 7 ++---
docs/content/docs/connectors/table/http.md | 7 ++---
.../table/lookup/HttpLookupConnectorOptions.java | 12 +++++++--
.../table/lookup/HttpLookupTableSourceFactory.java | 21 +++++++++++++--
.../flink/connector/http/StreamTableJob.java | 2 +-
.../lookup/HttpLookupTableSourceFactoryTest.java | 27 ++++++++++++++++---
.../lookup/HttpLookupTableSourceITCaseTest.java | 30 +++++++++++-----------
.../table/lookup/HttpLookupTableSourceTest.java | 16 +++++-------
8 files changed, 83 insertions(+), 39 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/http.md
b/docs/content.zh/docs/connectors/table/http.md
index 8781ae0..93f2731 100644
--- a/docs/content.zh/docs/connectors/table/http.md
+++ b/docs/content.zh/docs/connectors/table/http.md
@@ -195,7 +195,7 @@ CREATE TABLE Customers (
'connector' = 'http',
'format' = 'json',
'url' = 'http://localhost:8080/client',
-'asyncPolling' = 'true'
+'async-polling' = 'true'
)
```
@@ -250,7 +250,8 @@ Note the options with the prefix _http_ are the HTTP
connector specific options,
| connector |
required | The Value should be set to _http_
|
| format |
required | Flink's format name that should be used to decode REST response, Use
`json` for a typical REST endpoint.
|
| url |
required | The base URL that should be used for GET requests. For example
_http://localhost:8080/client_
|
-| asyncPolling |
optional | true/false - determines whether Async Polling should be used.
Mechanism is based on Flink's Async I/O.
|
+| async-polling |
optional | **Preferred.** true/false - determines whether async polling should
be used. Mechanism is based on Flink's Async I/O.
|
+| asyncPolling |
optional | **Deprecated.** Use `async-polling` instead. Setting both keys in
the same DDL is a validation error.
|
| lookup-method |
optional | GET/POST/PUT (and any other) - determines what REST method should be
used for lookup REST query. If not specified, `GET` method will be used.
|
| lookup.cache |
optional | Enum possible values: `NONE`, `PARTIAL`. The cache strategy for the
lookup table. Currently supports `NONE` (no caching) and `PARTIAL` (caching
entries on lookup operation in external API).
|
| lookup.partial-cache.max-rows |
optional | The max number of rows of lookup cache, over this value, the oldest
rows will be expired. `lookup.cache` must be set to `PARTIAL` to use this
option. See the following <a href="#lookup-cache">Lookup Cache</a> section for
more details.
|
@@ -494,7 +495,7 @@ CREATE TABLE http-lookup (
'connector' = 'http',
'format' = 'json',
'url' = 'http://localhost:8080/client',
- 'asyncPolling' = 'true',
+ 'async-polling' = 'true',
'http.source.lookup.header.Origin' = '*',
'http.source.lookup.header.X-Content-Type-Options' = 'nosniff',
'http.source.lookup.header.Content-Type' = 'application/json'
diff --git a/docs/content/docs/connectors/table/http.md
b/docs/content/docs/connectors/table/http.md
index 120847e..915f448 100644
--- a/docs/content/docs/connectors/table/http.md
+++ b/docs/content/docs/connectors/table/http.md
@@ -196,7 +196,7 @@ CREATE TABLE Customers (
'connector' = 'http',
'format' = 'json',
'url' = 'http://localhost:8080/client',
-'asyncPolling' = 'true'
+'async-polling' = 'true'
)
```
@@ -251,7 +251,8 @@ Note the options with the prefix _http_ are the HTTP
connector specific options,
| connector |
required | The Value should be set to _http_
|
| format |
required | Flink's format name that should be used to decode REST response, Use
`json` for a typical REST endpoint.
|
| url |
required | The base URL that should be used for GET requests. For example
_http://localhost:8080/client_
|
-| asyncPolling |
optional | true/false - determines whether Async Polling should be used.
Mechanism is based on Flink's Async I/O.
|
+| async-polling |
optional | **Preferred.** true/false - determines whether async polling should
be used. Mechanism is based on Flink's Async I/O.
|
+| asyncPolling |
optional | **Deprecated.** Use `async-polling` instead. Setting both keys in
the same DDL is a validation error.
|
| lookup-method |
optional | GET/POST/PUT (and any other) - determines what REST method should be
used for lookup REST query. If not specified, `GET` method will be used.
|
| lookup.cache |
optional | Enum possible values: `NONE`, `PARTIAL`. The cache strategy for the
lookup table. Currently supports `NONE` (no caching) and `PARTIAL` (caching
entries on lookup operation in external API).
|
| lookup.partial-cache.max-rows |
optional | The max number of rows of lookup cache, over this value, the oldest
rows will be expired. `lookup.cache` must be set to `PARTIAL` to use this
option. See the following <a href="#lookup-cache">Lookup Cache</a> section for
more details.
|
@@ -495,7 +496,7 @@ CREATE TABLE http-lookup (
'connector' = 'http',
'format' = 'json',
'url' = 'http://localhost:8080/client',
- 'asyncPolling' = 'true',
+ 'async-polling' = 'true',
'http.source.lookup.header.Origin' = '*',
'http.source.lookup.header.X-Content-Type-Options' = 'nosniff',
'http.source.lookup.header.Content-Type' = 'application/json'
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java
index 0a3843c..e4e1f4d 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java
@@ -63,11 +63,19 @@ public class HttpLookupConnectorOptions {
.noDefaultValue()
.withDescription("The arguments that should be used for
HTTP GET Request.");
+ /**
+ * Legacy DDL key; still accepted. Prefer {@link #ASYNC_POLLING} key
{@code async-polling}.
+ * Flink's {@code withDeprecatedKeys} emits a WARN log at runtime when
this key is used; that
+ * warning is intentional and expected, not a bug.
+ */
+ public static final String DEPRECATED_ASYNC_POLLING_KEY = "asyncPolling";
+
public static final ConfigOption<Boolean> ASYNC_POLLING =
- ConfigOptions.key("asyncPolling")
+ ConfigOptions.key("async-polling")
.booleanType()
.defaultValue(false)
- .withDescription("Whether to use Sync and Async polling
mechanism");
+ .withDeprecatedKeys(DEPRECATED_ASYNC_POLLING_KEY)
+ .withDescription("Whether to use sync or async polling
mechanism.");
public static final ConfigOption<String> LOOKUP_METHOD =
ConfigOptions.key("lookup-method")
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java
index 85d3fa2..00b72f1 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java
@@ -27,6 +27,7 @@ import
org.apache.flink.connector.http.config.HttpConnectorConfigConstants;
import org.apache.flink.connector.http.utils.ConfigUtils;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.format.DecodingFormat;
@@ -43,6 +44,7 @@ import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.function.Predicate;
@@ -115,7 +117,8 @@ public class HttpLookupTableSourceFactory implements
DynamicTableSourceFactory {
"lookup-request.",
HttpConnectorConfigConstants.FLINK_CONNECTOR_HTTP,
LOOKUP_REQUEST_FORMAT.key());
- validateHttpLookupSourceOptions(readable);
+ validateHttpLookupSourceOptions(
+ readable, dynamicTableContext.getCatalogTable().getOptions());
HttpLookupConfig lookupConfig =
getHttpLookupOptions(dynamicTableContext, readable);
@@ -132,8 +135,22 @@ public class HttpLookupTableSourceFactory implements
DynamicTableSourceFactory {
getLookupCache(readable));
}
- protected void validateHttpLookupSourceOptions(ReadableConfig tableOptions)
+ protected void validateHttpLookupSourceOptions(
+ ReadableConfig tableOptions, Map<String, String> rawOptions)
throws IllegalArgumentException {
+ if (rawOptions.containsKey(ASYNC_POLLING.key())
+ && rawOptions.containsKey(
+
HttpLookupConnectorOptions.DEPRECATED_ASYNC_POLLING_KEY)) {
+ throw new ValidationException(
+ "Cannot set both '"
+ + ASYNC_POLLING.key()
+ + "' and '"
+ +
HttpLookupConnectorOptions.DEPRECATED_ASYNC_POLLING_KEY
+ + "'. Use '"
+ + ASYNC_POLLING.key()
+ + "' for new configurations.");
+ }
+
// ensure that there is an OIDC token request if we have an OIDC token
endpoint
tableOptions
.getOptional(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL)
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java
index 03fda34..e37495a 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java
@@ -44,7 +44,7 @@ public class StreamTableJob {
+ " 'url' = 'http://localhost:";
+ WireMockServerPortAllocator.getServerPort()
+ "/client', "
- + "'asyncPolling' = 'true', "
+ + "'async-polling' = 'true', "
+ "'field.isActive.path' = '$.details.isActive', "
+ "'field.balance.path' =
'$.details.nestedDetails.balance')");
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java
index 2fbe4a9..e9171c6 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java
@@ -28,6 +28,8 @@ import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.util.Arrays;
import java.util.Collections;
@@ -70,12 +72,14 @@ public class HttpLookupTableSourceFactoryTest {
HttpLookupTableSourceFactory httpLookupTableSourceFactory =
new HttpLookupTableSourceFactory();
TableConfig tableConfig = TableConfig.getDefault();
-
httpLookupTableSourceFactory.validateHttpLookupSourceOptions(tableConfig);
+ httpLookupTableSourceFactory.validateHttpLookupSourceOptions(
+ tableConfig, Collections.emptyMap());
tableConfig.set(
HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL, "aaa");
try {
-
httpLookupTableSourceFactory.validateHttpLookupSourceOptions(tableConfig);
+ httpLookupTableSourceFactory.validateHttpLookupSourceOptions(
+ tableConfig, Collections.emptyMap());
fail("Expected an error.");
} catch (IllegalArgumentException e) {
// expected
@@ -83,7 +87,8 @@ public class HttpLookupTableSourceFactoryTest {
// should now work.
tableConfig.set(HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST,
"bbb");
-
httpLookupTableSourceFactory.validateHttpLookupSourceOptions(tableConfig);
+ httpLookupTableSourceFactory.validateHttpLookupSourceOptions(
+ tableConfig, Collections.emptyMap());
}
@Test
@@ -139,6 +144,22 @@ public class HttpLookupTableSourceFactoryTest {
assertThat(source).isInstanceOf(HttpLookupTableSource.class);
}
+ @ParameterizedTest
+ @CsvSource({"async-polling", "asyncPolling"})
+ void shouldAcceptAsyncPollingAtFactory(String asyncOptionKey) {
+ Map<String, String> options = getOptions(Map.of(asyncOptionKey,
"true"));
+ DynamicTableSource source = createTableSource(SCHEMA, options);
+ assertThat(source).isInstanceOf(HttpLookupTableSource.class);
+ }
+
+ @Test
+ void shouldRejectBothAsyncPollingKeys() {
+ Map<String, String> options =
+ getOptions(Map.of("async-polling", "true", "asyncPolling",
"false"));
+ assertThatExceptionOfType(ValidationException.class)
+ .isThrownBy(() -> createTableSource(SCHEMA, options));
+ }
+
private Map<String, String> getMandatoryOptions() {
return Map.of(
"connector", "http",
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 a2036e4..d6e8a3a 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
@@ -192,7 +192,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true',"
+ + "'async-polling' = 'true',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
+ "'table.exec.async-lookup.timeout' = '20s'"
+ ")";
@@ -228,7 +228,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
// WHEN/THEN
@@ -373,7 +373,7 @@ class HttpLookupTableSourceITCaseTest {
+ secServerPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true',"
+ + "'async-polling' = 'true',"
+ "'http.security.cert.server' = '%s',"
+ "'http.security.cert.client' = '%s',"
+ "'http.security.key.client' = '%s'"
@@ -433,7 +433,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -509,7 +509,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -585,7 +585,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -655,7 +655,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -773,7 +773,7 @@ class HttpLookupTableSourceITCaseTest {
+ authHeaderRawValue
+ "',")
: "")
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -874,7 +874,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true'"
+ + "'async-polling' = 'true'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -972,7 +972,7 @@ class HttpLookupTableSourceITCaseTest {
+ "'url' = '{{url}}',"
+ "'http.request.url-map' = 'url:url',"
+ "'format' = 'json',"
- + "'asyncPolling' = 'false',"
+ + "'async-polling' = 'false',"
+ "'lookup-method' = 'POST',"
+ "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ "'http.request.body-template' =
'{\"customerId\":\"1\"}',"
@@ -1040,7 +1040,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + (isAsync ? "'asyncPolling' = 'true'," : "")
+ + (isAsync ? "'async-polling' = 'true'," : "")
+ "'lookup.cache' = 'partial',"
+ "'lookup.partial-cache.max-rows' = '100'"
+ ")";
@@ -1182,7 +1182,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true',"
+ + "'async-polling' = 'true',"
+ "'lookup-request.format' = 'json',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
+ "'table.exec.async-lookup.timeout' = '20s',"
@@ -1279,7 +1279,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort2
+ "/client?customer={{customer}}&id2={{id2}}',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true',"
+ + "'async-polling' = 'true',"
+ "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
+ "'table.exec.async-lookup.timeout' = '20s',"
@@ -1369,7 +1369,7 @@ class HttpLookupTableSourceITCaseTest {
+ serverPort
+ "/client',"
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
- + "'asyncPolling' = 'true',"
+ + "'async-polling' = 'true',"
+ "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ "'lookup-request.format' = 'json',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
@@ -1954,7 +1954,7 @@ class HttpLookupTableSourceITCaseTest {
.append("'http.source.lookup.continue-on-error'='")
.append(spec.continueOnError)
.append("',");
- sql.append("'asyncPolling' = '")
+ sql.append("'async-polling' = '")
.append(spec.useAsync ? "true" : "false")
.append("',")
.append("'table.exec.async-lookup.buffer-capacity' = '50',")
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 a610b56..15e5c07 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
@@ -42,6 +42,7 @@ import org.apache.flink.table.types.logical.IntType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.util.ArrayList;
@@ -220,10 +221,12 @@ class HttpLookupTableSourceTest {
.isNotNull();
}
- @Test
+ @ParameterizedTest
+ @CsvSource({"async-polling", "asyncPolling"})
@SuppressWarnings("unchecked")
- void shouldCreateAsyncTableSourceWithParams() {
- Map<String, String> options = getOptionsWithAsync();
+ void shouldCreateAsyncTableSourceWithAsyncPollingKey(String
asyncOptionKey) {
+ Map<String, String> options = new HashMap<>(getOptions());
+ options.put(asyncOptionKey, "true");
HttpLookupTableSource tableSource =
(HttpLookupTableSource) createTableSource(SCHEMA, options);
@@ -336,13 +339,6 @@ class HttpLookupTableSourceTest {
return tableSource.getLookupRuntimeProvider(null, null, null);
}
- private Map<String, String> getOptionsWithAsync() {
- Map<String, String> options = getOptions();
- options = new HashMap<>(options);
- options.put("asyncPolling", "true");
- return options;
- }
-
private Map<String, String> getOptions() {
return Map.of(
"connector", "http",
