[bharathgunapati]

Purpose of the change

https://issues.apache.org/jira/browse/FLINK-40279

Bring HTTP sink configuration closer to the lookup source for OIDC authentication, HTTP proxy settings, and HTTP version selection (FLINK-40275 umbrella follow-up after #54).

Brief change log

• Add sink options for http.sink.http-version, http.sink.proxy.*, and shared http.security.oidc.* wiring on the sink path.
• OIDC on the sink works like lookup: each submitter holds one OidcAccessTokenManager. Each HTTP request (per-request mode, batch mode, and retries) gets a current Authorization: Bearer <token> header. The token is cached until expires_in - http.security.oidc.token.expiry.reduction and then refreshed. The bearer replaces any static http.sink.header.Authorization; other headers are unchanged. It doesn't need an Authorization header to be declared.
• http.security.oidc.token.request is required when http.security.oidc.token.endpoint.url is set (validated for both Table and DataStream).
• Lookup OIDC now caches the token as well. Before, OIDCAuthHeaderValuePreprocessor created a new OidcAccessTokenManager on every call and JavaNetHttpPollingClient built a new preprocessor for every lookup, so each lookup fetched a new token. Now the polling client builds the preprocessor once, and the preprocessor keeps its token manager (transient, thread-safe) and reuses the token until it expires.
• Proxy and HTTP version are applied to the sink HttpClient and requests.
• Document the new sink options in Table and DataStream docs (en + zh).

Review follow-ups:

• Renamed the constant SINK_QUERY_HTTP_VERSION to SINK_HTTP_VERSION; QUERY had no meaning on the sink.
• Rewrote the comment on the transient token manager in OIDCAuthHeaderValuePreprocessor to say exactly what survives (the token across requests of one instance) and what does not (the manager across deserialization).
• Bounded the token request. OidcAccessTokenManager previously sent the token request without a timeout and its HttpClient had no connect timeout, so an unanswered identity provider could block the calling thread indefinitely. The token is fetched on the thread that builds the HTTP request (the sink writer thread or the lookup thread), so this now uses the connector's request timeout: http.sink.request.timeout for the sink and http.source.lookup.request.timeout for lookups, for both connecting and reading. The default when constructed directly is 30s. A timeout fails with a message naming the token endpoint.
• Documented under Restrictions that the token request does not use the connector's proxy or TLS trust options, and that it is bounded by the request timeout.

Compatibility note

• New optional options only; existing jobs are unchanged when unset.
• Defaults preserve post-[FLINK-40277][connector-http] Add HTTP sink status-code handling and retries #54 behaviour (max-retries default 0, writer thread pool default 4, HTTP version HTTP_1_1).
• OIDC default change: http.security.oidc.token.expiry.reduction now defaults to 30s (was 1s) for both lookup and sink. The old default had no effect while lookup fetched a token for every request. Now that tokens are cached, 1s left no room for network latency, request duration, or clock skew. Jobs that set the option explicitly are unaffected. If a token's expires_in is not greater than the reduction, the full lifetime is used.
• Lookup OIDC behaviour change: lookups used to fetch a new token for every request. They now reuse the cached token until expires_in - http.security.oidc.token.expiry.reduction, as the OIDC docs already described ("cached and used for subsequent requests"). If the identity provider revokes a token before it expires, requests keep sending it until it expires, and the endpoint can reject them during that time. The token is still fetched when a request is sent, not at plan time, so EXPLAIN makes no network calls (HTTP-145).

Verifying this change

• RequestSubmitterOidcTest (WireMock token endpoint, both submitters):
  • The bearer is added and reused while valid (one token call for two requests).
  • An expired token is refreshed before the next request.
  • A static Authorization header is replaced and other headers are kept.
  • A token endpoint that does not answer within http.sink.request.timeout fails the submit with the timeout as root cause, and no sink request is sent.
• OIDCAuthHeaderValuePreprocessorTest: the token is reused while valid, refreshed after it expires, still works after serialization, and fails with a timeout when the token endpoint does not answer in time.
• HttpHeaderUtilsTest, HttpSinkConfigSerializationTest, HttpDynamicTableSinkFactoryTest, JavaNetHttpClientFactoryTest, PerRequestSubmitterTest, BatchRequestSubmitterTest.
• mvn -pl flink-connector-http spotless:check checkstyle:check test: all tests pass.

Impact and documentation

• Dependencies: no
• Public API: new optional connector options only
• Runtime per-record code paths: the sink builds request headers per HTTP request; with OIDC configured this reads the cached token under a lock and only calls the token endpoint when the token is about to expire
• Documentation: sink OIDC / proxy / HTTP version options documented in Table and DataStream docs (en + zh). The shared OIDC Bearer Authentication section now says lookup and sink tables use the same option names but are configured and authenticated independently, and that the token is applied per request (including sink batches and retries). It also fixes the existing Authentication -> Authorization header typo.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Cursor (auto mode) for the initial change; Claude Code (Claude Fable 5.1) for the review follow-ups

[davidradl]

we should not have QUERY in the variable name as it is confusing

[bharathgunapati]

Agreed. I renamed it to SINK_HTTP_VERSION. The lookup side still has SOURCE_LOOKUP_QUERY_HTTP_VERSION. I left that one alone to keep this PR focused on the sink, but I can rename it in a follow-up if you want both to match.

[davidradl]

is the comment true that the cached token survives. It seemed to work before when we did not cache the token manager. I think caching the token manager makes sense.

[bharathgunapati]

Yes, the comment is true, but it was too short, so I rewrote it.

What happens now: the token manager is created once per preprocessor instance and kept. Every lookup that goes through the same instance reuses the cached token until it expires. The field is transient, so after deserialization it is created again and a new token is fetched on the next request. The test shouldWorkAfterSerialization covers that case.

Before this PR it did work, but it fetched a new token for every lookup. The polling client created a new preprocessor for each lookup, and the preprocessor created a new token manager on each call, so nothing was ever reused. Now the polling client creates the preprocessor once and the preprocessor keeps its token manager. That is what makes the cache real.

[davidradl]

am I right in saying the documented and coded default was 1 second - we are changing it to 30 seconds. Looks like we need to be consistent with existing docs and use 1 second not 30

[bharathgunapati]

You are right that the old default was 1 second in both the code and the docs. I would like to keep 30 seconds. Here is why.

Before this PR the 1 second default did nothing. The lookup fetched a new token on every request, so no token ever lived long enough to get close to its expiry. Now tokens are cached, so this setting is what protects the last requests before a token expires. With 1 second, a request sent just before the cutoff can reach the server after the token has already expired. One second is not enough to cover the request time, the token endpoint time, and clock differences between Flink and the identity provider. The user then gets a 401 that they cannot explain from their config. 30 seconds is the usual value in OIDC clients.

The cost is small. A token is refreshed 29 seconds earlier than before. Tokens usually live for 5 to 60 minutes, so the extra token calls are negligible. Jobs that set http.security.oidc.token.expiry.reduction themselves are not affected. If a token's expires_in is 30 seconds or less, the full lifetime is used, so short test tokens still work. The change is listed in the compatibility note in the PR description.

If you still prefer 1 second in this PR, I will revert it and open a follow-up Jira for the default. But I think users are better off with 30 seconds from the first release where caching works.

[bharathgunapati]

@davidradl Thanks for the review. All three comments are addressed in the latest push, see the threads above.

One more change that is not from the review: the token request had no timeout, and its HTTP client had no connect timeout. If the identity provider did not answer, the sink writer thread would wait forever. The token request now uses the connector's request timeout: http.sink.request.timeout for the sink and http.source.lookup.request.timeout for lookups. A timeout fails the request with a message that names the token endpoint. New tests in RequestSubmitterOidcTest and OIDCAuthHeaderValuePreprocessorTest cover this.

All of this also runs end to end in a Flink mini-cluster in the examples repo, see commit 760b181 and DataStreamHttpSinkIT. The embedded server plays the token endpoint, a bearer-protected endpoint and a proxy that requires basic auth. The log shows one token call for three sink requests, a new token per request when tokens expire immediately, the static Authorization header replaced by the bearer, the 407 challenge followed by the authenticated retry through the proxy with the absolute target URI, the h2c upgrade for http.sink.http-version = HTTP_2, and the job failing with "OIDC access token" when the token endpoint does not answer within http.sink.request.timeout.

I also added two lines to the Restrictions section: the token request does not use the proxy or TLS trust options, and it is bounded by the request timeout. Making the token request use the proxy and trust settings is a good follow-up for lookup and sink together.