[bharathgunapati]

Purpose of the change

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

The HTTP sink only reported failures through Flink's generic numRecordsSendErrors counter and a few logs without request context. With configurable status-code classification and retries (FLINK-40277, FLINK-40278), operators need to see why writes fail or retry. This change adds sink metrics, status-code counters, a request latency histogram, and failure logs with request context, and keeps the cause of client-side exceptions.

Replaces the earlier stacked draft on my fork; rebuilt on top of main after #54 and #57.

Brief change log

• SinkHttpClientStatistics (new, @PublicEvolving): observability data a SinkHttpClient collects for one write across all retry attempts: HTTP request count, request latencies, request entries per status code, retryable responses, transport failures, and retry attempts. SinkHttpClientResponses carries it as one new field with a new constructor; the existing constructors keep working and return SinkHttpClientStatistics.empty(), so custom clients are unaffected.
• HttpSinkWriter registers metrics in the http_sink_connector group next to the lookup source's http_lookup_connector group: numHttpRequests, requestLatencyMs (histogram), numRetryableResponseFailures, numRequestExceptions, numRetryAttempts, numRetryExhausted, numFatalResponseFailures, numIgnoredResponses, and one counter per HTTP status code in the status_code subgroup, all counted in request entries. Failure logs now name the endpoint and the number of affected entries.
• JavaNetSinkHttpClient records a status code and latency per HTTP response, counts transport failures, and logs each retryable, fatal or ignored response with status, method, endpoint and entry count. The retryable-without-response log carries the original exception as its cause.
• Both request submitters measure request duration and hand a failed request's exception to the client instead of logging it themselves without context; JavaNetHttpResponseWrapper carries the duration and the failure.
• HttpSinkClientWithRetry aggregates statistics over attempts and counts entries scheduled for another attempt; the last allowed attempt counts as exhausted, not as a retry.
• HttpSinkRequestLatencyHistogram: a bounded, lock-free histogram over the last 1024 samples. Flink's DescriptiveStatisticsHistogram lives in flink-runtime, which the connector does not import in production code, so this avoids a new dependency surface.
• Docs: a "Sink failure observability" section with the metric table on the Table and DataStream pages (en + zh).

Verifying this change

• HttpSinkWriterTest: each client outcome increments the matching counters and status-code counters; latency samples reach the histogram; a transport error before any response counts as a request exception; a status-code counter is registered once and reused.
• JavaNetSinkHttpClientTest: statistics for a mixed set of responses; one batch response with three entries counts three for its status code; transport failures without a response; retry attempts counted across attempts and the exhausted attempt not counted as a retry; a failing HTTP client through both submitters; both submitters keep the original exception and elapsed time of a failed request; the old constructors return empty statistics; null statistics collections are rejected.
• BatchRequestSubmitterTest: a failed request keeps its cause and duration.
• HttpSinkRequestLatencyHistogramTest: statistics, bounded retention, empty state, concurrent updates.
• mvn -pl flink-connector-http -DskipITs test: full unit suite including ArchUnit; spotless:check checkstyle:check clean.
• End to end, in a Flink mini-cluster against an embedded HTTP server, in the examples repo: DataStreamHttpSinkMetricsIT (commit 6209f74, CI run, 32 tests green). A recording metric reporter reads the http_sink_connector metrics back after each job:
  • batch that succeeds on retry: numHttpRequests 2, numRetryableResponseFailures 2, numRetryAttempts 2, status_code.503 2, status_code.200 2, latency histogram count 2, numRecordsSendErrors 0
  • 503 on a three-entry batch: status_code.503 3, numRetryAttempts 3
  • retry exhaustion with one retry: numRetryableResponseFailures 2, numRetryAttempts 1, numRetryExhausted 1
  • fatal 400: numFatalResponseFailures 1, status_code.400 1, numRecordsSendErrors 1
  • ignored 404: numIgnoredResponses 1, no send errors
  • unreachable endpoint: numRequestExceptions 1, numRetryExhausted 1, no status-code counter

Impact and documentation

• Dependencies: no
• Public API: SinkHttpClientStatistics added; SinkHttpClientResponses gains a field and a constructor, existing constructors unchanged
• Runtime per-record code paths: yes, the sink records counters and a latency sample per HTTP request; the per-entry cost is one map merge per response
• Documentation: "Sink failure observability" section on all four connector pages
• Overlap with [FLINK-40279][connector-http] Add HTTP sink OIDC, proxy and HTTP version options #58: both PRs add helper methods after close() in AbstractRequestSubmitter; whichever merges second needs a trivial rebase (the two blocks are independent, verified by stacking both on main).

---

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

• Yes (please specify the tool below)

Generated-by: Cursor (auto mode) for the original draft; Claude Code (Claude Fable 5.1) for the port onto main