diff --git a/docs/content.zh/docs/connectors/datastream/http.md b/docs/content.zh/docs/connectors/datastream/http.md index ce9e6f5c..caff3446 100644 --- a/docs/content.zh/docs/connectors/datastream/http.md +++ b/docs/content.zh/docs/connectors/datastream/http.md @@ -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 ` 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. | @@ -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 >}} diff --git a/docs/content.zh/docs/connectors/table/http.md b/docs/content.zh/docs/connectors/table/http.md index 6c25f63d..c54f8f65 100644 --- a/docs/content.zh/docs/connectors/table/http.md +++ b/docs/content.zh/docs/connectors/table/http.md @@ -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. | @@ -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 @@ -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 ` 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. | @@ -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. @@ -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 >}} diff --git a/docs/content/docs/connectors/datastream/http.md b/docs/content/docs/connectors/datastream/http.md index ce9e6f5c..caff3446 100644 --- a/docs/content/docs/connectors/datastream/http.md +++ b/docs/content/docs/connectors/datastream/http.md @@ -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 ` 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. | @@ -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 >}} diff --git a/docs/content/docs/connectors/table/http.md b/docs/content/docs/connectors/table/http.md index 85822ea0..a9bf3e3d 100644 --- a/docs/content/docs/connectors/table/http.md +++ b/docs/content/docs/connectors/table/http.md @@ -266,7 +266,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. | @@ -503,7 +503,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 @@ -604,6 +604,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 ` 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. | @@ -832,8 +840,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. @@ -857,5 +867,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 >}} diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/auth/OidcAccessTokenManager.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/auth/OidcAccessTokenManager.java index 71458034..6c3c9c49 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/auth/OidcAccessTokenManager.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/auth/OidcAccessTokenManager.java @@ -28,6 +28,7 @@ import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; import java.time.Duration; import java.time.Instant; @@ -43,12 +44,21 @@ @Slf4j public class OidcAccessTokenManager { - private static final Duration DEFAULT_TOKEN_EXPIRATION_REDUCTION = Duration.ofSeconds(1); + /** Leaves headroom for network latency, request duration and clock skew. */ + public static final Duration DEFAULT_TOKEN_EXPIRATION_REDUCTION = Duration.ofSeconds(30); + + /** + * Upper bound for one call to the token endpoint. The token is fetched synchronously on the + * thread that builds the HTTP request, so an unanswered token request must not block forever. + */ + public static final Duration DEFAULT_TOKEN_REQUEST_TIMEOUT = Duration.ofSeconds(30); + private final HttpClient httpClient; private final String tokenRequest; private final String url; private final Duration tokenExpirationReduction; + private final Duration tokenRequestTimeout; private String cachedAccessToken; private Instant cachedAccessTokenExp; @@ -79,6 +89,32 @@ public OidcAccessTokenManager( String tokenRequest, String url, Duration tokenExpirationReduction) { + this( + httpClient, + tokenRequest, + url, + tokenExpirationReduction, + DEFAULT_TOKEN_REQUEST_TIMEOUT); + } + + /** + * Construct an Oidc access token manager with the supplied token expiration reduction and token + * request timeout. + * + * @param httpClient httpClient to use to call the token endpoint. + * @param tokenRequest token request this need to be form urlencoded + * @param url token endpoint url + * @param tokenExpirationReduction token expiry reduction, request a new token if the current + * time is later than the cached access token expiry time reduced by this value. This means + * that we will not use the cached token if it is about to expire. + * @param tokenRequestTimeout how long one call to the token endpoint may take before it fails + */ + public OidcAccessTokenManager( + HttpClient httpClient, + String tokenRequest, + String url, + Duration tokenExpirationReduction, + Duration tokenRequestTimeout) { this.tokenRequest = tokenRequest; this.httpClient = httpClient; this.url = url; @@ -87,6 +123,11 @@ public OidcAccessTokenManager( } else { this.tokenExpirationReduction = tokenExpirationReduction; } + if (null == tokenRequestTimeout) { + this.tokenRequestTimeout = DEFAULT_TOKEN_REQUEST_TIMEOUT; + } else { + this.tokenRequestTimeout = tokenRequestTimeout; + } } /** Request an access token from the token endpoint. */ @@ -95,6 +136,7 @@ private void requestAccessToken() { HttpRequest httpRequest = HttpRequest.newBuilder() .uri(URI.create(url)) + .timeout(tokenRequestTimeout) .header("Content-Type", "application/x-www-form-urlencoded") .method("POST", HttpRequest.BodyPublishers.ofString(tokenRequest)) .build(); @@ -126,8 +168,16 @@ private void requestAccessToken() { } } catch (JsonProcessingException e) { throw new IllegalStateException("Error found while trying to request a new token"); + } catch (HttpTimeoutException e) { + throw new IllegalStateException( + "Timed out after " + + tokenRequestTimeout + + " waiting for an OIDC access token from " + + url, + e); } catch (IOException e) { - throw new IllegalStateException("IO Exception occurred", e); + throw new IllegalStateException( + "IO Exception occurred while requesting an OIDC access token from " + url, e); } catch (InterruptedException e) { throw new IllegalStateException("Interrupted Exception occurred", e); } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java index 1619cdc8..8d6ef92b 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java @@ -132,6 +132,16 @@ public final class HttpConnectorConfigConstants { public static final String SINK_HTTP_TIMEOUT_SECONDS = FLINK_CONNECTOR_HTTP_SINK + "request.timeout"; + public static final String SINK_HTTP_VERSION = FLINK_CONNECTOR_HTTP_SINK + "http-version"; + + public static final String SINK_PROXY_HOST = FLINK_CONNECTOR_HTTP_SINK + "proxy.host"; + + public static final String SINK_PROXY_PORT = FLINK_CONNECTOR_HTTP_SINK + "proxy.port"; + + public static final String SINK_PROXY_USERNAME = FLINK_CONNECTOR_HTTP_SINK + "proxy.username"; + + public static final String SINK_PROXY_PASSWORD = FLINK_CONNECTOR_HTTP_SINK + "proxy.password"; + public static final String LOOKUP_HTTP_POLLING_THREAD_POOL_SIZE = SOURCE_LOOKUP_PREFIX + "request.thread-pool.size"; diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfig.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfig.java index d1828f7a..b54427db 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfig.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfig.java @@ -37,6 +37,7 @@ import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_IGNORED_RESPONSE_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_RETRY_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_SUCCESS_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_MAX_RETRIES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF; @@ -95,6 +96,10 @@ public Duration getRequestTimeout() { return readableConfig.get(SINK_REQUEST_TIMEOUT); } + public String getHttpVersion() { + return readableConfig.get(SINK_HTTP_VERSION); + } + public int getWriterThreadPoolSize() { return readableConfig.get(SINK_WRITER_THREAD_POOL_SIZE); } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfigFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfigFactory.java index c5e3236f..58c74401 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfigFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpSinkConfigFactory.java @@ -37,7 +37,15 @@ import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_IGNORED_RESPONSE_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_RETRY_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_SUCCESS_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_MAX_RETRIES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_HOST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PASSWORD; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PORT; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_USERNAME; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF; @@ -84,6 +92,54 @@ public static HttpSinkConfig fromDataStream( configuration.set(SINK_REQUEST_TIMEOUT, TimeUtils.parseDuration(requestTimeout)); } + String httpVersion = properties.getProperty(HttpConnectorConfigConstants.SINK_HTTP_VERSION); + if (httpVersion != null) { + configuration.set(SINK_HTTP_VERSION, httpVersion); + } + + String proxyHost = properties.getProperty(HttpConnectorConfigConstants.SINK_PROXY_HOST); + if (proxyHost != null) { + configuration.set(SINK_PROXY_HOST, proxyHost); + } + + String proxyPort = properties.getProperty(HttpConnectorConfigConstants.SINK_PROXY_PORT); + if (proxyPort != null) { + configuration.set(SINK_PROXY_PORT, Integer.parseInt(proxyPort)); + } + + String proxyUsername = + properties.getProperty(HttpConnectorConfigConstants.SINK_PROXY_USERNAME); + if (proxyUsername != null) { + configuration.set(SINK_PROXY_USERNAME, proxyUsername); + } + + String proxyPassword = + properties.getProperty(HttpConnectorConfigConstants.SINK_PROXY_PASSWORD); + if (proxyPassword != null) { + configuration.set(SINK_PROXY_PASSWORD, proxyPassword); + } + + String oidcTokenEndpointUrl = + properties.getProperty(HttpConnectorConfigConstants.OIDC_AUTH_TOKEN_ENDPOINT_URL); + if (oidcTokenEndpointUrl != null) { + configuration.set(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL, oidcTokenEndpointUrl); + } + + String oidcTokenRequest = + properties.getProperty(HttpConnectorConfigConstants.OIDC_AUTH_TOKEN_REQUEST); + if (oidcTokenRequest != null) { + configuration.set(SINK_OIDC_AUTH_TOKEN_REQUEST, oidcTokenRequest); + } + + String oidcTokenExpiryReduction = + properties.getProperty( + HttpConnectorConfigConstants.OIDC_AUTH_TOKEN_EXPIRY_REDUCTION); + if (oidcTokenExpiryReduction != null) { + configuration.set( + SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION, + TimeUtils.parseDuration(oidcTokenExpiryReduction)); + } + String writerThreadPoolSize = properties.getProperty( HttpConnectorConfigConstants.SINK_HTTP_WRITER_THREAD_POOL_SIZE); diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java index 4771d012..2bdf0bc9 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java @@ -31,7 +31,16 @@ public class OIDCAuthHeaderValuePreprocessor implements HeaderValuePreprocessor private final String oidcAuthURL; private final String oidcTokenRequest; - private Duration oidcExpiryReduction = Duration.ofSeconds(1); + private Duration oidcExpiryReduction = + OidcAccessTokenManager.DEFAULT_TOKEN_EXPIRATION_REDUCTION; + private final Duration tokenRequestTimeout; + + /** + * Created on first use and kept for the life of this preprocessor instance, so every request + * served by the same instance reuses the cached token until it expires. The field is transient: + * after deserialization it is created again and a new token is fetched on the next request. + */ + private transient OidcAccessTokenManager accessTokenManager; /** * Add the access token to the request using OidcAuth authenticate method that gives us a valid @@ -43,22 +52,51 @@ public class OIDCAuthHeaderValuePreprocessor implements HeaderValuePreprocessor */ public OIDCAuthHeaderValuePreprocessor( String oidcAuthURL, String oidcTokenRequest, Optional oidcExpiryReduction) { + this( + oidcAuthURL, + oidcTokenRequest, + oidcExpiryReduction, + OidcAccessTokenManager.DEFAULT_TOKEN_REQUEST_TIMEOUT); + } + + /** + * Add the access token to the request using OidcAuth authenticate method that gives us a valid + * access token. + * + * @param oidcAuthURL OIDC token endpoint + * @param oidcTokenRequest OIDC Token Request + * @param oidcExpiryReduction OIDC token expiry reduction + * @param tokenRequestTimeout how long one call to the token endpoint may take before it fails + */ + public OIDCAuthHeaderValuePreprocessor( + String oidcAuthURL, + String oidcTokenRequest, + Optional oidcExpiryReduction, + Duration tokenRequestTimeout) { this.oidcAuthURL = oidcAuthURL; this.oidcTokenRequest = oidcTokenRequest; if (oidcExpiryReduction.isPresent()) { this.oidcExpiryReduction = oidcExpiryReduction.get(); } + this.tokenRequestTimeout = tokenRequestTimeout; } + /** + * Returns the bearer header value, reusing the cached access token until it expires. + * Synchronized because the token manager is not thread-safe and lookups can run concurrently. + */ @Override - public String preprocessHeaderValue(String rawValue) { - OidcAccessTokenManager auth = - new OidcAccessTokenManager( - HttpClient.newBuilder().build(), - oidcTokenRequest, - oidcAuthURL, - oidcExpiryReduction); + public synchronized String preprocessHeaderValue(String rawValue) { + if (accessTokenManager == null) { + accessTokenManager = + new OidcAccessTokenManager( + HttpClient.newBuilder().connectTimeout(tokenRequestTimeout).build(), + oidcTokenRequest, + oidcAuthURL, + oidcExpiryReduction, + tokenRequestTimeout); + } // apply the OIDC authentication by adding the dynamically calculated header value. - return "Bearer " + auth.authenticate(); + return "Bearer " + accessTokenManager.authenticate(); } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java index e15582e2..0b23c441 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java @@ -17,12 +17,17 @@ package org.apache.flink.connector.http.sink.httpclient; +import org.apache.flink.connector.http.auth.OidcAccessTokenManager; import org.apache.flink.connector.http.config.HttpSinkConfig; +import org.apache.flink.connector.http.utils.HttpHeaderUtils; import org.apache.flink.connector.http.utils.ThreadUtils; import org.apache.flink.util.concurrent.ExecutorThreadFactory; import java.net.http.HttpClient; +import java.net.http.HttpRequest.Builder; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -43,6 +48,11 @@ public abstract class AbstractRequestSubmitter implements RequestSubmitter { protected final HttpClient httpClient; + protected final HttpClient.Version httpVersion; + + /** Supplies the OIDC bearer token for each request; {@code null} when OIDC is not set. */ + private final OidcAccessTokenManager oidcAccessTokenManager; + public AbstractRequestSubmitter( HttpSinkConfig sinkConfig, String[] headersAndValues, @@ -59,6 +69,9 @@ public AbstractRequestSubmitter( ThreadUtils.LOGGING_EXCEPTION_HANDLER)); this.httpRequestTimeout = sinkConfig.getRequestTimeout(); + this.httpVersion = HttpClient.Version.valueOf(sinkConfig.getHttpVersion()); + this.oidcAccessTokenManager = + HttpHeaderUtils.createSinkOidcAccessTokenManager(sinkConfig.getReadableConfig()); this.httpClient = httpClient; } @@ -68,4 +81,39 @@ public void close() { publishingThreadPool.shutdownNow(); httpClientExecutor.shutdownNow(); } + + protected Builder newRequestBuilder() { + return java.net.http.HttpRequest.newBuilder() + .version(httpVersion) + .timeout(httpRequestTimeout); + } + + /** + * Returns the headers for the next HTTP request. With OIDC configured, a current bearer token + * replaces any configured {@code Authorization} header, so an expired token is refreshed before + * the request is sent. + */ + protected String[] requestHeaders() { + if (oidcAccessTokenManager == null) { + return headersAndValues; + } + + String accessToken; + // The token manager caches the token without synchronization, and requests are built + // from both the writer thread and the retry scheduler. + synchronized (oidcAccessTokenManager) { + accessToken = oidcAccessTokenManager.authenticate(); + } + + List headers = new ArrayList<>(headersAndValues.length + 2); + for (int i = 0; i + 1 < headersAndValues.length; i += 2) { + if (!HttpHeaderUtils.AUTHORIZATION.equalsIgnoreCase(headersAndValues[i])) { + headers.add(headersAndValues[i]); + headers.add(headersAndValues[i + 1]); + } + } + headers.add(HttpHeaderUtils.AUTHORIZATION); + headers.add("Bearer " + accessToken); + return headers.toArray(new String[0]); + } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java index afb31d5e..25534d98 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java @@ -26,7 +26,6 @@ import java.net.URI; import java.net.http.HttpClient; -import java.net.http.HttpClient.Version; import java.net.http.HttpRequest.BodyPublisher; import java.net.http.HttpRequest.BodyPublishers; import java.net.http.HttpRequest.Builder; @@ -139,15 +138,11 @@ private HttpRequest buildHttpRequest(List requestBatch, UR elements.set(elements.size() - 1, BATCH_END_BYTES); publisher = BodyPublishers.ofByteArrays(elements); - Builder requestBuilder = - java.net.http.HttpRequest.newBuilder() - .uri(endpointUri) - .version(Version.HTTP_1_1) - .timeout(httpRequestTimeout) - .method(method, publisher); + Builder requestBuilder = newRequestBuilder().uri(endpointUri).method(method, publisher); - if (headersAndValues.length != 0) { - requestBuilder.headers(headersAndValues); + String[] headers = requestHeaders(); + if (headers.length != 0) { + requestBuilder.headers(headers); } return new HttpRequest( diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterFactory.java index 753c5073..bdcf51f9 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterFactory.java @@ -87,7 +87,7 @@ public BatchRequestSubmitter createSubmitter( return new BatchRequestSubmitter( sinkConfig, headersAndValues, - JavaNetHttpClientFactory.createClient(properties, httpClientExecutor), + JavaNetHttpClientFactory.createClient(sinkConfig, httpClientExecutor), httpClientExecutor); } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestRequestSubmitterFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestRequestSubmitterFactory.java index cccf4dcc..26ad93b4 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestRequestSubmitterFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestRequestSubmitterFactory.java @@ -45,8 +45,7 @@ public RequestSubmitter createSubmitter(HttpSinkConfig sinkConfig, String[] head return new PerRequestSubmitter( sinkConfig, headersAndValues, - JavaNetHttpClientFactory.createClient( - sinkConfig.getProperties(), httpClientExecutor), + JavaNetHttpClientFactory.createClient(sinkConfig, httpClientExecutor), httpClientExecutor); } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java index fbc7a41e..da2f0ea9 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java @@ -24,7 +24,6 @@ import java.net.URI; import java.net.http.HttpClient; -import java.net.http.HttpClient.Version; import java.net.http.HttpRequest.BodyPublishers; import java.net.http.HttpRequest.Builder; import java.net.http.HttpResponse; @@ -79,16 +78,15 @@ public List> submit( private HttpRequest buildHttpRequest(HttpSinkRequestEntry requestEntry, URI endpointUri) { Builder requestBuilder = - java.net.http.HttpRequest.newBuilder() + newRequestBuilder() .uri(endpointUri) - .version(Version.HTTP_1_1) - .timeout(httpRequestTimeout) .method( requestEntry.method, BodyPublishers.ofByteArray(requestEntry.element)); - if (headersAndValues.length != 0) { - requestBuilder.headers(headersAndValues); + String[] headers = requestHeaders(); + if (headers.length != 0) { + requestBuilder.headers(headers); } return new HttpRequest( diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java index e4e1f4d1..6861d77c 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConnectorOptions.java @@ -20,6 +20,7 @@ import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.connector.http.HttpLoggingLevelType; +import org.apache.flink.connector.http.auth.OidcAccessTokenManager; import org.apache.flink.connector.http.retry.RetryStrategyType; import java.time.Duration; @@ -127,7 +128,7 @@ public class HttpLookupConnectorOptions { public static final ConfigOption SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION = ConfigOptions.key(OIDC_AUTH_TOKEN_EXPIRY_REDUCTION) .durationType() - .defaultValue(Duration.ofSeconds(1)) + .defaultValue(OidcAccessTokenManager.DEFAULT_TOKEN_EXPIRATION_REDUCTION) .withDescription( "OIDC authorization access token expiry" + " reduction as a Duration." diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java index 01e8883b..8252599f 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java @@ -85,6 +85,9 @@ public class JavaNetHttpPollingClient implements PollingClient { private final boolean continueOnError; private final HttpLogger httpLogger; + /** Shared across lookups so the OIDC token is cached; {@code null} when OIDC is not set. */ + private final HeaderPreprocessor oidcHeaderPreprocessor; + public JavaNetHttpPollingClient( HttpClient httpClient, DeserializationSchema responseBodyDecoder, @@ -118,6 +121,7 @@ public JavaNetHttpPollingClient( .responseChecker(new HttpResponseChecker(successCodes, errorCodes)) .build(); this.httpLogger = HttpLogger.getHttpLogger(options.getProperties()); + this.oidcHeaderPreprocessor = HttpHeaderUtils.createOIDCHeaderPreprocessor(config); } public void open(FunctionContext context) { @@ -150,17 +154,13 @@ public HttpRowDataWrapper pull(RowData lookupRow) { private HttpRowDataWrapper queryAndProcess(RowData lookupData) throws Exception { var request = requestFactory.buildLookupRequest(lookupData); - var oidcProcessor = - (options.getReadableConfig() != null) - ? HttpHeaderUtils.createOIDCHeaderPreprocessor(options.getReadableConfig()) - : null; HttpResponse response = null; HttpRowDataWrapper httpRowDataWrapper = null; try { httpLogger.logRequest(request.getHttpRequest()); response = httpClient.send( - () -> updateHttpRequestIfRequired(request, oidcProcessor), + () -> updateHttpRequestIfRequired(request, oidcHeaderPreprocessor), BodyHandlers.ofString()); httpLogger.logResponse(response); } catch (HttpStatusCodeValidationFailedException e) { diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java index 3f9d4c8f..fda7036b 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java @@ -19,6 +19,7 @@ import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.connector.http.auth.OidcAccessTokenManager; import org.apache.flink.connector.http.config.HttpConnectorConfigConstants; import org.apache.flink.connector.http.retry.RetryStrategyType; @@ -64,6 +65,14 @@ public class HttpDynamicSinkConnectorOptions { + "before timing out a single request. " + "Specified as a Duration, e.g. '30s' or '1min'."); + public static final ConfigOption SINK_HTTP_VERSION = + ConfigOptions.key(HttpConnectorConfigConstants.SINK_HTTP_VERSION) + .stringType() + .defaultValue(String.valueOf(java.net.http.HttpClient.Version.HTTP_1_1)) + .withDescription( + "Version of HTTP to use for sink HTTP requests. " + + "The valid values are HTTP_1_1 and HTTP_2."); + public static final ConfigOption REQUEST_CALLBACK_IDENTIFIER = ConfigOptions.key(SINK_REQUEST_CALLBACK_IDENTIFIER) .stringType() @@ -117,6 +126,49 @@ public class HttpDynamicSinkConnectorOptions { + "without retrying. Cannot be set together with legacy " + "http.sink.error.code.exclude."); + public static final ConfigOption SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL = + ConfigOptions.key(HttpConnectorConfigConstants.OIDC_AUTH_TOKEN_ENDPOINT_URL) + .stringType() + .noDefaultValue() + .withDescription("OIDC Token endpoint url."); + + public static final ConfigOption SINK_OIDC_AUTH_TOKEN_REQUEST = + ConfigOptions.key(HttpConnectorConfigConstants.OIDC_AUTH_TOKEN_REQUEST) + .stringType() + .noDefaultValue() + .withDescription("OIDC token request."); + + public static final ConfigOption SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION = + ConfigOptions.key(HttpConnectorConfigConstants.OIDC_AUTH_TOKEN_EXPIRY_REDUCTION) + .durationType() + .defaultValue(OidcAccessTokenManager.DEFAULT_TOKEN_EXPIRATION_REDUCTION) + .withDescription( + "OIDC authorization access token expiry reduction as a Duration."); + + public static final ConfigOption SINK_PROXY_HOST = + ConfigOptions.key(HttpConnectorConfigConstants.SINK_PROXY_HOST) + .stringType() + .noDefaultValue() + .withDescription("Http client proxy host."); + + public static final ConfigOption SINK_PROXY_PORT = + ConfigOptions.key(HttpConnectorConfigConstants.SINK_PROXY_PORT) + .intType() + .noDefaultValue() + .withDescription("Http client proxy port."); + + public static final ConfigOption SINK_PROXY_USERNAME = + ConfigOptions.key(HttpConnectorConfigConstants.SINK_PROXY_USERNAME) + .stringType() + .noDefaultValue() + .withDescription("Http client proxy username for authentication."); + + public static final ConfigOption SINK_PROXY_PASSWORD = + ConfigOptions.key(HttpConnectorConfigConstants.SINK_PROXY_PASSWORD) + .stringType() + .noDefaultValue() + .withDescription("Http client proxy password for authentication."); + public static final ConfigOption SINK_RETRY_STRATEGY = ConfigOptions.key(SINK_RETRY_STRATEGY_TYPE) .stringType() diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java index 3f2aa1ca..12260dfa 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java @@ -44,7 +44,15 @@ import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_IGNORED_RESPONSE_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_RETRY_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_SUCCESS_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_MAX_RETRIES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_HOST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PASSWORD; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PORT; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_USERNAME; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF; @@ -118,12 +126,20 @@ public Set> optionalOptions() { var options = super.optionalOptions(); options.add(INSERT_METHOD); options.add(SINK_REQUEST_TIMEOUT); + options.add(SINK_HTTP_VERSION); options.add(SINK_WRITER_THREAD_POOL_SIZE); options.add(REQUEST_CALLBACK_IDENTIFIER); options.add(SINK_MAX_RETRIES); options.add(SINK_HTTP_SUCCESS_CODES); options.add(SINK_HTTP_RETRY_CODES); options.add(SINK_HTTP_IGNORED_RESPONSE_CODES); + options.add(SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION); + options.add(SINK_OIDC_AUTH_TOKEN_REQUEST); + options.add(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL); + options.add(SINK_PROXY_HOST); + options.add(SINK_PROXY_PORT); + options.add(SINK_PROXY_USERNAME); + options.add(SINK_PROXY_PASSWORD); options.add(SINK_RETRY_STRATEGY); options.add(SINK_RETRY_FIXED_DELAY_DELAY); options.add(SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF); @@ -145,6 +161,33 @@ private void validateHttpSinkOptions(ReadableConfig tableOptions) INSERT_METHOD.key())); } }); + tableOptions + .getOptional(SINK_HTTP_VERSION) + .ifPresent( + httpVersion -> { + try { + java.net.http.HttpClient.Version.valueOf(httpVersion); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + String.format( + "Invalid option '%s'. Supported values are: HTTP_1_1, HTTP_2.", + SINK_HTTP_VERSION.key()), + e); + } + }); + tableOptions + .getOptional(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL) + .ifPresent( + url -> { + if (tableOptions.getOptional(SINK_OIDC_AUTH_TOKEN_REQUEST).isEmpty()) { + throw new IllegalArgumentException( + "Config option " + + SINK_OIDC_AUTH_TOKEN_REQUEST.key() + + " is required, if " + + SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL.key() + + " is configured."); + } + }); tableOptions .getOptional(SINK_WRITER_THREAD_POOL_SIZE) .ifPresent( diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/HttpHeaderUtils.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/HttpHeaderUtils.java index 129633fe..b8bdb710 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/HttpHeaderUtils.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/HttpHeaderUtils.java @@ -18,6 +18,7 @@ package org.apache.flink.connector.http.utils; import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.connector.http.auth.OidcAccessTokenManager; import org.apache.flink.connector.http.preprocessor.BasicAuthHeaderValuePreprocessor; import org.apache.flink.connector.http.preprocessor.ComposeHeaderPreprocessor; import org.apache.flink.connector.http.preprocessor.HeaderPreprocessor; @@ -29,6 +30,7 @@ import lombok.experimental.UtilityClass; import lombok.extern.slf4j.Slf4j; +import java.net.http.HttpClient; import java.time.Duration; import java.util.Collections; import java.util.HashMap; @@ -41,6 +43,10 @@ import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; /** Http header utils. */ @UtilityClass @@ -112,11 +118,26 @@ public static HeaderPreprocessor createBasicAuthorizationHeaderPreprocessor( public static HeaderPreprocessor createOIDCAuthorizationHeaderPreprocessor( String oidcAuthURL, String oidcTokenRequest, Optional oidcExpiryReduction) { + return createOIDCAuthorizationHeaderPreprocessor( + oidcAuthURL, + oidcTokenRequest, + oidcExpiryReduction, + OidcAccessTokenManager.DEFAULT_TOKEN_REQUEST_TIMEOUT); + } + + public static HeaderPreprocessor createOIDCAuthorizationHeaderPreprocessor( + String oidcAuthURL, + String oidcTokenRequest, + Optional oidcExpiryReduction, + Duration tokenRequestTimeout) { return new ComposeHeaderPreprocessor( Collections.singletonMap( AUTHORIZATION, new OIDCAuthHeaderValuePreprocessor( - oidcAuthURL, oidcTokenRequest, oidcExpiryReduction))); + oidcAuthURL, + oidcTokenRequest, + oidcExpiryReduction, + tokenRequestTimeout))); } public static HeaderPreprocessor createHeaderPreprocessor(ReadableConfig readableConfig) { @@ -139,11 +160,54 @@ public static HeaderPreprocessor createOIDCHeaderPreprocessor(ReadableConfig rea Optional oidcExpiryReduction = readableConfig.getOptional(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION); + // The token request shares the lookup request timeout so a silent token endpoint + // cannot block a lookup for longer than the lookup itself may take. headerPreprocessor = HttpHeaderUtils.createOIDCAuthorizationHeaderPreprocessor( - oidcAuthURL.get(), oidcTokenRequest.get(), oidcExpiryReduction); + oidcAuthURL.get(), + oidcTokenRequest.get(), + oidcExpiryReduction, + readableConfig.get( + HttpLookupConnectorOptions.SOURCE_LOOKUP_REQUEST_TIMEOUT)); log.info("created OIDC HeaderPreprocessor"); } return headerPreprocessor; } + + /** + * Creates the OIDC access token manager used by the HTTP sink to authorize each request. + * + * @param readableConfig sink configuration + * @return access token manager, or {@code null} if no OIDC token endpoint is configured + */ + public static OidcAccessTokenManager createSinkOidcAccessTokenManager( + ReadableConfig readableConfig) { + Optional oidcAuthURL = + readableConfig.getOptional(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL); + if (oidcAuthURL.isEmpty()) { + return null; + } + + String oidcTokenRequest = + readableConfig + .getOptional(SINK_OIDC_AUTH_TOKEN_REQUEST) + .orElseThrow( + () -> + new IllegalArgumentException( + "Config option " + + SINK_OIDC_AUTH_TOKEN_REQUEST.key() + + " is required, if " + + SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL.key() + + " is configured.")); + // The token is fetched on the thread that builds the sink request, so the token request + // is bounded by the sink request timeout for both connecting and reading the response. + Duration tokenRequestTimeout = readableConfig.get(SINK_REQUEST_TIMEOUT); + log.info("created Sink OIDC access token manager"); + return new OidcAccessTokenManager( + HttpClient.newBuilder().connectTimeout(tokenRequestTimeout).build(), + oidcTokenRequest, + oidcAuthURL.get(), + readableConfig.get(SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION), + tokenRequestTimeout); + } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactory.java index 792cddff..ed0d83e8 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactory.java @@ -19,10 +19,12 @@ import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.connector.http.config.HttpConnectorConfigConstants; +import org.apache.flink.connector.http.config.HttpSinkConfig; import org.apache.flink.connector.http.security.SecurityContext; import org.apache.flink.connector.http.security.SelfSignedTrustManager; import org.apache.flink.connector.http.table.lookup.HttpLookupConfig; import org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions; +import org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions; import org.apache.flink.util.StringUtils; import lombok.AccessLevel; @@ -115,6 +117,47 @@ public static HttpClient createClient(Properties properties, Executor executor) .build(); } + /** + * Creates Java's {@link HttpClient} instance for HTTP sink requests using provided Executor for + * async calls. + * + * @param sinkConfig sink configuration + * @param executor {@link Executor} for async calls. + * @return new {@link HttpClient} instance. + */ + public static HttpClient createClient(HttpSinkConfig sinkConfig, Executor executor) { + + SSLContext sslContext = getSslContext(sinkConfig.getProperties()); + + var clientBuilder = + HttpClient.newBuilder() + .followRedirects(Redirect.NORMAL) + .sslContext(sslContext) + .executor(executor); + + ReadableConfig readableConfig = sinkConfig.getReadableConfig(); + Optional proxyHost = + readableConfig.getOptional(HttpDynamicSinkConnectorOptions.SINK_PROXY_HOST); + Optional proxyPort = + readableConfig.getOptional(HttpDynamicSinkConnectorOptions.SINK_PROXY_PORT); + + if (proxyHost.isPresent() && proxyPort.isPresent()) { + Optional proxyUsername = + readableConfig.getOptional(HttpDynamicSinkConnectorOptions.SINK_PROXY_USERNAME); + Optional proxyPassword = + readableConfig.getOptional(HttpDynamicSinkConnectorOptions.SINK_PROXY_PASSWORD); + + ProxyConfig proxyConfig = + new ProxyConfig(proxyHost.get(), proxyPort.get(), proxyUsername, proxyPassword); + clientBuilder.proxy( + ProxySelector.of( + new InetSocketAddress(proxyConfig.getHost(), proxyConfig.getPort()))); + proxyConfig.getAuthenticator().ifPresent(clientBuilder::authenticator); + } + + return clientBuilder.build(); + } + /** * Creates an {@link SSLContext} based on provided properties. * diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java index 56e2bbff..9c5ae994 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java @@ -18,20 +18,25 @@ package org.apache.flink.connector.http.preprocessor; import org.apache.flink.connector.http.WireMockServerPortAllocator; +import org.apache.flink.util.InstantiationUtil; import com.github.tomakehurst.wiremock.WireMockServer; import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import com.github.tomakehurst.wiremock.stubbing.Scenario; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.net.http.HttpTimeoutException; import java.time.Duration; import java.util.Optional; import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; import static com.github.tomakehurst.wiremock.client.WireMock.post; +import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor; import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class OIDCAuthHeaderValuePreprocessorTest { @@ -112,4 +117,91 @@ public void shouldReturnBearerTokenWithDefaultExpiryReduction() { assertThat(headerValue).startsWith("Bearer "); assertThat(headerValue).isEqualTo("Bearer " + accessToken); } + + @Test + public void shouldReuseCachedTokenWhileValid() { + stubToken(Scenario.STARTED, "token-1", 3600, "second"); + stubToken("second", "token-2", 3600, "second"); + + OIDCAuthHeaderValuePreprocessor preprocessor = preprocessor(); + + assertThat(preprocessor.preprocessHeaderValue("ignored")).isEqualTo("Bearer token-1"); + assertThat(preprocessor.preprocessHeaderValue("ignored")).isEqualTo("Bearer token-1"); + wireMockServer.verify(1, postRequestedFor(urlEqualTo(TOKEN_ENDPOINT))); + } + + @Test + public void shouldRefreshExpiredToken() { + stubToken(Scenario.STARTED, "token-1", 0, "second"); + stubToken("second", "token-2", 3600, "second"); + + OIDCAuthHeaderValuePreprocessor preprocessor = preprocessor(); + + assertThat(preprocessor.preprocessHeaderValue("ignored")).isEqualTo("Bearer token-1"); + assertThat(preprocessor.preprocessHeaderValue("ignored")).isEqualTo("Bearer token-2"); + wireMockServer.verify(2, postRequestedFor(urlEqualTo(TOKEN_ENDPOINT))); + } + + @Test + public void shouldWorkAfterSerialization() throws Exception { + stubToken(Scenario.STARTED, "token-1", 3600, Scenario.STARTED); + + OIDCAuthHeaderValuePreprocessor preprocessor = preprocessor(); + preprocessor.preprocessHeaderValue("ignored"); + OIDCAuthHeaderValuePreprocessor copy = + InstantiationUtil.clone(preprocessor, getClass().getClassLoader()); + + assertThat(copy.preprocessHeaderValue("ignored")).isEqualTo("Bearer token-1"); + } + + @Test + public void shouldFailWhenTokenEndpointDoesNotAnswerWithinTimeout() { + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_ENDPOINT)) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\": \"token-1\", \"expires_in\": 3600}") + .withFixedDelay(2_000))); + + OIDCAuthHeaderValuePreprocessor preprocessor = + new OIDCAuthHeaderValuePreprocessor( + "http://localhost:" + serverPort + TOKEN_ENDPOINT, + "grant_type=client_credentials", + Optional.of(Duration.ZERO), + Duration.ofMillis(200)); + + assertThatThrownBy(() -> preprocessor.preprocessHeaderValue("ignored")) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("OIDC access token") + .hasCauseInstanceOf(HttpTimeoutException.class); + } + + private OIDCAuthHeaderValuePreprocessor preprocessor() { + return new OIDCAuthHeaderValuePreprocessor( + "http://localhost:" + serverPort + TOKEN_ENDPOINT, + "grant_type=client_credentials", + Optional.of(Duration.ZERO)); + } + + private void stubToken( + String state, String accessToken, int expiresInSeconds, String nextState) { + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_ENDPOINT)) + .inScenario("token") + .whenScenarioStateIs(state) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\": \"" + + accessToken + + "\", \"expires_in\": " + + expiresInSeconds + + "}")) + .willSetStateTo(nextState)); + } } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java index 1bc7de36..92ef1e50 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java @@ -34,6 +34,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import java.net.http.HttpClient; +import java.net.http.HttpResponse; import java.time.Duration; import java.util.List; import java.util.Properties; @@ -44,9 +45,11 @@ import java.util.stream.IntStream; import java.util.stream.Stream; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -132,6 +135,44 @@ public void requestTimeoutDefaultsToThirtySeconds() { assertThat(submitter.httpRequestTimeout).isEqualTo(Duration.ofSeconds(30)); } + @Test + public void shouldUseConfiguredHttpVersion() { + Properties properties = new Properties(); + properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_BATCH_REQUEST_SIZE, "50"); + + Configuration configuration = new Configuration(); + configuration.set(SINK_HTTP_VERSION, "HTTP_2"); + + HttpSinkConfig sinkConfig = + HttpSinkConfig.builder() + .url("http://hello.pl") + .properties(properties) + .readableConfig(configuration) + .httpPostRequestCallback(new Slf4jHttpPostRequestCallback()) + .build(); + HttpResponse httpResponse = org.mockito.Mockito.mock(HttpResponse.class); + doReturn(CompletableFuture.completedFuture(httpResponse)) + .when(mockHttpClient) + .sendAsync(any(), any()); + + BatchRequestSubmitter submitter = + new BatchRequestSubmitter( + sinkConfig, + new String[0], + mockHttpClient, + Executors.newSingleThreadExecutor()); + JavaNetHttpResponseWrapper responseWrapper = + submitter + .submit( + "http://hello.pl", + List.of(new HttpSinkRequestEntry("PUT", new byte[] {1}))) + .get(0) + .join(); + + assertThat(responseWrapper.getHttpRequest().getHttpRequest().version()) + .contains(HttpClient.Version.HTTP_2); + } + private static Stream httpRequestMethods() { return Stream.of( Arguments.of(List.of("PUT", "PUT", "PUT", "PUT", "POST"), 2), diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitterTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitterTest.java new file mode 100644 index 00000000..b4863361 --- /dev/null +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitterTest.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.http.sink.httpclient; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.http.config.HttpSinkConfig; +import org.apache.flink.connector.http.sink.HttpSinkRequestEntry; +import org.apache.flink.connector.http.table.sink.Slf4jHttpPostRequestCallback; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.net.http.HttpClient; +import java.net.http.HttpResponse; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; + +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; + +/** Test for {@link PerRequestSubmitter}. */ +@ExtendWith(MockitoExtension.class) +class PerRequestSubmitterTest { + + @Mock private HttpClient mockHttpClient; + + @Test + public void shouldUseConfiguredHttpVersion() { + Configuration configuration = new Configuration(); + configuration.set(SINK_HTTP_VERSION, "HTTP_2"); + HttpSinkConfig sinkConfig = + HttpSinkConfig.builder() + .url("http://hello.pl") + .properties(new Properties()) + .readableConfig(configuration) + .httpPostRequestCallback(new Slf4jHttpPostRequestCallback()) + .build(); + HttpResponse httpResponse = org.mockito.Mockito.mock(HttpResponse.class); + doReturn(CompletableFuture.completedFuture(httpResponse)) + .when(mockHttpClient) + .sendAsync(any(), any()); + + PerRequestSubmitter submitter = + new PerRequestSubmitter( + sinkConfig, + new String[0], + mockHttpClient, + Executors.newSingleThreadExecutor()); + JavaNetHttpResponseWrapper responseWrapper = + submitter + .submit( + "http://hello.pl", + List.of(new HttpSinkRequestEntry("PUT", new byte[] {1}))) + .get(0) + .join(); + + assertThat(responseWrapper.getHttpRequest().getHttpRequest().version()) + .contains(HttpClient.Version.HTTP_2); + } +} diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/RequestSubmitterOidcTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/RequestSubmitterOidcTest.java new file mode 100644 index 00000000..dfac3774 --- /dev/null +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/RequestSubmitterOidcTest.java @@ -0,0 +1,233 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.http.sink.httpclient; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.http.WireMockServerPortAllocator; +import org.apache.flink.connector.http.config.HttpConnectorConfigConstants; +import org.apache.flink.connector.http.config.HttpSinkConfig; +import org.apache.flink.connector.http.sink.HttpSinkRequestEntry; +import org.apache.flink.connector.http.table.sink.Slf4jHttpPostRequestCallback; + +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.stubbing.Scenario; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.ArgumentCaptor; + +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.time.Duration; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.stream.Collectors; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.post; +import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; +import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** Tests that both sink request submitters authorize every request with a current OIDC token. */ +class RequestSubmitterOidcTest { + + private static final String TOKEN_PATH = "/token"; + + private static final String TOKEN_REQUEST = "grant_type=client_credentials&client_id=sink"; + + private WireMockServer wireMockServer; + + private HttpClient sinkHttpClient; + + @BeforeEach + void setUp() { + wireMockServer = + new WireMockServer( + wireMockConfig().port(WireMockServerPortAllocator.getServerPort())); + wireMockServer.start(); + + sinkHttpClient = mock(HttpClient.class); + HttpResponse httpResponse = mock(HttpResponse.class); + doReturn(CompletableFuture.completedFuture(httpResponse)) + .when(sinkHttpClient) + .sendAsync(any(), any()); + } + + @AfterEach + void tearDown() { + wireMockServer.stop(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void shouldAddBearerTokenAndReuseItWhileValid(boolean batchMode) { + stubToken("token-1", 3600); + AbstractRequestSubmitter submitter = submitter(batchMode, new String[0]); + + submitter.submit("http://sink", List.of(entry())).forEach(CompletableFuture::join); + submitter.submit("http://sink", List.of(entry())).forEach(CompletableFuture::join); + + assertThat(authorizationHeaders()).containsExactly("Bearer token-1", "Bearer token-1"); + wireMockServer.verify(1, postRequestedFor(urlEqualTo(TOKEN_PATH))); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void shouldRefreshExpiredTokenBeforeNextRequest(boolean batchMode) { + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_PATH)) + .inScenario("refresh") + .whenScenarioStateIs(Scenario.STARTED) + .willReturn(tokenResponse("token-1", 0)) + .willSetStateTo("expired")); + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_PATH)) + .inScenario("refresh") + .whenScenarioStateIs("expired") + .willReturn(tokenResponse("token-2", 3600))); + AbstractRequestSubmitter submitter = submitter(batchMode, new String[0]); + + submitter.submit("http://sink", List.of(entry())).forEach(CompletableFuture::join); + submitter.submit("http://sink", List.of(entry())).forEach(CompletableFuture::join); + + assertThat(authorizationHeaders()).containsExactly("Bearer token-1", "Bearer token-2"); + wireMockServer.verify(2, postRequestedFor(urlEqualTo(TOKEN_PATH))); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void shouldReplaceConfiguredAuthorizationHeaderAndKeepOthers(boolean batchMode) { + stubToken("token-1", 3600); + AbstractRequestSubmitter submitter = + submitter( + batchMode, + new String[] { + "authorization", "Basic static", "Content-Type", "application/json" + }); + + submitter.submit("http://sink", List.of(entry())).forEach(CompletableFuture::join); + + HttpRequest request = sentRequests().get(0); + assertThat(request.headers().allValues("Authorization")).containsExactly("Bearer token-1"); + assertThat(request.headers().firstValue("Content-Type")).contains("application/json"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void shouldFailWhenTokenEndpointDoesNotAnswerWithinRequestTimeout(boolean batchMode) { + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_PATH)) + .willReturn(tokenResponse("token-1", 3600).withFixedDelay(2_000))); + AbstractRequestSubmitter submitter = + submitter(batchMode, new String[0], Duration.ofMillis(200)); + + assertThatThrownBy(() -> submitter.submit("http://sink", List.of(entry()))) + .hasRootCauseInstanceOf(HttpTimeoutException.class) + .hasStackTraceContaining("OIDC access token"); + verify(sinkHttpClient, never()).sendAsync(any(), any()); + } + + private void stubToken(String token, int expiresInSeconds) { + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_PATH)).willReturn(tokenResponse(token, expiresInSeconds))); + } + + private static com.github.tomakehurst.wiremock.client.ResponseDefinitionBuilder tokenResponse( + String token, int expiresInSeconds) { + return aResponse() + .withStatus(200) + .withBody( + "{\"access_token\": \"" + + token + + "\", \"expires_in\": " + + expiresInSeconds + + "}"); + } + + private AbstractRequestSubmitter submitter(boolean batchMode, String[] headersAndValues) { + return submitter(batchMode, headersAndValues, null); + } + + private AbstractRequestSubmitter submitter( + boolean batchMode, String[] headersAndValues, Duration requestTimeout) { + Configuration configuration = new Configuration(); + configuration.set(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL, wireMockServer.baseUrl() + TOKEN_PATH); + configuration.set(SINK_OIDC_AUTH_TOKEN_REQUEST, TOKEN_REQUEST); + configuration.set(SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION, Duration.ZERO); + if (requestTimeout != null) { + configuration.set(SINK_REQUEST_TIMEOUT, requestTimeout); + } + + Properties properties = new Properties(); + properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_BATCH_REQUEST_SIZE, "10"); + HttpSinkConfig sinkConfig = + HttpSinkConfig.builder() + .url("http://sink") + .properties(properties) + .readableConfig(configuration) + .httpPostRequestCallback(new Slf4jHttpPostRequestCallback()) + .build(); + + return batchMode + ? new BatchRequestSubmitter( + sinkConfig, + headersAndValues, + sinkHttpClient, + Executors.newSingleThreadExecutor()) + : new PerRequestSubmitter( + sinkConfig, + headersAndValues, + sinkHttpClient, + Executors.newSingleThreadExecutor()); + } + + private static HttpSinkRequestEntry entry() { + return new HttpSinkRequestEntry("POST", "{}".getBytes()); + } + + private List sentRequests() { + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(HttpRequest.class); + verify(sinkHttpClient, atLeastOnce()).sendAsync(requestCaptor.capture(), any()); + return requestCaptor.getAllValues(); + } + + private List authorizationHeaders() { + return sentRequests().stream() + .map(request -> request.headers().firstValue("Authorization").orElse(null)) + .collect(Collectors.toList()); + } +} diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java index 3d578670..2db0095c 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java @@ -405,4 +405,86 @@ private void assertInvalidRetryOption(String tableName, String retryOptionLine) () -> tEnv.executeSql("INSERT INTO " + tableName + " VALUES (1)").await()) .isInstanceOf(ValidationException.class); } + + @Test + public void acceptsSinkOidcProxyAndHttpVersionOptionsTest() { + final String parityOptions = + String.format( + "CREATE TABLE httpSinkParityOptions (\n" + + " id bigint\n" + + ") with (\n" + + " 'connector' = '%s',\n" + + " 'url' = '%s',\n" + + " 'format' = 'json',\n" + + " '%s' = 'HTTP_2',\n" + + " '%s' = 'http://localhost/token',\n" + + " '%s' = 'grant_type=client_credentials',\n" + + " '%s' = '2s',\n" + + " '%s' = 'proxy.local',\n" + + " '%s' = '8080',\n" + + " '%s' = 'user',\n" + + " '%s' = 'password'\n" + + ")", + HttpDynamicTableSinkFactory.IDENTIFIER, + "http://localhost/", + HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION.key(), + HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL.key(), + HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST.key(), + HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION.key(), + HttpDynamicSinkConnectorOptions.SINK_PROXY_HOST.key(), + HttpDynamicSinkConnectorOptions.SINK_PROXY_PORT.key(), + HttpDynamicSinkConnectorOptions.SINK_PROXY_USERNAME.key(), + HttpDynamicSinkConnectorOptions.SINK_PROXY_PASSWORD.key()); + tEnv.executeSql(parityOptions); + Throwable insertFailure = + catchThrowable( + () -> + tEnv.executeSql("INSERT INTO httpSinkParityOptions VALUES (1)") + .await()); + if (insertFailure != null) { + assertThat(insertFailure) + .as("sink OIDC/proxy/http-version options must be recognized") + .isNotInstanceOf(ValidationException.class); + } + } + + @Test + public void validateSinkHttpVersionTest() { + final String invalidHttpVersion = + String.format( + "CREATE TABLE httpVersion (\n" + + " id bigint\n" + + ") with (\n" + + " 'connector' = '%s',\n" + + " 'url' = '%s',\n" + + " 'format' = 'json',\n" + + " '%s' = 'HTTP_3'\n" + + ")", + HttpDynamicTableSinkFactory.IDENTIFIER, + "http://localhost/", + HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION.key()); + tEnv.executeSql(invalidHttpVersion); + assertThatThrownBy(() -> tEnv.executeSql("INSERT INTO httpVersion VALUES (1)").await()) + .isInstanceOf(ValidationException.class); + } + + @Test + public void validateSinkOidcTokenRequestTest() { + final String missingTokenRequest = + String.format( + "CREATE TABLE httpOidc (\n" + + " id bigint\n" + + ") with (\n" + + " 'connector' = '%s',\n" + + " 'url' = '%s',\n" + + " 'format' = 'json',\n" + + " '%s' = 'http://localhost/token'\n" + + ")", + HttpDynamicTableSinkFactory.IDENTIFIER, + "http://localhost/", + HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL.key()); + tEnv.executeSql(missingTokenRequest); + assertThatThrownBy(() -> tEnv.executeSql("INSERT INTO httpOidc VALUES (1)").await()) + .isInstanceOf(ValidationException.class); + } } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpSinkConfigSerializationTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpSinkConfigSerializationTest.java index 4411c587..ab640a4c 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpSinkConfigSerializationTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpSinkConfigSerializationTest.java @@ -35,7 +35,15 @@ import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_IGNORED_RESPONSE_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_RETRY_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_SUCCESS_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_VERSION; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_MAX_RETRIES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_HOST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PASSWORD; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PORT; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_USERNAME; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_FIXED_DELAY_DELAY; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_WRITER_THREAD_POOL_SIZE; @@ -51,6 +59,7 @@ public void testHttpSinkConfigSerialization() throws Exception { config.set(SINK_REQUEST_TIMEOUT, Duration.ofMinutes(2)); config.set(SINK_WRITER_THREAD_POOL_SIZE, 3); config.set(SINK_MAX_RETRIES, 7); + config.set(SINK_HTTP_VERSION, "HTTP_2"); config.set(SINK_HTTP_SUCCESS_CODES, "2XX,404"); config.set(SINK_HTTP_RETRY_CODES, "500"); config.set(SINK_HTTP_IGNORED_RESPONSE_CODES, "404"); @@ -82,6 +91,7 @@ public void testHttpSinkConfigSerialization() throws Exception { assertThat(deserialized.getRequestTimeout()).isEqualTo(Duration.ofMinutes(2)); assertThat(deserialized.getWriterThreadPoolSize()).isEqualTo(3); assertThat(deserialized.getMaxRetries()).isEqualTo(7); + assertThat(deserialized.getHttpVersion()).isEqualTo("HTTP_2"); assertThat(deserialized.getSuccessCodes()).isEqualTo("2XX,404"); assertThat(deserialized.getRetryCodes()).isEqualTo("500"); assertThat(deserialized.getIgnoredResponseCodes()).isEqualTo("404"); @@ -97,6 +107,7 @@ public void testHttpSinkConfigDefaults() { .build(); assertThat(config.getRequestTimeout()).isEqualTo(Duration.ofSeconds(30)); + assertThat(config.getHttpVersion()).isEqualTo("HTTP_1_1"); assertThat(config.getWriterThreadPoolSize()).isEqualTo(4); assertThat(config.getMaxRetries()).isEqualTo(0); assertThat(config.getSuccessCodes()).isEqualTo("2XX"); @@ -108,6 +119,14 @@ public void testHttpSinkConfigDefaults() { public void testFromDataStreamMergesPropertiesIntoReadableConfig() { Properties properties = new Properties(); properties.setProperty(SINK_REQUEST_TIMEOUT.key(), "45s"); + properties.setProperty(SINK_HTTP_VERSION.key(), "HTTP_2"); + properties.setProperty(SINK_PROXY_HOST.key(), "proxy"); + properties.setProperty(SINK_PROXY_PORT.key(), "8080"); + properties.setProperty(SINK_PROXY_USERNAME.key(), "user"); + properties.setProperty(SINK_PROXY_PASSWORD.key(), "password"); + properties.setProperty(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL.key(), "http://token"); + properties.setProperty(SINK_OIDC_AUTH_TOKEN_REQUEST.key(), "grant_type=client_credentials"); + properties.setProperty(SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION.key(), "2s"); properties.setProperty(SINK_WRITER_THREAD_POOL_SIZE.key(), "2"); properties.setProperty(SINK_MAX_RETRIES.key(), "8"); properties.setProperty(SINK_HTTP_SUCCESS_CODES.key(), "2XX,201"); @@ -120,6 +139,17 @@ public void testFromDataStreamMergesPropertiesIntoReadableConfig() { "http://localhost", properties, new Slf4jHttpPostRequestCallback()); assertThat(config.getRequestTimeout()).isEqualTo(Duration.ofSeconds(45)); + assertThat(config.getHttpVersion()).isEqualTo("HTTP_2"); + assertThat(config.getReadableConfig().get(SINK_PROXY_HOST)).isEqualTo("proxy"); + assertThat(config.getReadableConfig().get(SINK_PROXY_PORT)).isEqualTo(8080); + assertThat(config.getReadableConfig().get(SINK_PROXY_USERNAME)).isEqualTo("user"); + assertThat(config.getReadableConfig().get(SINK_PROXY_PASSWORD)).isEqualTo("password"); + assertThat(config.getReadableConfig().get(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL)) + .isEqualTo("http://token"); + assertThat(config.getReadableConfig().get(SINK_OIDC_AUTH_TOKEN_REQUEST)) + .isEqualTo("grant_type=client_credentials"); + assertThat(config.getReadableConfig().get(SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION)) + .isEqualTo(Duration.ofSeconds(2)); assertThat(config.getWriterThreadPoolSize()).isEqualTo(2); assertThat(config.getMaxRetries()).isEqualTo(8); assertThat(config.getSuccessCodes()).isEqualTo("2XX,201"); diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java index bcb16447..ad7d4716 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java @@ -28,7 +28,11 @@ import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_OIDC_AUTH_TOKEN_REQUEST; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; +import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; /** tests for {@link HttpHeaderUtils}. */ public class HttpHeaderUtilsTest { @@ -44,4 +48,23 @@ void shouldCreateOIDCHeaderPreprocessorTest() { headerPreprocessor = HttpHeaderUtils.createOIDCHeaderPreprocessor(configuration); assertThat(headerPreprocessor).isNotNull(); } + + @Test + void shouldCreateSinkOidcAccessTokenManagerTest() { + Configuration configuration = new Configuration(); + assertThat(HttpHeaderUtils.createSinkOidcAccessTokenManager(configuration)).isNull(); + configuration.set(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL, "http://aaa"); + configuration.set(SINK_OIDC_AUTH_TOKEN_REQUEST, "ccc"); + configuration.set(SINK_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION, Duration.ofSeconds(1)); + assertThat(HttpHeaderUtils.createSinkOidcAccessTokenManager(configuration)).isNotNull(); + } + + @Test + void shouldRequireSinkOidcTokenRequestWhenEndpointIsSet() { + Configuration configuration = new Configuration(); + configuration.set(SINK_OIDC_AUTH_TOKEN_ENDPOINT_URL, "http://aaa"); + assertThatThrownBy(() -> HttpHeaderUtils.createSinkOidcAccessTokenManager(configuration)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(SINK_OIDC_AUTH_TOKEN_REQUEST.key()); + } } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java index 2c671e66..8aca5296 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java @@ -20,8 +20,10 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.http.WireMockServerPortAllocator; +import org.apache.flink.connector.http.config.HttpSinkConfig; import org.apache.flink.connector.http.table.lookup.HttpLookupConfig; import org.apache.flink.connector.http.table.lookup.Slf4JHttpLookupPostRequestCallback; +import org.apache.flink.connector.http.table.sink.Slf4jHttpPostRequestCallback; import org.apache.flink.util.concurrent.ExecutorThreadFactory; import org.junit.jupiter.api.Test; @@ -40,6 +42,10 @@ import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_PASSWORD; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_PORT; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_USERNAME; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_HOST; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PASSWORD; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_PORT; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_PROXY_USERNAME; import static org.assertj.core.api.Assertions.assertThat; class JavaNetHttpClientFactoryTest { @@ -138,4 +144,49 @@ public void shouldGetClientWithExecutor() { HttpClient client = JavaNetHttpClientFactory.createClient(properties, httpClientExecutor); assertThat(client.followRedirects().equals(HttpClient.Redirect.NORMAL)).isTrue(); } + + @Test + public void shouldGetSinkClientWithProxyAuthenticator() throws UnknownHostException { + Properties properties = new Properties(); + Configuration configuration = new Configuration(); + configuration.set(SINK_PROXY_HOST, "google"); + configuration.set(SINK_PROXY_PORT, PROXY_SERVER_PORT); + configuration.set(SINK_PROXY_USERNAME, "username"); + configuration.set(SINK_PROXY_PASSWORD, "password"); + + HttpSinkConfig sinkConfig = + HttpSinkConfig.builder() + .url("https://google.com") + .readableConfig(configuration) + .properties(properties) + .httpPostRequestCallback(new Slf4jHttpPostRequestCallback()) + .build(); + ExecutorService httpClientExecutor = + Executors.newFixedThreadPool( + 1, + new ExecutorThreadFactory( + "http-sink-client-batch-request-worker", + ThreadUtils.LOGGING_EXCEPTION_HANDLER)); + + HttpClient client = JavaNetHttpClientFactory.createClient(sinkConfig, httpClientExecutor); + + assertThat(client.authenticator().isPresent()).isTrue(); + assertThat(client.proxy().isPresent()).isTrue(); + + PasswordAuthentication auth = + client.authenticator() + .get() + .requestPasswordAuthenticationInstance( + "google", + InetAddress.getByName("127.0.0.1"), + SERVER_PORT, + "http", + "Please authenticate", + "basic", + null, + Authenticator.RequestorType.PROXY); + + assertThat(auth.getUserName().equals("username")).isTrue(); + assertThat(Arrays.equals(auth.getPassword(), "password".toCharArray())).isTrue(); + } }