Skip to content

[FLINK-40279][connector-http] Add HTTP sink OIDC, proxy and HTTP version options #58

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

Open
bharathgunapati wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[FLINK-40279][connector-http] Add HTTP sink OIDC, proxy and HTTP version options #58

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
16 changes: 14 additions & 2 deletions docs/content.zh/docs/connectors/datastream/http.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,14 @@ These options are specified on the builder using the setProperty method.
| 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. Required when `http.security.oidc.token.endpoint.url` is configured. |
| http.security.oidc.token.endpoint.url | optional | OIDC `Token Endpoint` URL to which the token request will be issued. When set, every sink request gets an `Authorization: Bearer <token>` header with a cached token that is refreshed when it expires; this replaces any configured `http.sink.header.Authorization`. |
| 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. The default is `30s`. |
| http.sink.http-version | optional | Version of HTTP to use for sink HTTP requests. Valid values are `HTTP_1_1` and `HTTP_2`. The default is `HTTP_1_1`. |
| http.sink.proxy.host | optional | Hostname of the proxy used by the sink HTTP client. |
| http.sink.proxy.port | optional | Port of the proxy used by the sink HTTP client. |
| http.sink.proxy.username | optional | Username used for sink proxy authentication. |
| http.sink.proxy.password | optional | Password used for sink proxy authentication. |
| http.sink.request.timeout | optional | Sets HTTP request timeout for the HTTP sink as a Duration (e.g. `30s`, `1min`). If not specified, the default value of `30s` will be used. |
| http.sink.max-retries | optional | The maximum number of retries for failed HTTP sink requests. Defaults to `0` (retries disabled) so existing jobs keep send-once behaviour. Set a positive value to enable retries. |
| http.sink.retry-strategy.type | optional | Retry strategy for failed HTTP sink requests. Valid values are `fixed-delay` (the default) and `exponential-delay`. Applies only when `http.sink.max-retries` is greater than 0. |
Expand Down Expand Up @@ -187,10 +195,14 @@ in configuration property `http.security.oidc.token.request`. See [grant extensi
an example of a customised grant type token request. The supplied `token request` will be issued to the
[token end point](https://datatracker.ietf.org/doc/html/rfc6749#section-3.2), whose url should be supplied in configuration property
`http.security.oidc.token.endpoint.url`. The returned `access token` is then cached and used for subsequent requests; if the token has expired then
a new one is requested. There is a property `http.security.oidc.token.expiry.reduction`, that defaults to 1 second; new tokens will
be requested if the current time is later than the cached token expiry time minus `http.security.oidc.token.expiry.reduction`.
a new one is requested. There is a property `http.security.oidc.token.expiry.reduction`, that defaults to 30 seconds; new tokens will
be requested if the current time is later than the cached token expiry time minus `http.security.oidc.token.expiry.reduction`. If a token's `expires_in` is not greater than this value, the full token lifetime is used.

Each `HttpSink` is configured and authenticated independently, so sinks in the same job can write to different services with different token endpoints and credentials. The bearer token is added to every HTTP request, including batch requests and retries, and it replaces any configured `Authorization` header. The token is fetched when a request is sent, not when the sink is created.

### Restrictions at this time
* No authentication is applied to the token request.
* The token request is sent with a plain HTTP client. It does not use the sink's proxy or TLS trust options.
* The token request times out after `http.sink.request.timeout`. A timeout fails the sink request that needed the token.
* The processing does not use the refresh token if it is present.
{{< top >}}
20 changes: 16 additions & 4 deletions 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 @@ -265,7 +265,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options,
| 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.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. The default is `30s`. |
| 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. |
Expand Down Expand Up @@ -502,7 +502,7 @@ CREATE TABLE http-lookup (
)
```

Note that when using OIDC, it adds an `Authentication` header with the bearer token; this will override
Note that when using OIDC, it adds an `Authorization` header with the bearer token; this will override
an existing `Authorization` header specified in configuration.

### Timeouts
Expand Down Expand Up @@ -603,6 +603,14 @@ another format name.
| 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. Required when `http.security.oidc.token.endpoint.url` is configured. |
| http.security.oidc.token.endpoint.url | optional | OIDC `Token Endpoint` URL to which the token request will be issued. When set, every sink request gets an `Authorization: Bearer <token>` header with a cached token that is refreshed when it expires; this replaces any configured `http.sink.header.Authorization`. |
| 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. The default is `30s`. |
| http.sink.http-version | optional | Version of HTTP to use for sink HTTP requests. Valid values are `HTTP_1_1` and `HTTP_2`. The default is `HTTP_1_1`. |
| http.sink.proxy.host | optional | Hostname of the proxy used by the sink HTTP client. |
| http.sink.proxy.port | optional | Port of the proxy used by the sink HTTP client. |
| http.sink.proxy.username | optional | Username used for sink proxy authentication. |
| http.sink.proxy.password | optional | Password used for sink proxy authentication. |
| http.sink.request.timeout | optional | Sets HTTP request timeout for the HTTP sink as a Duration (e.g. `'30s'`, `'1min'`). If not specified, the default value of `30s` will be used. |
| http.sink.max-retries | optional | The maximum number of retries for failed HTTP sink requests. Defaults to `0` (retries disabled) so existing jobs keep send-once behaviour. Set a positive value to enable retries. |
| http.sink.retry-strategy.type | optional | Retry strategy for failed HTTP sink requests. Valid values are `fixed-delay` (the default) and `exponential-delay`. Applies only when `http.sink.max-retries` is greater than 0. |
Expand Down Expand Up @@ -831,8 +839,10 @@ in configuration property `http.security.oidc.token.request`. See [grant extensi
an example of a customised grant type token request. The supplied `token request` will be issued to the
[token end point](https://datatracker.ietf.org/doc/html/rfc6749#section-3.2), whose url should be supplied in configuration property
`http.security.oidc.token.endpoint.url`. The returned `access token` is then cached and used for subsequent requests; if the token has expired then
a new one is requested. There is a property `http.security.oidc.token.expiry.reduction`, that defaults to 1 second; new tokens will
be requested if the current time is later than the cached token expiry time minus `http.security.oidc.token.expiry.reduction`.
a new one is requested. There is a property `http.security.oidc.token.expiry.reduction`, that defaults to 30 seconds; new tokens will
be requested if the current time is later than the cached token expiry time minus `http.security.oidc.token.expiry.reduction`. If a token's `expires_in` is not greater than this value, the full token lifetime is used.

Lookup and sink tables use the same option names, but each table is configured and authenticated independently, so a job can read from one service and write to another with different token endpoints and credentials. The bearer token is added to every HTTP request, including sink batch requests and retries, and it replaces any configured `Authorization` header. The token is fetched when a request is sent, not at planning time, so `EXPLAIN` does not call the token endpoint.

## Logging the HTTP content
Debug level logging has been added for class `org.apache.flink.connector.http.HttpLogger`. To enable this, alter the log4j properties.
Expand All @@ -856,5 +866,7 @@ Notes:

#### Restrictions at this time
* No authentication is applied to the token request.
* The token request is sent with a plain HTTP client. It does not use the connector's proxy or TLS trust options.
* The token request times out after the connector's request timeout (`http.source.lookup.request.timeout` for lookups, `http.sink.request.timeout` for the sink). A timeout fails the request that needed the token.
* The processing does not use the refresh token if it is present.
{{< top >}}
Loading