You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fc...@apache.org on 2026/05/16 09:35:36 UTC
(flink-connector-http) branch main updated: [FLINK-39675] Remove redundant query param config
This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky 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 7211b2f [FLINK-39675] Remove redundant query param config
7211b2f is described below
commit 7211b2fc7b94563f041ba5e0241374259b124390
Author: David Radley <39...@users.noreply.github.com>
AuthorDate: Sat May 16 10:35:32 2026 +0100
[FLINK-39675] Remove redundant query param config
---
docs/content.zh/docs/connectors/table/http.md | 157 +++++------
docs/content/docs/connectors/table/http.md | 135 ++++-----
.../http/table/lookup/BodyBasedRequestFactory.java | 24 +-
.../http/table/lookup/RequestFactoryBase.java | 14 +-
.../GenericJsonAndUrlQueryCreator.java | 109 ++------
.../GenericJsonAndUrlQueryCreatorFactory.java | 150 +---------
.../table/lookup/BodyBasedRequestFactoryTest.java | 60 +++-
.../lookup/HttpLookupTableSourceITCaseTest.java | 12 +-
.../GenericJsonAndUrlQueryCreatorFactoryTest.java | 309 +--------------------
.../GenericJsonAndUrlQueryCreatorTest.java | 223 +--------------
10 files changed, 248 insertions(+), 945 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/http.md b/docs/content.zh/docs/connectors/table/http.md
index c0a383a..2713343 100644
--- a/docs/content.zh/docs/connectors/table/http.md
+++ b/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
@@ -87,12 +87,12 @@ The Flink connector does have some changes that you need to be aware of if you a
* Existing java applications will need to be recompiled to pick up the new flink package names.
* Existing application and SQL need to be amended to use the new connector option names. The new option names do not have
-the _com.getindata.http_ prefix, the prefix is now _http_.
+ the _com.getindata.http_ prefix, the prefix is now _http_.
* The name of the connector and the identifiers of components that are discovered have been changed, so that the GetInData jar file can co-exist
-with this connector's jar file. Be aware that if you have created custom pluggable components; you will need to recompile against this connector.
-* Note that the `http-generic-json-url` query creator now processes HTTP bodies differently using `http.request.body-template`.
-* Note that if you were incorrectly using `gid.connector.http.request.query-param-fields` with POST or PUT did not give an error. This connector corrects the behaviour so specifying `http.request.query-param-fields` with POST or PUT does give an error.
-* The GetInData HTTP connector was built against Flink version 1, so works with that level of Flink and also Flink version 2. This connector is built against and supports Flink 2.2.
+ with this connector's jar file. Be aware that if you have created custom pluggable components; you will need to recompile against this connector.
+* Note that the `http-generic-json-url` query creator now processes HTTP bodies differently using `http.request.body-template`.
+* Note that if you were incorrectly using `gid.connector.http.request.query-param-fields` with POST or PUT did not give an error. This connector corrects the behaviour so specifying `http.request.query-param-fields` with POST or PUT does give an error.
+* The GetInData HTTP connector was built against Flink version 1, so works with that level of Flink and also Flink version 2. This connector is built against and supports Flink 2.2.
## Working with HTTP lookup source tables
@@ -165,50 +165,48 @@ Or for REST POST method they will be converted to Json and used as request body.
Note the options with the prefix _http_ are the HTTP connector specific options, the others are Flink options.
-| Option | Required | Description/Value [...]
-|:-----------------------------------------------------------------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- [...]
-| 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. [...]
-| 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. [...]
-| lookup.partial-cache.expire-after-write | optional | The max time to live for each rows in lookup cache after writing into the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. [...]
-| lookup.partial-cache.expire-after-access | optional | The max time to live for each rows in lookup cache after accessing the entry in the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. [...]
-| lookup.partial-cache.cache-missing-key | optional | This is a boolean that defaults to true. Whether to store an empty value into the cache if the lookup key doesn't match any rows in the table. `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. [...]
-| lookup.max-retries | optional | The max retry times if the lookup failed; default is 3. See the following <a href="#lookup-cache">Lookup Cache</a> section for more detail. Set value 0 to disable retries. [...]
-| http.security.cert.server | optional | Comma separated paths to trusted HTTP server certificates that should be added to the connectors trust store. [...]
-| http.security.cert.client | optional | Path to trusted certificate that should be used by connector's HTTP client for mTLS communication. [...]
-| http.security.key.client | optional | Path to trusted private key that should be used by connector's HTTP client for mTLS communication. [...]
-| http.security.cert.server.allowSelfSigned | optional | Accept untrusted certificates for TLS communication. [...]
-| http.security.oidc.token.request | optional | OIDC `Token Request` body in `application/x-www-form-urlencoded` encoding [...]
-| http.security.oidc.token.endpoint.url | optional | OIDC `Token Endpoint` url, to which the token request will be issued [...]
-| http.security.oidc.token.expiry.reduction | optional | OIDC tokens will be requested if the current time is later than the cached token expiry time minus this value. [...]
-| http.source.lookup.continue-on-error | optional | When true, the flow will continue on errors, returning row content. When false (the default) the job ends on errors. [...]
-| http.source.lookup.request.timeout | optional | Sets HTTP request timeout for the lookup source as a Duration (e.g. `'30s'`, `'1min'`). Controls how long the HTTP client waits for a response before timing out a single request. If not specified, the default value of `30s` will be used. [...]
-| http.source.lookup.request.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 8 threads will be used. [...]
-| http.source.lookup.response.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup response processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 4 threads will be used. [...]
-| http.source.lookup.use-raw-authorization-header | optional | If set to `'true'`, uses the raw value set for the `Authorization` header, without transformation for Basic Authentication (base64, addition of "Basic " prefix). If not specified, defaults to `'false'`. [...]
-| http.source.lookup.request-callback | optional | Specify which `HttpLookupPostRequestCallback` implementation to use. By default, it is set to `slf4j-lookup-logger` corresponding to `Slf4jHttpLookupPostRequestCallback`. [...]
-| http.source.lookup.connection.timeout | optional | Source table connection timeout. Default - no value. [...]
-| http.source.lookup.http-version | optional | Version of HTTP to use for lookup http requests. The valid values are HTTP_1_1 and HTTP_2, which specify HTTP 1.1 or 2 respectively. This option may be required as HTTP_1_1, if the endpoint is HTTP 1.1, because some http 1.1 endpoints reject HTTP Version 2 calls, with 'Invalid HTTP request received' and 'HTTP/2 upgrade not supported'. [...]
-| http.source.lookup.success-codes | optional | Comma separated http codes considered as success response. Use [1-5]XX for groups and '!' character for excluding. The default is 2XX. [...]
-| http.source.lookup.retry-codes | optional | Comma separated http codes considered as transient errors. Use [1-5]XX for groups and '!' character for excluding. The default is 500,503,504. [...]
-| http.source.lookup.ignored-response-codes | optional | Comma separated http codes. Content for these responses will be ignored. Use [1-5]XX for groups and '!' character for excluding. Ignored responses together with `http.source.lookup.success-codes` are considered as successful. [...]
-| http.source.lookup.retry-strategy.type | optional | Auto retry strategy type: fixed-delay (default) or exponential-delay. [...]
-| http.source.lookup.retry-strategy.fixed-delay.delay | optional | Fixed-delay interval between retries. Default 1 second. Use with`lookup.max-retries` parameter. [...]
-| http.source.lookup.retry-strategy.exponential-delay.initial-backoff | optional | Exponential-delay initial delay. Default 1 second. [...]
-| http.source.lookup.retry-strategy.exponential-delay.max-backoff | optional | Exponential-delay maximum delay. Default 1 minute. Use with `lookup.max-retries` parameter. [...]
-| http.source.lookup.retry-strategy.exponential-delay.backoff-multiplier | optional | Exponential-delay multiplier. Default value 1.5 [...]
-| http.source.lookup.proxy.host | optional | Specify the hostname of the proxy. [...]
-| http.source.lookup.proxy.port | optional | Specify the port of the proxy. [...]
-| http.source.lookup.proxy.username | optional | Specify the username used for proxy authentication. [...]
-| http.source.lookup.proxy.password | optional | Specify the password used for proxy authentication. [...]
-| http.request.query-param-fields | optional | Used for the `http-generic-json-url` query creator. The names of the fields that will be mapped to query parameters. The parameters are separated by semicolons, such as `param1;param2`. [...]
-| http.request.query-param-fields-with-key | optional | A map of column names to query parameter keys. See the [Query Parameter Mapping](#query-parameter-mapping) section for details and examples. [...]
-| http.request.body-template | optional | Used for the `http-generic-json-url` query creator. A JSON template string for constructing the request body for PUT and POST operations. Use `{{fieldName}}` placeholders to reference top-level columns from the lookup table. Supports creating complex nested JSON structures with both placeholders and literal values. See the [Body Template](#body-template) section for details and examples. [...]
-| http.request.url-map | optional | Used for the `http-generic-json-url` query creator. The map of insert names to column names used as url segments. Parses a string as a map of strings. For example if there are table columns called `customerId` and `orderId`, then specifying value `customerId:cid,orderID:oid` and a url of https://myendpoint/customers/{cid}/orders/{oid} will mean that the url used for the lookup query will dynamically pic [...]
+| Option | Required | Description/Value |
+|:-----------------------------------------------------------------------|----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 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. |
+| 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. |
+| lookup.partial-cache.expire-after-write | optional | The max time to live for each rows in lookup cache after writing into the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. |
+| lookup.partial-cache.expire-after-access | optional | The max time to live for each rows in lookup cache after accessing the entry in the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. |
+| lookup.partial-cache.cache-missing-key | optional | This is a boolean that defaults to true. Whether to store an empty value into the cache if the lookup key doesn't match any rows in the table. `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. |
+| lookup.max-retries | optional | The max retry times if the lookup failed; default is 3. See the following <a href="#lookup-cache">Lookup Cache</a> section for more detail. Set value 0 to disable retries. |
+| http.security.cert.server | optional | Comma separated paths to trusted HTTP server certificates that should be added to the connectors trust store. |
+| http.security.cert.client | optional | Path to trusted certificate that should be used by connector's HTTP client for mTLS communication. |
+| http.security.key.client | optional | Path to trusted private key that should be used by connector's HTTP client for mTLS communication. |
+| http.security.cert.server.allowSelfSigned | optional | Accept untrusted certificates for TLS communication. |
+| http.security.oidc.token.request | optional | OIDC `Token Request` body in `application/x-www-form-urlencoded` encoding |
+| http.security.oidc.token.endpoint.url | optional | OIDC `Token Endpoint` url, to which the token request will be issued |
+| http.security.oidc.token.expiry.reduction | optional | OIDC tokens will be requested if the current time is later than the cached token expiry time minus this value. |
+| http.source.lookup.continue-on-error | optional | When true, the flow will continue on errors, returning row content. When false (the default) the job ends on errors. |
+| http.source.lookup.request.timeout | optional | Sets HTTP request timeout for the lookup source as a Duration (e.g. `'30s'`, `'1min'`). Controls how long the HTTP client waits for a response before timing out a single request. If not specified, the default value of `30s` will be used. |
+| http.source.lookup.request.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 8 threads will be used. |
+| http.source.lookup.response.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup response processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 4 threads will be used. |
+| http.source.lookup.use-raw-authorization-header | optional | If set to `'true'`, uses the raw value set for the `Authorization` header, without transformation for Basic Authentication (base64, addition of "Basic " prefix). If not specified, defaults to `'false'`. |
+| http.source.lookup.request-callback | optional | Specify which `HttpLookupPostRequestCallback` implementation to use. By default, it is set to `slf4j-lookup-logger` corresponding to `Slf4jHttpLookupPostRequestCallback`. |
+| http.source.lookup.connection.timeout | optional | Source table connection timeout. Default - no value. |
+| http.source.lookup.http-version | optional | Version of HTTP to use for lookup http requests. The valid values are HTTP_1_1 and HTTP_2, which specify HTTP 1.1 or 2 respectively. This option may be required as HTTP_1_1, if the endpoint is HTTP 1.1, because some http 1.1 endpoints reject HTTP Version 2 calls, with 'Invalid HTTP request received' and 'HTTP/2 upgrade not supported'. |
+| http.source.lookup.success-codes | optional | Comma separated http codes considered as success response. Use [1-5]XX for groups and '!' character for excluding. The default is 2XX. |
+| http.source.lookup.retry-codes | optional | Comma separated http codes considered as transient errors. Use [1-5]XX for groups and '!' character for excluding. The default is 500,503,504. |
+| http.source.lookup.ignored-response-codes | optional | Comma separated http codes. Content for these responses will be ignored. Use [1-5]XX for groups and '!' character for excluding. Ignored responses together with `http.source.lookup.success-codes` are considered as successful. |
+| http.source.lookup.retry-strategy.type | optional | Auto retry strategy type: fixed-delay (default) or exponential-delay. |
+| http.source.lookup.retry-strategy.fixed-delay.delay | optional | Fixed-delay interval between retries. Default 1 second. Use with`lookup.max-retries` parameter. |
+| http.source.lookup.retry-strategy.exponential-delay.initial-backoff | optional | Exponential-delay initial delay. Default 1 second. |
+| http.source.lookup.retry-strategy.exponential-delay.max-backoff | optional | Exponential-delay maximum delay. Default 1 minute. Use with `lookup.max-retries` parameter. |
+| http.source.lookup.retry-strategy.exponential-delay.backoff-multiplier | optional | Exponential-delay multiplier. Default value 1.5 |
+| http.source.lookup.proxy.host | optional | Specify the hostname of the proxy. |
+| http.source.lookup.proxy.port | optional | Specify the port of the proxy. |
+| http.source.lookup.proxy.username | optional | Specify the username used for proxy authentication. |
+| http.source.lookup.proxy.password | optional | Specify the password used for proxy authentication. |
+| http.request.body-template | optional | Used for the `http-generic-json-url` query creator. A JSON template string for constructing the request body for PUT and POST operations. Use `{{fieldName}}` placeholders to reference top-level columns from the lookup table. Supports creating complex nested JSON structures with both placeholders and literal values. See the [Body Template](#body-template) section for details and examples. |
+| http.request.url-map | optional | Used for the `http-generic-json-url` query creator. The map of insert names to column names used as url segments or query parameters. |
### Query Creators
@@ -247,28 +245,20 @@ The HTTP connector supplies a number of Query Creators that you can use to defin
The recommended Query creator for json is called _http-generic-json-url_, which allows column content to be mapped as URL, path, body and query parameter request values; it supports
POST, PUT and GET operations. This query creator allows you to issue json requests without needing to code
your own custom http connector. The mappings from columns to the json request are supplied in the query creator configuration
-parameters `http.request.query-param-fields`, `http.request.query-param-fields-with-key`, `http.request.body-template` and `http.request.url-map`.
-#### Query Parameter Mapping
+parameters `http.request.body-template` and `http.request.url-map`.
+#### Mapping the URL
-The `http.request.query-param-fields` and `http.request.query-param-fields-with-key` options provide flexible ways to map table columns to HTTP query parameters for GET requests.
+The `http.request.url-map` option provides a flexible way to map table columns to parts of the URL, either URL segments or HTTP query parameters.
+Parses a string as a map of strings. For example if there are table columns called `customerId` and `orderId`,
+then specifying value `customerId:cid,orderID:oid` and a url of https://myendpoint/customers/{cid}?orders={oid} will mean that the url used for the
+lookup query will dynamically pickup the values for `customerId`, `orderId` and use them in the url e.g. https://myendpoint/customers/cid1?orders=oid1.
+The expected format of the map is: `key1:value1,key2:value2`.
-**Basic Query Parameter Mapping (`http.request.query-param-fields`):**
+As these values are being supplied as URL segments or part or query parameters, the connector url encodes that content so characters like spaces
+do not appear invalidly in the URL. In the case where the complete url is the insert then url encoding is not performed; the url needs to be valid and already
+properly url encoded as appropriate.
-Use this option to map column names directly to query parameters with the same name. The parameters are separated by semicolons.
-
-```sql
-'http.request.query-param-fields' = 'userId;orderId'
-```
-
-This will map the `userId` and `orderId` columns to query parameters `?userId=value1&orderId=value2`.
-
-**Advanced Query Parameter Mapping (`http.request.query-param-fields-with-key`):**
-
-Use this option when you need to use different names for query parameters than the table column names. This is necessary when request API fields have the same name as response fields and incompatible types.
-
-**Format:** `columnName1:queryParamKey1,columnName2:queryParamKey2`
-
-**Example Scenario:**
+**Example Scenario around clashing request and response columns:**
If your API response has a field `customer` defined as an object (complex type), but you need to send a customer ID as a query parameter with the same name `customer`, you can:
@@ -276,14 +266,6 @@ If your API response has a field `customer` defined as an object (complex type),
2. Keep the `customer` column for the response object
3. Map the request column to the query parameter:
-```sql
-'http.request.query-param-fields-with-key' = 'qp_customer:customer'
-```
-
-This will map the `qp_customer` column value to the query parameter `?customer=value`.
-
-**Complete Example:**
-
```sql
CREATE TABLE CustomerLookup (
qp_customer STRING, -- Request: customer ID as string
@@ -299,9 +281,9 @@ CREATE TABLE CustomerLookup (
) WITH (
'connector' = 'http',
'format' = 'json',
- 'url' = 'http://api.example.com/lookup',
+ 'url' = 'http://api.example.com/lookup?customer={qp_customer}&order={qp_order}',
'lookup-method' = 'GET',
- 'http.request.query-param-fields-with-key' = 'qp_customer:customer,qp_order:order'
+ 'http.request.url-map' = 'qp_customer:qp_customer,qp_order:qp_order'
)
```
@@ -312,13 +294,6 @@ GET http://api.example.com/lookup?customer=C123&order=O456
The response will populate the `customer` and `order` complex objects.
-**Important Notes for `http.request.query-param-fields-with-key`:**
-- Can only be used with GET requests
-- Column names and query parameter keys cannot be null or empty
-- Query parameter keys must be unique (no duplicates allowed)
-- Cannot conflict with columns defined in `http.request.query-param-fields`
-
-
#### Body Template
The `http.request.body-template` option provides a flexible way to construct HTTP request bodies for PUT and POST operations using a JSON template with placeholders.
@@ -385,10 +360,10 @@ This query creator allows you to populate API calls very flexibly. To do this ef
8) If you start from an OpenAPI specification that contains nested content required as a lookup join key, then use `http.request.body-template` to map top-level columns into that structure.
9) Response content is mapped to matching named top-level columns in the lookup table. You should arrange your table columns so that some are request columns (all top level) and some are response columns.
10) Use single quotes for the value of `http.request.body-template` so you do not need to escape the double quotes, and add newline characters for readability.
-11) If you want to enrich every event with the same API content, you can specify a placeholder as the complete URL the `url`, then use `http.request.url-map` to map it. In this scenario switching on caching is advised to avoid repeated identical API calls.
+11) If you want to enrich every event with the same API content, you can specify a placeholder as the complete URL the `url`, then use `http.request.url-map` to map it. In this scenario switching on caching is advised to avoid repeated identical API calls.
12) Note that columns in SQL tables (the DDL) do not have a natural way to distinguish between request and response fields. Where possible, use the API field name as column names in the DDL; this minimizes the number of columns you need to define.
13) The exception to 12) is when a response API field name is the same as a request API field **and** they have incompatible types. In this case, define the request column with a different name, then use `http.request.query-param-fields-with-key`, `http.request.body-template`, and/or `http.request.url-map` to provide the mapping to the API field.
-14) Note the columns representing the response are those that should be used for enrichment.
+14) Note the columns representing the response are those that should be used for enrichment.
### Format considerations
@@ -679,8 +654,8 @@ Metadata columns can be specified and hold http information. They are optional r
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
+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
diff --git a/docs/content/docs/connectors/table/http.md b/docs/content/docs/connectors/table/http.md
index c0a383a..95089cb 100644
--- a/docs/content/docs/connectors/table/http.md
+++ b/docs/content/docs/connectors/table/http.md
@@ -165,50 +165,48 @@ Or for REST POST method they will be converted to Json and used as request body.
Note the options with the prefix _http_ are the HTTP connector specific options, the others are Flink options.
-| Option | Required | Description/Value [...]
-|:-----------------------------------------------------------------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- [...]
-| 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. [...]
-| 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. [...]
-| lookup.partial-cache.expire-after-write | optional | The max time to live for each rows in lookup cache after writing into the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. [...]
-| lookup.partial-cache.expire-after-access | optional | The max time to live for each rows in lookup cache after accessing the entry in the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. [...]
-| lookup.partial-cache.cache-missing-key | optional | This is a boolean that defaults to true. Whether to store an empty value into the cache if the lookup key doesn't match any rows in the table. `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. [...]
-| lookup.max-retries | optional | The max retry times if the lookup failed; default is 3. See the following <a href="#lookup-cache">Lookup Cache</a> section for more detail. Set value 0 to disable retries. [...]
-| http.security.cert.server | optional | Comma separated paths to trusted HTTP server certificates that should be added to the connectors trust store. [...]
-| http.security.cert.client | optional | Path to trusted certificate that should be used by connector's HTTP client for mTLS communication. [...]
-| http.security.key.client | optional | Path to trusted private key that should be used by connector's HTTP client for mTLS communication. [...]
-| http.security.cert.server.allowSelfSigned | optional | Accept untrusted certificates for TLS communication. [...]
-| http.security.oidc.token.request | optional | OIDC `Token Request` body in `application/x-www-form-urlencoded` encoding [...]
-| http.security.oidc.token.endpoint.url | optional | OIDC `Token Endpoint` url, to which the token request will be issued [...]
-| http.security.oidc.token.expiry.reduction | optional | OIDC tokens will be requested if the current time is later than the cached token expiry time minus this value. [...]
-| http.source.lookup.continue-on-error | optional | When true, the flow will continue on errors, returning row content. When false (the default) the job ends on errors. [...]
-| http.source.lookup.request.timeout | optional | Sets HTTP request timeout for the lookup source as a Duration (e.g. `'30s'`, `'1min'`). Controls how long the HTTP client waits for a response before timing out a single request. If not specified, the default value of `30s` will be used. [...]
-| http.source.lookup.request.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 8 threads will be used. [...]
-| http.source.lookup.response.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup response processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 4 threads will be used. [...]
-| http.source.lookup.use-raw-authorization-header | optional | If set to `'true'`, uses the raw value set for the `Authorization` header, without transformation for Basic Authentication (base64, addition of "Basic " prefix). If not specified, defaults to `'false'`. [...]
-| http.source.lookup.request-callback | optional | Specify which `HttpLookupPostRequestCallback` implementation to use. By default, it is set to `slf4j-lookup-logger` corresponding to `Slf4jHttpLookupPostRequestCallback`. [...]
-| http.source.lookup.connection.timeout | optional | Source table connection timeout. Default - no value. [...]
-| http.source.lookup.http-version | optional | Version of HTTP to use for lookup http requests. The valid values are HTTP_1_1 and HTTP_2, which specify HTTP 1.1 or 2 respectively. This option may be required as HTTP_1_1, if the endpoint is HTTP 1.1, because some http 1.1 endpoints reject HTTP Version 2 calls, with 'Invalid HTTP request received' and 'HTTP/2 upgrade not supported'. [...]
-| http.source.lookup.success-codes | optional | Comma separated http codes considered as success response. Use [1-5]XX for groups and '!' character for excluding. The default is 2XX. [...]
-| http.source.lookup.retry-codes | optional | Comma separated http codes considered as transient errors. Use [1-5]XX for groups and '!' character for excluding. The default is 500,503,504. [...]
-| http.source.lookup.ignored-response-codes | optional | Comma separated http codes. Content for these responses will be ignored. Use [1-5]XX for groups and '!' character for excluding. Ignored responses together with `http.source.lookup.success-codes` are considered as successful. [...]
-| http.source.lookup.retry-strategy.type | optional | Auto retry strategy type: fixed-delay (default) or exponential-delay. [...]
-| http.source.lookup.retry-strategy.fixed-delay.delay | optional | Fixed-delay interval between retries. Default 1 second. Use with`lookup.max-retries` parameter. [...]
-| http.source.lookup.retry-strategy.exponential-delay.initial-backoff | optional | Exponential-delay initial delay. Default 1 second. [...]
-| http.source.lookup.retry-strategy.exponential-delay.max-backoff | optional | Exponential-delay maximum delay. Default 1 minute. Use with `lookup.max-retries` parameter. [...]
-| http.source.lookup.retry-strategy.exponential-delay.backoff-multiplier | optional | Exponential-delay multiplier. Default value 1.5 [...]
-| http.source.lookup.proxy.host | optional | Specify the hostname of the proxy. [...]
-| http.source.lookup.proxy.port | optional | Specify the port of the proxy. [...]
-| http.source.lookup.proxy.username | optional | Specify the username used for proxy authentication. [...]
-| http.source.lookup.proxy.password | optional | Specify the password used for proxy authentication. [...]
-| http.request.query-param-fields | optional | Used for the `http-generic-json-url` query creator. The names of the fields that will be mapped to query parameters. The parameters are separated by semicolons, such as `param1;param2`. [...]
-| http.request.query-param-fields-with-key | optional | A map of column names to query parameter keys. See the [Query Parameter Mapping](#query-parameter-mapping) section for details and examples. [...]
-| http.request.body-template | optional | Used for the `http-generic-json-url` query creator. A JSON template string for constructing the request body for PUT and POST operations. Use `{{fieldName}}` placeholders to reference top-level columns from the lookup table. Supports creating complex nested JSON structures with both placeholders and literal values. See the [Body Template](#body-template) section for details and examples. [...]
-| http.request.url-map | optional | Used for the `http-generic-json-url` query creator. The map of insert names to column names used as url segments. Parses a string as a map of strings. For example if there are table columns called `customerId` and `orderId`, then specifying value `customerId:cid,orderID:oid` and a url of https://myendpoint/customers/{cid}/orders/{oid} will mean that the url used for the lookup query will dynamically pic [...]
+| Option | Required | Description/Value |
+|:-----------------------------------------------------------------------|----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 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. |
+| 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. |
+| lookup.partial-cache.expire-after-write | optional | The max time to live for each rows in lookup cache after writing into the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. |
+| lookup.partial-cache.expire-after-access | optional | The max time to live for each rows in lookup cache after accessing the entry in the cache. Specify as a [Duration](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#duration). `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. |
+| lookup.partial-cache.cache-missing-key | optional | This is a boolean that defaults to true. Whether to store an empty value into the cache if the lookup key doesn't match any rows in the table. `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. |
+| lookup.max-retries | optional | The max retry times if the lookup failed; default is 3. See the following <a href="#lookup-cache">Lookup Cache</a> section for more detail. Set value 0 to disable retries. |
+| http.security.cert.server | optional | Comma separated paths to trusted HTTP server certificates that should be added to the connectors trust store. |
+| http.security.cert.client | optional | Path to trusted certificate that should be used by connector's HTTP client for mTLS communication. |
+| http.security.key.client | optional | Path to trusted private key that should be used by connector's HTTP client for mTLS communication. |
+| http.security.cert.server.allowSelfSigned | optional | Accept untrusted certificates for TLS communication. |
+| http.security.oidc.token.request | optional | OIDC `Token Request` body in `application/x-www-form-urlencoded` encoding |
+| http.security.oidc.token.endpoint.url | optional | OIDC `Token Endpoint` url, to which the token request will be issued |
+| http.security.oidc.token.expiry.reduction | optional | OIDC tokens will be requested if the current time is later than the cached token expiry time minus this value. |
+| http.source.lookup.continue-on-error | optional | When true, the flow will continue on errors, returning row content. When false (the default) the job ends on errors. |
+| http.source.lookup.request.timeout | optional | Sets HTTP request timeout for the lookup source as a Duration (e.g. `'30s'`, `'1min'`). Controls how long the HTTP client waits for a response before timing out a single request. If not specified, the default value of `30s` will be used. |
+| http.source.lookup.request.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 8 threads will be used. |
+| http.source.lookup.response.thread-pool.size | optional | Sets the size of pool thread for HTTP lookup response processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 4 threads will be used. |
+| http.source.lookup.use-raw-authorization-header | optional | If set to `'true'`, uses the raw value set for the `Authorization` header, without transformation for Basic Authentication (base64, addition of "Basic " prefix). If not specified, defaults to `'false'`. |
+| http.source.lookup.request-callback | optional | Specify which `HttpLookupPostRequestCallback` implementation to use. By default, it is set to `slf4j-lookup-logger` corresponding to `Slf4jHttpLookupPostRequestCallback`. |
+| http.source.lookup.connection.timeout | optional | Source table connection timeout. Default - no value. |
+| http.source.lookup.http-version | optional | Version of HTTP to use for lookup http requests. The valid values are HTTP_1_1 and HTTP_2, which specify HTTP 1.1 or 2 respectively. This option may be required as HTTP_1_1, if the endpoint is HTTP 1.1, because some http 1.1 endpoints reject HTTP Version 2 calls, with 'Invalid HTTP request received' and 'HTTP/2 upgrade not supported'. |
+| http.source.lookup.success-codes | optional | Comma separated http codes considered as success response. Use [1-5]XX for groups and '!' character for excluding. The default is 2XX. |
+| http.source.lookup.retry-codes | optional | Comma separated http codes considered as transient errors. Use [1-5]XX for groups and '!' character for excluding. The default is 500,503,504. |
+| http.source.lookup.ignored-response-codes | optional | Comma separated http codes. Content for these responses will be ignored. Use [1-5]XX for groups and '!' character for excluding. Ignored responses together with `http.source.lookup.success-codes` are considered as successful. |
+| http.source.lookup.retry-strategy.type | optional | Auto retry strategy type: fixed-delay (default) or exponential-delay. |
+| http.source.lookup.retry-strategy.fixed-delay.delay | optional | Fixed-delay interval between retries. Default 1 second. Use with`lookup.max-retries` parameter. |
+| http.source.lookup.retry-strategy.exponential-delay.initial-backoff | optional | Exponential-delay initial delay. Default 1 second. |
+| http.source.lookup.retry-strategy.exponential-delay.max-backoff | optional | Exponential-delay maximum delay. Default 1 minute. Use with `lookup.max-retries` parameter. |
+| http.source.lookup.retry-strategy.exponential-delay.backoff-multiplier | optional | Exponential-delay multiplier. Default value 1.5 |
+| http.source.lookup.proxy.host | optional | Specify the hostname of the proxy. |
+| http.source.lookup.proxy.port | optional | Specify the port of the proxy. |
+| http.source.lookup.proxy.username | optional | Specify the username used for proxy authentication. |
+| http.source.lookup.proxy.password | optional | Specify the password used for proxy authentication. |
+| http.request.body-template | optional | Used for the `http-generic-json-url` query creator. A JSON template string for constructing the request body for PUT and POST operations. Use `{{fieldName}}` placeholders to reference top-level columns from the lookup table. Supports creating complex nested JSON structures with both placeholders and literal values. See the [Body Template](#body-template) section for details and examples. |
+| http.request.url-map | optional | Used for the `http-generic-json-url` query creator. The map of insert names to column names used as url segments or query parameters. |
### Query Creators
@@ -247,28 +245,20 @@ The HTTP connector supplies a number of Query Creators that you can use to defin
The recommended Query creator for json is called _http-generic-json-url_, which allows column content to be mapped as URL, path, body and query parameter request values; it supports
POST, PUT and GET operations. This query creator allows you to issue json requests without needing to code
your own custom http connector. The mappings from columns to the json request are supplied in the query creator configuration
-parameters `http.request.query-param-fields`, `http.request.query-param-fields-with-key`, `http.request.body-template` and `http.request.url-map`.
-#### Query Parameter Mapping
+parameters `http.request.body-template` and `http.request.url-map`.
+#### Mapping the URL
-The `http.request.query-param-fields` and `http.request.query-param-fields-with-key` options provide flexible ways to map table columns to HTTP query parameters for GET requests.
+The `http.request.url-map` option provides a flexible way to map table columns to parts of the URL, either URL segments or HTTP query parameters.
+ Parses a string as a map of strings. For example if there are table columns called `customerId` and `orderId`,
+then specifying value `customerId:cid,orderID:oid` and a url of https://myendpoint/customers/{cid}?orders={oid} will mean that the url used for the
+lookup query will dynamically pickup the values for `customerId`, `orderId` and use them in the url e.g. https://myendpoint/customers/cid1?orders=oid1.
+The expected format of the map is: `key1:value1,key2:value2`.
-**Basic Query Parameter Mapping (`http.request.query-param-fields`):**
+As these values are being supplied as URL segments or part or query parameters, the connector url encodes that content so characters like spaces
+do not appear invalidly in the URL. In the case where the complete url is the insert then url encoding is not performed; the url needs to be valid and already
+properly url encoded as appropriate.
-Use this option to map column names directly to query parameters with the same name. The parameters are separated by semicolons.
-
-```sql
-'http.request.query-param-fields' = 'userId;orderId'
-```
-
-This will map the `userId` and `orderId` columns to query parameters `?userId=value1&orderId=value2`.
-
-**Advanced Query Parameter Mapping (`http.request.query-param-fields-with-key`):**
-
-Use this option when you need to use different names for query parameters than the table column names. This is necessary when request API fields have the same name as response fields and incompatible types.
-
-**Format:** `columnName1:queryParamKey1,columnName2:queryParamKey2`
-
-**Example Scenario:**
+**Example Scenario around clashing request and response columns:**
If your API response has a field `customer` defined as an object (complex type), but you need to send a customer ID as a query parameter with the same name `customer`, you can:
@@ -276,14 +266,6 @@ If your API response has a field `customer` defined as an object (complex type),
2. Keep the `customer` column for the response object
3. Map the request column to the query parameter:
-```sql
-'http.request.query-param-fields-with-key' = 'qp_customer:customer'
-```
-
-This will map the `qp_customer` column value to the query parameter `?customer=value`.
-
-**Complete Example:**
-
```sql
CREATE TABLE CustomerLookup (
qp_customer STRING, -- Request: customer ID as string
@@ -299,9 +281,9 @@ CREATE TABLE CustomerLookup (
) WITH (
'connector' = 'http',
'format' = 'json',
- 'url' = 'http://api.example.com/lookup',
+ 'url' = 'http://api.example.com/lookup?customer={qp_customer}&order={qp_order}',
'lookup-method' = 'GET',
- 'http.request.query-param-fields-with-key' = 'qp_customer:customer,qp_order:order'
+ 'http.request.url-map' = 'qp_customer:qp_customer,qp_order:qp_order'
)
```
@@ -312,13 +294,6 @@ GET http://api.example.com/lookup?customer=C123&order=O456
The response will populate the `customer` and `order` complex objects.
-**Important Notes for `http.request.query-param-fields-with-key`:**
-- Can only be used with GET requests
-- Column names and query parameter keys cannot be null or empty
-- Query parameter keys must be unique (no duplicates allowed)
-- Cannot conflict with columns defined in `http.request.query-param-fields`
-
-
#### Body Template
The `http.request.body-template` option provides a flexible way to construct HTTP request bodies for PUT and POST operations using a JSON template with placeholders.
diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactory.java
index 886fe5d..81a63a0 100644
--- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactory.java
+++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactory.java
@@ -63,8 +63,19 @@ public class BodyBasedRequestFactory extends RequestFactoryBase {
protected Builder setUpRequestMethod(LookupQueryInfo lookupQueryInfo) {
HttpRequest.Builder builder = super.setUpRequestMethod(lookupQueryInfo);
String body = lookupQueryInfo.getLookupQuery();
- builder.uri(constructUri(lookupQueryInfo))
- .method(methodName, BodyPublishers.ofString(body));
+ URI uri = constructUri(lookupQueryInfo);
+ try {
+ builder.uri(uri).method(methodName, BodyPublishers.ofString(body));
+ } catch (IllegalArgumentException e) {
+ throw new RuntimeException(
+ "Failed to build URI '"
+ + uri
+ + "', baseUrl is '"
+ + baseUrl
+ + "', Path parameters: "
+ + lookupQueryInfo.getPathBasedUrlParameters(),
+ e);
+ }
this.httpLogger.logRequestBody(body);
return builder;
}
@@ -86,7 +97,14 @@ public class BodyBasedRequestFactory extends RequestFactoryBase {
try {
return new URIBuilder(resolvedUrl.toString()).build();
} catch (URISyntaxException e) {
- throw new RuntimeException(e);
+ throw new RuntimeException(
+ "Failed to construct URI. Base URL: '"
+ + baseUrl
+ + "', Resolved URL: '"
+ + resolvedUrl.toString()
+ + "', Path parameters: "
+ + lookupQueryInfo.getPathBasedUrlParameters(),
+ e);
}
}
}
diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
index 15b56ad..5379736 100644
--- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
+++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
@@ -28,9 +28,11 @@ import org.apache.flink.util.FlinkRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
+import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpRequest.Builder;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Map;
@@ -135,7 +137,17 @@ public abstract class RequestFactoryBase implements HttpRequestFactory {
"Unexpected error while parsing the URL for path parameters.");
}
int endIndex = startIndex + pathParam.length();
- resolvedUrl = resolvedUrl.replace(startIndex, endIndex, entry.getValue());
+ String value = entry.getValue();
+ // Don't encode if the entire URL is just a placeholder (e.g., "{url}")
+ // Check if resolvedUrl starts with { and ends with }
+ String replacementValue;
+ if (resolvedUrl.charAt(0) == '{'
+ && resolvedUrl.charAt(resolvedUrl.length() - 1) == '}') {
+ replacementValue = value;
+ } else {
+ replacementValue = URLEncoder.encode(value, StandardCharsets.UTF_8);
+ }
+ resolvedUrl = resolvedUrl.replace(startIndex, endIndex, replacementValue);
}
}
return resolvedUrl;
diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreator.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreator.java
index 50d964a..b273dc5 100644
--- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreator.java
+++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreator.java
@@ -39,34 +39,29 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.Obje
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.URLEncoder;
-import java.nio.charset.StandardCharsets;
import java.util.Collection;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.StringJoiner;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* Generic JSON and URL query creator; in addition to be able to map columns to json requests, it
* allows url inserts to be mapped to column names using templating. <br>
- * For GETs, column names are mapped to query parameters. e.g. for <code>
- * GenericJsonAndUrlQueryCreator.REQUEST_PARAM_FIELDS</code> = "id1;id2" and url of http://base. At
- * lookup time with values of id1=1 and id2=2 a call of http/base?id1=1&id2=2 will be issued.
* <br>
- * For PUT and POST, parameters are mapped to the json body e.g. for REQUEST_PARAM_FIELDS =
- * "id1;id2" and url of http://base. At lookup time with values of id1=1 and id2=2 as call of
- * http/base will be issued with a json payload of {"id1":1,"id2":2} <br>
- * For all http methods, url segments can be used to include lookup up values. Using the map from
- * <code>GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP</code> which has a key of the insert and the
- * value of the associated column. e.g. for <code>GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP
+ * For PUT and POST, parameters are mapped to the json body e.g. for the body template "id1;id2" and
+ * url of http://base. At lookup time with values of id1=1 and id2=2 as call of http/base will be
+ * issued with a json payload of {"id1":1,"id2":2} <br>
+ * For all http methods, url segments and query parameters can be used to include lookup up values.
+ * Using the map from <code>GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP</code> which has a key of
+ * the insert name and the value of the associated column. e.g. for <code>
+ * GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP
* </code> = "key1":"col1" and url of http://base/{key1}. At lookup time with values of col1="aaaa"
- * a call of http/base/aaaa will be issued.
+ * a call of http/base/aaaa will be issued. For query parameters, the query param should be supplied
+ * in the URL with a place-holder that will be resolved using <code>
+ * GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP</code>
*/
@Slf4j
public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
@@ -79,7 +74,6 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
private boolean schemaOpened = false;
private LookupRow lookupRow;
private final String httpMethod;
- private final Map<String, String> requestQueryParamsMap;
private final Map<String, String> requestUrlMap;
private final String bodyTemplate;
@@ -88,7 +82,6 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
*
* @param httpMethod the requested http method
* @param serializationSchema serialization schema for RowData
- * @param requestQueryParamsMap map of column names to query parameter keys
* @param requestUrlMap url map
* @param bodyTemplate template string for request body with placeholders like {@code
* {{fieldName}}}
@@ -97,14 +90,12 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
public GenericJsonAndUrlQueryCreator(
final String httpMethod,
final SerializationSchema<RowData> serializationSchema,
- final Map<String, String> requestQueryParamsMap,
final Map<String, String> requestUrlMap,
final String bodyTemplate,
final LookupRow lookupRow) {
this.httpMethod = httpMethod;
this.serializationSchema = serializationSchema;
this.lookupRow = lookupRow;
- this.requestQueryParamsMap = requestQueryParamsMap;
this.requestUrlMap = requestUrlMap;
this.bodyTemplate = bodyTemplate;
}
@@ -119,7 +110,6 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
this.checkOpened();
final String lookupQuery;
- Map<String, String> bodyBasedUrlQueryParams = new HashMap<>();
final Collection<LookupArg> lookupArgs = lookupRow.convertToLookupArgs(lookupDataRow);
ObjectNode jsonObject;
try {
@@ -132,31 +122,13 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
log.error(message, e);
throw new RuntimeException(message, e);
}
- // Parameters are encoded as query params for GET and none GET.
- // Later code will turn these query params into the body for PUTs and POSTs
- // Use the map to rename columns to query param keys
- ObjectNode jsonObjectForQueryParams = ObjectMapperAdapter.instance().createObjectNode();
- for (Map.Entry<String, String> entry : this.requestQueryParamsMap.entrySet()) {
- String columnName = entry.getKey();
- String queryParamKey = entry.getValue();
- JsonNode value = jsonObject.get(columnName);
- if (value == null) {
- throw new IllegalArgumentException(
- String.format(
- "Query parameter mapping references column '%s' that does not exist in the lookup row. Available columns: %s",
- columnName, getAvailableColumnNames(jsonObject)));
- }
- jsonObjectForQueryParams.set(queryParamKey, value);
- }
- // TODO can we convertToQueryParameters for all ops
- // and not use/deprecate bodyBasedUrlQueryParams
+
if (httpMethod.equalsIgnoreCase("GET")) {
- // add the query parameters
- lookupQuery =
- convertToQueryParameters(
- jsonObjectForQueryParams, StandardCharsets.UTF_8.toString());
+ // For GET requests, query parameters are now handled via URL placeholders
+ // No query string is generated here
+ lookupQuery = "";
} else {
- // Body-based queries
+ // Body-based queries (POST/PUT)
// Check if body template is provided
if (bodyTemplate != null && !bodyTemplate.trim().isEmpty()) {
// Use template substitution
@@ -165,15 +137,12 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
// No template provided - no body (consistent with ElasticSearchLiteQueryCreator)
lookupQuery = "";
}
- // body parameters
- // use the request json object to scope the required fields and the lookupArgs as values
- bodyBasedUrlQueryParams = createBodyBasedParams(lookupArgs, jsonObjectForQueryParams);
}
// add the path map
final Map<String, String> pathBasedUrlParams =
createPathBasedParams(lookupArgs, requestUrlMap);
- return new LookupQueryInfo(lookupQuery, bodyBasedUrlQueryParams, pathBasedUrlParams);
+ return new LookupQueryInfo(lookupQuery, null, pathBasedUrlParams);
}
/**
@@ -210,18 +179,6 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
return result.toString();
}
- /**
- * Get a comma-separated list of available column names from the JSON object.
- *
- * @param jsonObject the JSON object containing field names
- * @return comma-separated string of available column names
- */
- private String getAvailableColumnNames(ObjectNode jsonObject) {
- StringJoiner joiner = new StringJoiner(", ");
- jsonObject.fieldNames().forEachRemaining(joiner::add);
- return joiner.toString();
- }
-
/**
* Create a Row from a RowData and DataType.
*
@@ -291,40 +248,6 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
return mapOfJsonKeyToLookupArg;
}
- /**
- * Convert json object to query params string.
- *
- * @param jsonObject supplies json object
- * @param enc encoding string - used in unit test to drive unsupported encoding
- * @return query params string
- */
- @VisibleForTesting
- static String convertToQueryParameters(final ObjectNode jsonObject, String enc) {
- Preconditions.checkNotNull(jsonObject);
-
- final StringJoiner result = new StringJoiner("&");
- jsonObject
- .fields()
- .forEachRemaining(
- field -> {
- final String fieldName = field.getKey();
- final String fieldValue = field.getValue().asText();
-
- try {
- result.add(fieldName + "=" + URLEncoder.encode(fieldValue, enc));
- } catch (UnsupportedEncodingException e) {
- final String message =
- "Failed to encode the value of the query parameter name "
- + fieldName
- + ": "
- + fieldValue;
- throw new RuntimeException(message, e);
- }
- });
-
- return result.toString();
- }
-
private void checkOpened() {
if (!this.schemaOpened) {
try {
diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java
index 94ec897..d7b08db 100644
--- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java
+++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java
@@ -33,9 +33,6 @@ import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.SerializationFormatFactory;
import org.apache.flink.table.types.DataType;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -59,22 +56,6 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
public static final String ID = "http-generic-json-url";
- public static final ConfigOption<List<String>> REQUEST_QUERY_PARAM_FIELDS =
- key("http.request.query-param-fields")
- .stringType()
- .asList()
- .defaultValues() // default to empty list so we do not need to check for null
- .withDescription(
- "The names of the fields that will be mapped to query parameters."
- + " The parameters are separated by semicolons,"
- + " such as 'param1;param2'.");
-
- public static final ConfigOption<Map<String, String>> REQUEST_QUERY_PARAM_FIELDS_WITH_KEY =
- ConfigOptions.key("http.request.query-param-fields-with-key")
- .mapType()
- .noDefaultValue()
- .withDescription("A map of column names to query parameter keys.");
-
public static final ConfigOption<Map<String, String>> REQUEST_URL_MAP =
ConfigOptions.key("http.request.url-map")
.mapType()
@@ -115,23 +96,6 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
// Validate configuration based on HTTP method
validateConfigurationForHttpMethod(httpMethod, readableConfig);
- // Handle query param fields - merge both list and map formats
- Map<String, String> requestQueryParamsMap = new LinkedHashMap<>();
-
- // First, add entries from list format (key = value for backward compatibility)
- final List<String> requestQueryParamsFields =
- readableConfig.get(REQUEST_QUERY_PARAM_FIELDS);
- requestQueryParamsFields.forEach(field -> requestQueryParamsMap.put(field, field));
-
- // Then, add entries from map format
- if (readableConfig.getOptional(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY).isPresent()) {
- Map<String, String> queryParamFieldsWithKey =
- readableConfig.get(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY);
-
- validateQueryParamMap(queryParamFieldsWithKey, readableConfig);
- requestQueryParamsMap.putAll(queryParamFieldsWithKey);
- }
-
Map<String, String> requestUrlMap = readableConfig.get(REQUEST_URL_MAP);
String bodyTemplate = readableConfig.getOptional(REQUEST_BODY_TEMPLATE).orElse(null);
@@ -142,12 +106,7 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
// create using config parameter values and specify serialization
// schema from json format.
return new GenericJsonAndUrlQueryCreator(
- httpMethod,
- jsonSerializationSchema,
- requestQueryParamsMap,
- requestUrlMap,
- bodyTemplate,
- lookupRow);
+ httpMethod, jsonSerializationSchema, requestUrlMap, bodyTemplate, lookupRow);
}
@Override
@@ -162,87 +121,7 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
@Override
public Set<ConfigOption<?>> optionalOptions() {
- return Set.of(
- REQUEST_QUERY_PARAM_FIELDS,
- REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- REQUEST_URL_MAP,
- REQUEST_BODY_TEMPLATE);
- }
-
- /**
- * Validates the query parameter map to ensure all constraints are met.
- *
- * <ul>
- * <li>Column names (keys) are not null or empty
- * <li>Query param keys (values) are not null or empty
- * <li>Query param keys do not conflict with old REQUEST_QUERY_PARAM_FIELDS list
- * <li>Query param keys are unique (no duplicates)
- * <li>Column names in new map do not conflict with column names in old list
- * </ul>
- *
- * @param queryParamMap the map to validate
- * @param readableConfig the configuration to check for conflicts
- * @throws IllegalArgumentException if validation fails
- */
- private void validateQueryParamMap(
- Map<String, String> queryParamMap, ReadableConfig readableConfig) {
- if (queryParamMap == null || queryParamMap.isEmpty()) {
- return; // Empty map is valid
- }
-
- // Get the list format for conflict checking
- final List<String> queryParamFields = readableConfig.get(REQUEST_QUERY_PARAM_FIELDS);
-
- // Track seen query param keys to detect duplicates
- Set<String> seenQueryParamKeys = new HashSet<>();
-
- for (Map.Entry<String, String> entry : queryParamMap.entrySet()) {
- String columnName = entry.getKey();
- String queryParamKey = entry.getValue();
-
- // Validate column name (map key)
- if (columnName == null || columnName.trim().isEmpty()) {
- throw new IllegalArgumentException(
- "Column name in "
- + REQUEST_QUERY_PARAM_FIELDS_WITH_KEY.key()
- + " cannot be null or empty");
- }
-
- // Validate query param key (map value)
- if (queryParamKey == null || queryParamKey.trim().isEmpty()) {
- throw new IllegalArgumentException(
- "Query parameter key for column '"
- + columnName
- + "' in "
- + REQUEST_QUERY_PARAM_FIELDS_WITH_KEY.key()
- + " cannot be null or empty");
- }
-
- // Check for duplicate query param keys
- if (seenQueryParamKeys.contains(queryParamKey)) {
- throw new IllegalArgumentException(
- "Duplicate query parameter key '"
- + queryParamKey
- + "' in "
- + REQUEST_QUERY_PARAM_FIELDS_WITH_KEY.key()
- + ". Each query parameter key must be unique.");
- }
- seenQueryParamKeys.add(queryParamKey);
-
- // Check for conflicts with list format - both query param keys and column names
- if (!queryParamFields.isEmpty()
- && (queryParamFields.contains(queryParamKey)
- || queryParamFields.contains(columnName))) {
- throw new IllegalArgumentException(
- (queryParamFields.contains(queryParamKey)
- ? "Query parameter key '" + queryParamKey
- : "Column name '" + columnName)
- + "' in "
- + REQUEST_QUERY_PARAM_FIELDS_WITH_KEY.key()
- + " conflicts with existing columns defined in "
- + REQUEST_QUERY_PARAM_FIELDS.key());
- }
- }
+ return Set.of(REQUEST_URL_MAP, REQUEST_BODY_TEMPLATE);
}
/**
@@ -251,7 +130,6 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
* <p>Rules:
*
* <ul>
- * <li>Query parameter configs can only be used with GET
* <li>Body template can only be used with POST/PUT
* </ul>
*
@@ -261,9 +139,6 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
*/
private void validateConfigurationForHttpMethod(
String httpMethod, ReadableConfig readableConfig) {
- boolean hasQueryParams = !readableConfig.get(REQUEST_QUERY_PARAM_FIELDS).isEmpty();
- boolean hasQueryParamsWithKey =
- readableConfig.getOptional(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY).isPresent();
boolean hasBodyTemplate =
readableConfig.getOptional(REQUEST_BODY_TEMPLATE).isPresent()
&& !readableConfig.get(REQUEST_BODY_TEMPLATE).trim().isEmpty();
@@ -274,24 +149,9 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
"Body template configuration ("
+ REQUEST_BODY_TEMPLATE.key()
+ ") cannot be used with GET method. "
- + "For GET requests, use "
- + REQUEST_QUERY_PARAM_FIELDS.key()
- + " or "
- + REQUEST_QUERY_PARAM_FIELDS_WITH_KEY.key()
- + " instead.");
- }
- } else {
- // POST/PUT
- if (hasQueryParams || hasQueryParamsWithKey) {
- throw new IllegalArgumentException(
- "Query parameter configuration ("
- + REQUEST_QUERY_PARAM_FIELDS.key()
- + " or "
- + REQUEST_QUERY_PARAM_FIELDS_WITH_KEY.key()
- + ") can only be used with GET method. "
- + "For POST/PUT requests, use "
- + REQUEST_BODY_TEMPLATE.key()
- + " instead.");
+ + "For GET requests, use query parameters in the URL with "
+ + REQUEST_URL_MAP.key()
+ + " to map columns to placeholders.");
}
}
}
diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
index b6d3950..056aff5 100644
--- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
+++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
@@ -162,6 +162,64 @@ public class BodyBasedRequestFactoryTest {
Map.of("param1", "value1", "param2", "value2"),
"http://service/{param1}/param2/{param2}",
lookupMethod,
- "http://service/value1/param2/value2?param3=value3¶m4=value4"));
+ "http://service/value1/param2/value2?param3=value3¶m4=value4"),
+ // URL encoding: path param with spaces
+ new TestSpec(
+ null,
+ Map.of("param1", "hello world"),
+ "http://service/{param1}",
+ lookupMethod,
+ "http://service/hello+world"),
+ // URL encoding: path param with special characters
+ new TestSpec(
+ null,
+ Map.of("param1", "user@example.com"),
+ "http://service/{param1}",
+ lookupMethod,
+ "http://service/user%40example.com"),
+ // URL encoding: path param with slash
+ new TestSpec(
+ null,
+ Map.of("param1", "path/to/resource"),
+ "http://service/{param1}",
+ lookupMethod,
+ "http://service/path%2Fto%2Fresource"),
+ // URL encoding: multiple path params with special characters
+ new TestSpec(
+ null,
+ Map.of("param1", "hello world", "param2", "user@example.com"),
+ "http://service/{param1}/users/{param2}",
+ lookupMethod,
+ "http://service/hello+world/users/user%40example.com"),
+ // URL encoding: query param with special characters (?, &, ;, space)
+ new TestSpec(
+ Map.of("param1", "value with ?&; chars"),
+ null,
+ "http://service",
+ lookupMethod,
+ "http://service?param1=value+with+%3F%26%3B+chars"),
+ // URL encoding: multiple query params with special characters
+ new TestSpec(
+ Map.of("param1", "hello world", "param2", "a=b&c=d"),
+ null,
+ "http://service",
+ lookupMethod,
+ "http://service?param1=hello+world¶m2=a%3Db%26c%3Dd"),
+ // URL encoding: combined path and query params with special characters
+ new TestSpec(
+ Map.of("query1", "value?test"),
+ Map.of("path1", "user@domain"),
+ "http://service/{path1}",
+ lookupMethod,
+ "http://service/user%40domain?query1=value%3Ftest"),
+ // Complete URL replacement with URL-encoded parts
+ new TestSpec(
+ null,
+ Map.of(
+ "url",
+ "https://api.example.com/search?q=hello%20world&filter=type%3Dbook&sort=date%3Adesc"),
+ "{url}",
+ lookupMethod,
+ "https://api.example.com/search?q=hello%20world&filter=type%3Dbook&sort=date%3Adesc"));
}
}
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 27194a9..769303d 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
@@ -1169,8 +1169,8 @@ class HttpLookupTableSourceITCaseTest {
// - Orders table has "id" column (string)
// - API expects query param named "customer" (string)
// - API response has field "customer" (ROW type with nested fields)
- // Solution: Use http.request.query-param-fields-with-key to map:
- // - Lookup table column "id" → query parameter "customer"
+ // Solution: Use http.request.url-map with URL placeholders to map:
+ // - Lookup table column "id" → query parameter "customer" via {customer} placeholder
// - Lookup table column "customer" → response field "customer" (ROW type)
int serverPort2 = WireMockServerPortAllocator.getServerPort();
@@ -1235,15 +1235,14 @@ class HttpLookupTableSourceITCaseTest {
+ "'lookup-method' = 'GET',"
+ "'url' = 'http://localhost:"
+ serverPort2
- + "/client',"
+ + "/client?customer={customer}&id2={id2}',"
+ "'http.source.lookup.header.Content-Type' = 'application/json',"
+ "'asyncPolling' = 'true',"
+ "'http.source.lookup.query-creator' = 'http-generic-json-url',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
+ "'table.exec.async-lookup.timeout' = '20s',"
- // Map "id" column to "customer" query parameter
- + "'http.request.query-param-fields-with-key' = 'id:customer',"
- + "'http.request.query-param-fields' = 'id2'"
+ // Map "id" column to "customer" placeholder and "id2" to "id2" placeholder
+ + "'http.request.url-map' = 'id:customer,id2:id2'"
+ ")";
tEnv.executeSql(sourceTable);
@@ -1336,7 +1335,6 @@ class HttpLookupTableSourceITCaseTest {
// Template maps body_customer to "customer" in request body
+ "'http.request.body-template' = '{\"customer\": {{body_customer}}, \"id2\": {{id2}}}'"
+ ")";
-
tEnv.executeSql(sourceTable);
tEnv.executeSql(lookupTable);
diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
index d592d8d..118b33f 100644
--- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
+++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
@@ -40,7 +40,6 @@ import java.util.List;
import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.LOOKUP_REQUEST_FORMAT;
import static org.apache.flink.connector.http.table.lookup.HttpLookupTableSourceFactory.row;
import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_BODY_TEMPLATE;
-import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS;
import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_URL_MAP;
import static org.apache.flink.connector.http.table.lookup.querycreators.QueryCreatorUtils.getTableContext;
import static org.assertj.core.api.Assertions.assertThat;
@@ -78,7 +77,6 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
CustomJsonFormatFactory.IDENTIFIER,
CustomJsonFormatFactory.REQUIRED_OPTION),
"optionValue");
- this.config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key1"));
// with sync
createUsingFactory(false);
// with async
@@ -116,7 +114,7 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
LookupQueryInfo lookupQueryInfo = lookupQueryCreator.createLookupQuery(lookupRowData);
assertThat(CustomJsonFormatFactory.requiredOptionsWereUsed).isTrue();
- assertThat(lookupQueryInfo.hasLookupQuery()).isTrue();
+ assertThat(lookupQueryInfo.hasLookupQuery()).isFalse();
assertThat(lookupQueryInfo.hasBodyBasedUrlQueryParameters()).isFalse();
assertThat(lookupQueryInfo.hasPathBasedUrlParameters()).isFalse();
}
@@ -125,7 +123,6 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
void optionsTests() {
GenericJsonAndUrlQueryCreatorFactory factory = new GenericJsonAndUrlQueryCreatorFactory();
assertThat(factory.requiredOptions()).isEmpty();
- assertThat(factory.optionalOptions()).contains(REQUEST_QUERY_PARAM_FIELDS);
assertThat(factory.optionalOptions()).contains(REQUEST_BODY_TEMPLATE);
assertThat(factory.optionalOptions()).contains(REQUEST_URL_MAP);
}
@@ -150,198 +147,6 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
.createLookupQueryCreator(config, lookupRow, tableContext);
}
- @Test
- void testValidationRejectsNullColumnName() {
- // GIVEN - Map with null column name
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put(null, "qp1");
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Column name")
- .hasMessageContaining("cannot be null or empty");
- }
-
- @Test
- void testValidationRejectsEmptyColumnName() {
- // GIVEN - Map with empty column name
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("", "qp1");
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Column name")
- .hasMessageContaining("cannot be null or empty");
- }
-
- @Test
- void testValidationRejectsNullQueryParamKey() {
- // GIVEN - Map with null query param key
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("key1", null);
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Query parameter key for column 'key1'")
- .hasMessageContaining("cannot be null or empty");
- }
-
- @Test
- void testValidationRejectsEmptyQueryParamKey() {
- // GIVEN - Map with empty query param key
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("key1", " ");
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Query parameter key for column 'key1'")
- .hasMessageContaining("cannot be null or empty");
- }
-
- @Test
- void testValidationRejectsConflictBetweenQueryParamKeyAndListFormat() {
- // GIVEN - REQUEST_QUERY_PARAM_FIELDS_WITH_KEY with query param key that conflicts with
- // REQUEST_QUERY_PARAM_FIELDS
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("qp1", "qp2"));
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("key1", "qp1"); // qp1 is already in REQUEST_QUERY_PARAM_FIELDS
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Query parameter key 'qp1'")
- .hasMessageContaining("conflicts with existing columns");
- }
-
- @Test
- void testValidationRejectsConflictBetweenColumnNameAndListFormat() {
- // GIVEN - REQUEST_QUERY_PARAM_FIELDS_WITH_KEY with column name that conflicts with
- // REQUEST_QUERY_PARAM_FIELDS
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key1", "qp2"));
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("key1", "customParam"); // key1 is already in REQUEST_QUERY_PARAM_FIELDS
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Column name 'key1'")
- .hasMessageContaining("conflicts with existing columns");
- }
-
- @Test
- void testValidationRejectsDuplicateQueryParamKeys() {
- // GIVEN - Map with duplicate query param keys (different columns mapping to same key)
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- java.util.Map<String, String> queryParamMap = new java.util.LinkedHashMap<>();
- queryParamMap.put("customerId", "id");
- queryParamMap.put("orderId", "id"); // Same key "id" used twice
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "customerId",
- RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "orderId",
- RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 1)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Duplicate query parameter key 'id'")
- .hasMessageContaining("must be unique");
- }
-
@Test
void testBodyTemplateWithNestedStructure() {
// GIVEN - Body template with nested JSON structure
@@ -415,76 +220,6 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
assertThat(creator).isNotNull();
}
- @Test
- void testValidationRejectsQueryParamsWithPost() {
- // GIVEN - POST request with query param fields (old format)
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "POST");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key1"));
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Query parameter configuration")
- .hasMessageContaining("can only be used with GET method");
- }
-
- @Test
- void testValidationRejectsQueryParamsWithKeyWithPost() {
- // GIVEN - POST request with query param fields with key (new format)
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "POST");
- java.util.Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("key1", "qp1");
- config.set(
- GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY,
- queryParamMap);
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Query parameter configuration")
- .hasMessageContaining("can only be used with GET method");
- }
-
- @Test
- void testValidationRejectsQueryParamsWithPut() {
- // GIVEN - PUT request with query param fields
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "PUT");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key1"));
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
-
- // WHEN/THEN - Should throw IllegalArgumentException
- assertThatThrownBy(
- () ->
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Query parameter configuration")
- .hasMessageContaining("can only be used with GET method");
- }
-
@Test
void testValidationRejectsBodyTemplateWithGet() {
// GIVEN - GET request with body template
@@ -507,27 +242,6 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
.hasMessageContaining("cannot be used with GET method");
}
- @Test
- void testValidationAllowsQueryParamsWithGet() {
- // GIVEN - GET request with query param fields (should succeed)
- Configuration config = new Configuration();
- config.set(HttpLookupConnectorOptions.LOOKUP_METHOD, "GET");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key1"));
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
- lookupRow.setLookupPhysicalRowDataType(
- row(List.of(DataTypes.FIELD("key1", DataTypes.STRING()))));
-
- // WHEN/THEN - Should succeed
- LookupQueryCreator creator =
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext);
- assertThat(creator).isNotNull();
- }
-
@Test
void testValidationRejectsBodyTemplateWithDefaultGetMethod() {
// GIVEN - No lookup-method specified (defaults to GET) with body template
@@ -551,27 +265,6 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
.hasMessageContaining("cannot be used with GET method");
}
- @Test
- void testValidationAllowsQueryParamsWithDefaultGetMethod() {
- // GIVEN - No lookup-method specified (defaults to GET) with query param fields
- Configuration config = new Configuration();
- // Note: NOT setting LOOKUP_METHOD - should default to GET
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key1"));
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
- lookupRow.setLookupPhysicalRowDataType(
- row(List.of(DataTypes.FIELD("key1", DataTypes.STRING()))));
-
- // WHEN/THEN - Should succeed because default GET allows query parameters
- LookupQueryCreator creator =
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(config, lookupRow, tableContext);
- assertThat(creator).isNotNull();
- }
-
@Test
void testValidationAllowsBodyTemplateWithPost() {
// GIVEN - POST request with body template (should succeed)
diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorTest.java
index 9f7620d..defb4ed 100644
--- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorTest.java
+++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorTest.java
@@ -34,7 +34,6 @@ import org.apache.flink.types.Row;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -46,8 +45,6 @@ import java.util.Map;
import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.LOOKUP_METHOD;
import static org.apache.flink.connector.http.table.lookup.HttpLookupTableSourceFactory.row;
import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_BODY_TEMPLATE;
-import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS;
-import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS_WITH_KEY;
import static org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_URL_MAP;
import static org.apache.flink.connector.http.table.lookup.querycreators.QueryCreatorUtils.getTableContext;
import static org.assertj.core.api.Assertions.assertThat;
@@ -59,7 +56,6 @@ class GenericJsonAndUrlQueryCreatorTest {
private static final String KEY_1 = "key1";
private static final String KEY_2 = "key2";
private static final String VALUE = "val1";
- private static final List<String> QUERY_PARAMS = List.of(KEY_1);
private static final Map<String, String> urlParams = Map.of(KEY_1, KEY_1);
private static final DataType DATATYPE_1 =
row(List.of(DataTypes.FIELD(KEY_1, DataTypes.STRING())));
@@ -114,7 +110,6 @@ class GenericJsonAndUrlQueryCreatorTest {
// GIVEN
LookupRow lookupRow = getLookupRow(KEY_1, KEY_2);
Configuration config = new Configuration();
- config.set(REQUEST_QUERY_PARAM_FIELDS, QUERY_PARAMS);
config.set(REQUEST_URL_MAP, urlParams);
config.set(LOOKUP_METHOD, "GET");
lookupRow.setLookupPhysicalRowDataType(DATATYPE_1_2);
@@ -127,11 +122,12 @@ class GenericJsonAndUrlQueryCreatorTest {
getTableContext(config, RESOLVED_SCHEMA));
// WHEN
GenericRowData lookupRowData =
- GenericRowData.of(StringData.fromString("val1"), StringData.fromString("val2"));
+ GenericRowData.of(StringData.fromString(VALUE), StringData.fromString("val2"));
LookupQueryInfo createdQuery =
genericJsonAndUrlQueryCreator.createLookupQuery(lookupRowData);
- // THEN - Only KEY_1 is in QUERY_PARAMS, so only key1=val1 in query string
- assertThat(createdQuery.getLookupQuery()).isEqualTo("key1=val1");
+ // THEN - no lookup query
+ assertThat(createdQuery.getLookupQuery()).isEqualTo("");
+ assertThat(createdQuery.getPathBasedUrlParameters()).isEqualTo(Map.of(KEY_1, VALUE));
}
@Test
@@ -166,24 +162,6 @@ class GenericJsonAndUrlQueryCreatorTest {
.isInstanceOf(RuntimeException.class);
}
- @Test
- void convertToQueryParametersUnsupportedEncodingTest() {
- // GIVEN
- PersonBean person = new PersonBean("aaa", "bbb");
- JsonNode personNode;
- try {
- personNode = OBJECT_MAPPER.readTree(OBJECT_MAPPER.writeValueAsString(person));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- // WHEN/THEN
- assertThatThrownBy(
- () ->
- GenericJsonAndUrlQueryCreator.convertToQueryParameters(
- (ObjectNode) personNode, "bad encoding"))
- .isInstanceOf(RuntimeException.class);
- }
-
@Test
void rowDataToRowTest() {
// GIVEN
@@ -316,7 +294,7 @@ class GenericJsonAndUrlQueryCreatorTest {
// GIVEN - GET request with body template (should be rejected by validation)
Configuration config = new Configuration();
config.set(LOOKUP_METHOD, "GET");
- config.set(REQUEST_QUERY_PARAM_FIELDS, QUERY_PARAMS);
+ // config.set(REQUEST_QUERY_PARAM_FIELDS, QUERY_PARAMS);
config.set(REQUEST_BODY_TEMPLATE, "{\"userId\":{{key1}}}");
LookupRow lookupRow = getLookupRow(KEY_1);
@@ -564,190 +542,6 @@ class GenericJsonAndUrlQueryCreatorTest {
assertThat(literalObject.get("key").asText()).isEqualTo("value");
}
- @Test
- public void testQueryParamFieldsWithKeyForGet() {
- // GIVEN - Use new map format to rename query params
- Configuration config = new Configuration();
- config.set(LOOKUP_METHOD, "GET");
- Map<String, String> queryParamMap = Map.of("key1", "qp1");
- config.set(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY, queryParamMap);
-
- LookupRow lookupRow = getLookupRow(KEY_1);
- GenericJsonAndUrlQueryCreator creator =
- (GenericJsonAndUrlQueryCreator)
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(
- config,
- lookupRow,
- getTableContext(config, RESOLVED_SCHEMA));
-
- // WHEN
- LookupQueryInfo createdQuery = creator.createLookupQuery(ROWDATA);
-
- // THEN - Query params should use renamed key
- assertThat(createdQuery.hasLookupQuery()).isTrue();
- String queryString = createdQuery.getLookupQuery();
- assertThat(queryString).isEqualTo("qp1=val1");
- // Should NOT contain original column name
- assertThat(queryString).doesNotContain("key1=");
- }
-
- @Test
- public void testBackwardCompatibilityWithOldListFormat() {
- // GIVEN - Use old list format (should still work)
- Configuration config = new Configuration();
- config.set(LOOKUP_METHOD, "GET");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of(KEY_1));
-
- LookupRow lookupRow = getLookupRow(KEY_1);
- GenericJsonAndUrlQueryCreator creator =
- (GenericJsonAndUrlQueryCreator)
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(
- config,
- lookupRow,
- getTableContext(config, RESOLVED_SCHEMA));
-
- // WHEN
- LookupQueryInfo createdQuery = creator.createLookupQuery(ROWDATA);
-
- // THEN - Should work as before (column name = query param key)
- assertThat(createdQuery.hasLookupQuery()).isTrue();
- assertThat(createdQuery.getLookupQuery()).isEqualTo("key1=val1");
- }
-
- @Test
- public void testBothFormatsAreMergedWhenNoConflict() {
- // GIVEN - Both old list and new map are provided with different fields
- Configuration config = new Configuration();
- config.set(LOOKUP_METHOD, "GET");
- config.set(REQUEST_QUERY_PARAM_FIELDS, List.of("key2")); // Old format for key2
- Map<String, String> queryParamMap = Map.of("key1", "renamed_key"); // New format for key1
- config.set(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY, queryParamMap);
-
- // Create schema with both key1 and key2
- ResolvedSchema schemaWithBothKeys =
- ResolvedSchema.of(
- Column.physical("key1", DataTypes.STRING()),
- Column.physical("key2", DataTypes.STRING()));
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key1", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "key2", RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 1)));
-
- GenericJsonAndUrlQueryCreator creator =
- (GenericJsonAndUrlQueryCreator)
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(
- config,
- lookupRow,
- getTableContext(config, schemaWithBothKeys));
-
- // WHEN
- GenericRowData lookupRowData =
- GenericRowData.of(StringData.fromString("val1"), StringData.fromString("val2"));
- LookupQueryInfo createdQuery = creator.createLookupQuery(lookupRowData);
-
- // THEN - Both formats should be merged: key1 renamed to renamed_key, key2 stays as key2
- assertThat(createdQuery.hasLookupQuery()).isTrue();
- String queryString = createdQuery.getLookupQuery();
- assertThat(queryString).contains("key2=val2"); // From old list format
- assertThat(queryString).contains("renamed_key=val1"); // From new map format
- assertThat(queryString).doesNotContain("key1="); // key1 should be renamed
- }
-
- @Test
- public void testQueryParamFieldsWithKeyMultipleFields() {
- // GIVEN - Multiple fields with different renamed keys
- Configuration config = new Configuration();
- config.set(LOOKUP_METHOD, "GET");
- Map<String, String> queryParamMap =
- Map.of(
- "customerId", "cid",
- "orderName", "oname",
- "status", "st");
- config.set(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY, queryParamMap);
-
- // Create schema with all three fields
- ResolvedSchema schemaWithAllFields =
- ResolvedSchema.of(
- Column.physical("customerId", DataTypes.STRING()),
- Column.physical("orderName", DataTypes.STRING()),
- Column.physical("status", DataTypes.STRING()));
-
- LookupRow lookupRow = new LookupRow();
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "customerId",
- RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 0)));
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "orderName",
- RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 1)));
- lookupRow.addLookupEntry(
- new RowDataSingleValueLookupSchemaEntry(
- "status",
- RowData.createFieldGetter(DataTypes.STRING().getLogicalType(), 2)));
-
- GenericJsonAndUrlQueryCreator creator =
- (GenericJsonAndUrlQueryCreator)
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(
- config,
- lookupRow,
- getTableContext(config, schemaWithAllFields));
-
- // WHEN
- GenericRowData lookupRowData =
- GenericRowData.of(
- StringData.fromString("123"),
- StringData.fromString("Order1"),
- StringData.fromString("active"));
- LookupQueryInfo createdQuery = creator.createLookupQuery(lookupRowData);
-
- // THEN - All fields should be renamed
- assertThat(createdQuery.hasLookupQuery()).isTrue();
- String queryString = createdQuery.getLookupQuery();
- assertThat(queryString).contains("cid=123");
- assertThat(queryString).contains("oname=Order1");
- assertThat(queryString).contains("st=active");
- // Should NOT contain original column names
- assertThat(queryString).doesNotContain("customerId=");
- assertThat(queryString).doesNotContain("orderName=");
- assertThat(queryString).doesNotContain("status=");
- }
-
- @Test
- public void testQueryParamFieldsWithKeyMissingColumn() {
- // GIVEN - Query param map references a column that doesn't exist
- Configuration config = new Configuration();
- config.set(LOOKUP_METHOD, "GET");
- Map<String, String> queryParamMap = new java.util.HashMap<>();
- queryParamMap.put("misspelledCol", "customer"); // misspelledCol doesn't exist
- config.set(REQUEST_QUERY_PARAM_FIELDS_WITH_KEY, queryParamMap);
-
- LookupRow lookupRow = getLookupRow(KEY_1); // Only has key1
-
- GenericJsonAndUrlQueryCreator creator =
- (GenericJsonAndUrlQueryCreator)
- new GenericJsonAndUrlQueryCreatorFactory()
- .createLookupQueryCreator(
- config,
- lookupRow,
- getTableContext(config, RESOLVED_SCHEMA));
-
- // WHEN/THEN - Should throw IllegalArgumentException with helpful message
- assertThatThrownBy(() -> creator.createLookupQuery(ROWDATA))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("misspelledCol")
- .hasMessageContaining("does not exist")
- .hasMessageContaining("Available columns:");
- }
-
// Helper methods
private static GenericRowData getRowData(int numFields, String value) {
if (numFields == 1) {
@@ -771,7 +565,6 @@ class GenericJsonAndUrlQueryCreatorTest {
private Configuration getConfigurationForGet() {
Configuration config = new Configuration();
- config.set(REQUEST_QUERY_PARAM_FIELDS, QUERY_PARAMS);
config.set(REQUEST_URL_MAP, urlParams);
config.set(LOOKUP_METHOD, "GET");
return config;
@@ -786,16 +579,14 @@ class GenericJsonAndUrlQueryCreatorTest {
}
private void validateCreatedQueryForGet(LookupQueryInfo createdQuery) {
- assertThat(createdQuery.hasLookupQuery()).isTrue();
- assertThat(createdQuery.getLookupQuery()).isEqualTo("key1=val1");
+ assertThat(createdQuery.hasLookupQuery()).isFalse();
assertThat(createdQuery.hasBodyBasedUrlQueryParameters()).isFalse();
assertThat(createdQuery.hasPathBasedUrlParameters()).isTrue();
}
private void validateCreatedQueryForPutAndPost(LookupQueryInfo createdQuery) {
- // When no template is provided, body is empty (no body sent)
+ // no body template so no lookup query
assertThat(createdQuery.hasLookupQuery()).isFalse();
- assertThat(createdQuery.getLookupQuery()).isEqualTo("");
assertThat(createdQuery.hasBodyBasedUrlQueryParameters()).isFalse();
assertThat(createdQuery.hasPathBasedUrlParameters()).isTrue();
}