[bharathgunapati]

Purpose of the change

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

HttpSinkClientWithRetry hands each attempt to Resilience4j's Retry.decorateCompletionStage. Resilience4j 1.7.1 does not catch an exception thrown while starting an attempt. On the first attempt the exception reaches the caller, but a retry attempt runs on the retry scheduler, where the exception is lost and the returned future never completes. The sink request then stays in flight until the AsyncSink request timeout (10 minutes by default), blocking checkpoints, and is afterwards resent as a whole batch including entries the first attempt already delivered.

On main this needs an attempt that fails deterministically only on a retry. With FLINK-40279 (#58) the OIDC token is fetched while building each request, so a token endpoint failure during a retry triggers it. Introduced with FLINK-40277, not released.

Brief change log

• HttpSinkClientWithRetry wraps the attempt submitter: an exception thrown while starting an attempt becomes a failed stage. Resilience4j then applies the existing retryOnException rule, so a failure with an IOException in its cause chain is retried like a transport error, and anything else fails the request, which HttpSinkWriter reports through its existing exception path.
• Side effect for the first attempt: an exception there is now also reported through the returned future instead of being thrown from putRequests, which is what HttpSinkWriter expects anyway.

Verifying this change

• HttpSinkClientWithRetryTest: a retry attempt that throws completes the request exceptionally with the original cause; a retry attempt that throws with an IOException cause is retried and succeeds on the next attempt; a first attempt that throws completes exceptionally. All three time out on the unfixed code (5 s bound) and pass with the fix.
• JavaNetSinkHttpClientTest: the same scenario through the real sink client with a submitter that returns 503 first and throws on the retry.
• Full unit suite including ArchUnit; spotless:check checkstyle:check clean.

Impact and documentation

• Dependencies: no
• Public API: no
• Runtime per-record code paths: one try/catch around starting an attempt
• Documentation: not needed
• Follow-up worth considering separately: HttpSinkWriter could set an AsyncSink request timeout derived from http.sink.request.timeout and the retry settings, so that any future hang is bounded. Left out here to keep the bug fix minimal.

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Fable 5.1)

[bharathgunapati]

@MartijnVisser this is the fix for FLINK-41192, which you filed. An exception thrown while starting an attempt now becomes a failed stage, so Resilience4j's retryOnException sees it: an I/O cause is retried like a transport error, anything else fails the request through the writer's existing exception path.

The three new tests in HttpSinkClientWithRetryTest and the one in JavaNetSinkHttpClientTest follow your reproduction (retryable result on the first attempt, throw on the second). They time out after 5 s on main and pass with the fix.

Since #58 makes the bug reachable through the per-request token fetch, this should go in before #58. Could you approve the CI run when you get a chance? Thanks.