GeezerVison

blp left a comment

…nnel

A transport output endpoint got only an error closure from
`OutputEndpoint::connect`. It had no way to report its health or to
export its own metrics, which only integrated sinks could do because
they hold the controller.

Replace the closure with the trait object `OutputEndpointCallbacks`,
the output counterpart of the reporting methods of `InputConsumer`:

| Method                    | Controller side                       |
|---------------------------|---------------------------------------|
| `error`                   | `output_transport_error`, as before   |
| `update_connector_health` | `update_output_connector_health`      |
| `set_custom_metrics`      | `ControllerStatus::set_output_custom_metrics` |

The two Kafka sinks keep the trait object where they kept the closure.
The other endpoints ignore it as they ignored the closure. Tests that
drive an endpoint without a controller use `MockOutputCallbacks`.

The new controller test connects an endpoint that reports a gauge, its
health and an error through the channel, and checks the endpoint status
and the Prometheus text. With the health and metrics forwarding stubbed
out, it fails on the health assertion.

First step for #7578: the Kafka sinks need this channel to show queue
depth and to report themselves unhealthy while they retry.

Signed-off-by: Xavier Bestel <[email protected]>

The non-fault-tolerant Kafka sink dropped undelivered messages after
librdkafka's five-minute default timeout while the pipeline kept running.
Default to unbounded delivery retries so a sustained outage fills the
producer queue and applies backpressure. Preserve either explicit timeout
alias so deployments can retain a finite deadline.

Report health from pending delivery age and delivery results. An enqueue
can succeed because a failed delivery freed space, so it cannot establish
recovery. A periodic task on the shared runtime detects quiet stalled
sinks even with statistics disabled. Track delivery identities so progress
on one partition cannot hide an older pending record on another.

Expose queue, request, delivery, and wait metrics through the output
callback API. Keep broker counters monotonic across resets and include
active and interrupted queue waits. Log temporary disconnections without
reporting lost output. Document the changed default and its limits.

Validate timeout aliases, outage recovery and payloads, quiet sinks,
expiry followed by enqueue, out-of-order delivery, fatal health, counter
resets, and cancellation. Check controller callback forwarding and
backpressure with the static engine. The alias and enqueue-health tests
fail when their respective fixes are removed.

Second step for #7578, stacked on #7639.

Signed-off-by: Xavier Bestel <[email protected]>