Skip to content

FLINK-39040 Additional body json for GenericJsonAndUrlQueryCreator #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ferenc-csaky merged 6 commits into from
Feb 11, 2026
Merged

FLINK-39040 Additional body json for GenericJsonAndUrlQueryCreator #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6ab4bbd
FLINK-39039 Additional body json for GenericJsonAndUrlQueryCreator
davidradl Feb 6, 2026
4922933
Additional validation are per review conversation
davidradl Feb 9, 2026
3b832bd
Prevent double json parse
davidradl Feb 10, 2026
6c65f35
Optimize to json object
davidradl Feb 10, 2026
414b874
improve method name
davidradl Feb 10, 2026
f072536
small fixups from review
davidradl Feb 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion docs/content.zh/docs/connectors/table/http.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options,

| Option | Required | Description/Value |
|:-----------------------------------------------------------------------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| connector | required | The Value should be set to _http_ |
| 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 use 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. |
Expand Down Expand Up @@ -205,6 +205,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options,
| http.source.lookup.proxy.password | optional | Specify the password used for proxy authentication. |
| http.request.query-param-fields | optional | Used for the `GenericJsonAndUrlQueryCreator` 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.body-fields | optional | Used for the `GenericJsonAndUrlQueryCreator` query creator. The names of the fields that will be mapped to the body. The parameters are separated by semicolons, such as `param1;param2`. | |
| http.request.additional-body-json | optional | Used for the `GenericJsonAndUrlQueryCreator` query creator. Additional JSON content to be merged into the request body for PUT and POST operations. The value should be a valid JSON object string (e.g., `'{"opportunity":{"source":"flink"},"priority":1}'`) that will be parsed and its fields merged at the top level with the generated request body. For example, if the body is `{"id":123}` and additional-body-json is `'{"extra":"value"}'`, the result will be `{"id":123,"extra":"value"}`. Supports nested objects and arrays. |
| http.request.url-map | optional | Used for the `GenericJsonAndUrlQueryCreator` 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:cid1,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. The expected format of the map is: `key1:value1,key2:value2`. |

### Query Creators
Expand Down
3 changes: 2 additions & 1 deletion docs/content/docs/connectors/table/http.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options,

| Option | Required | Description/Value |
|:-----------------------------------------------------------------------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| connector | required | The Value should be set to _http_ |
| 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 use 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. |
Expand Down Expand Up @@ -205,6 +205,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options,
| http.source.lookup.proxy.password | optional | Specify the password used for proxy authentication. |
| http.request.query-param-fields | optional | Used for the `GenericJsonAndUrlQueryCreator` 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.body-fields | optional | Used for the `GenericJsonAndUrlQueryCreator` query creator. The names of the fields that will be mapped to the body. The parameters are separated by semicolons, such as `param1;param2`. | |
| http.request.additional-body-json | optional | Used for the `GenericJsonAndUrlQueryCreator` query creator. Additional JSON content to be merged into the request body for PUT and POST operations. The value should be a valid JSON object string (e.g., `'{"opportunity":{"source":"flink"},"priority":1}'`) that will be parsed and its fields merged at the top level with the generated request body. For example, if the body is `{"id":123}` and additional-body-json is `'{"extra":"value"}'`, the result will be `{"id":123,"extra":"value"}`. Supports nested objects and arrays. |
| http.request.url-map | optional | Used for the `GenericJsonAndUrlQueryCreator` 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:cid1,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. The expected format of the map is: `key1:value1,key2:value2`. |

### Query Creators
Expand Down
21 changes: 18 additions & 3 deletions ...apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
private final List<String> requestQueryParamsFields;
private final List<String> requestBodyFields;
private final Map<String, String> requestUrlMap;
private final ObjectNode additionalRequestObject;

/**
* Construct a Generic JSON and URL query creator.
Expand All @@ -88,6 +89,8 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
* @param requestQueryParamsFields query param fields
* @param requestBodyFields body fields used for PUT and POSTs
* @param requestUrlMap url map
* @param additionalRequestObject pre-parsed additional JSON object to merge into request body
* (parsed once in factory to avoid re-parsing on every lookup)
* @param lookupRow lookup row itself.
*/
public GenericJsonAndUrlQueryCreator(
Expand All @@ -96,13 +99,15 @@ public GenericJsonAndUrlQueryCreator(
final List<String> requestQueryParamsFields,
final List<String> requestBodyFields,
final Map<String, String> requestUrlMap,
final ObjectNode additionalRequestObject,
final LookupRow lookupRow) {
this.httpMethod = httpMethod;
this.serializationSchema = serializationSchema;
this.lookupRow = lookupRow;
this.requestQueryParamsFields = requestQueryParamsFields;
this.requestBodyFields = requestBodyFields;
this.requestUrlMap = requestUrlMap;
this.additionalRequestObject = additionalRequestObject;
}

@VisibleForTesting
Expand Down Expand Up @@ -145,9 +150,19 @@ public LookupQueryInfo createLookupQuery(final RowData lookupDataRow) {
// Body-based queries
// serialize to a string for the body.
try {
lookupQuery =
ObjectMapperAdapter.instance()
.writeValueAsString(jsonObject.retain(requestBodyFields));
ObjectNode bodyJsonObject = jsonObject.retain(requestBodyFields);

// Merge additional JSON if provided (already validated as object in factory)
if (additionalRequestObject != null) {
// Merge all fields from additional JSON into the body
// This preserves nested objects and arrays as-is
additionalRequestObject
.fields()
.forEachRemaining(
entry -> bodyJsonObject.set(entry.getKey(), entry.getValue()));
}

lookupQuery = ObjectMapperAdapter.instance().writeValueAsString(bodyJsonObject);
} catch (JsonProcessingException e) {
final String message = "Unable to convert Json Object to a string";
throw new RuntimeException(message, e);
Expand Down
Loading
Loading