Proposal: KAFKA CONNECT: JMX metrics for the iceberg-kafka-connect commit pipeline

### Proposed Change

## Status

Draft. Implementation is up as [PR #17025](https://github.com/apache/iceberg/pull/17025).

The PR is functionally complete and reviewed, but the reviewers (@AnatolyPopov,
@laskoviymishka) asked that the **metric surface itself** — the JMX domain, the metric names,
and the tag keys — be settled by the wider community before it merges, since those strings
become a public operational contract the moment they ship. This proposal exists to get that
agreement.

## Discussion Thread

[Discussion Thread] (https://lists.apache.org/thread/n30rqprw6p6wmsqxvot1k4k4j31j2pt6)

## Proposed Change

Add JMX metrics to `iceberg-kafka-connect` covering the Worker → Coordinator commit pipeline:
per-record control-topic decode/dispatch timings, `Worker.save()` timing, emitted-event
counters, `Coordinator.commit()` timings split by full vs. timeout-driven partial commit, and
gauges for the coordinator's in-memory commit buffers.

The connector today is close to unobservable in production. Everything below is derived from
running this connector at a few thousand sink tasks, where diagnosing a stalled commit
pipeline currently means reading task logs across the fleet.

## Motivation

Kafka Connect exposes rich JMX metrics for the framework's own machinery — poll rates,
put-batch times, offset commit latency, per-task record counts. None of it reaches inside a
sink connector's own pipeline. For `iceberg-kafka-connect` specifically, the interesting
failure modes are all invisible:

1. **Coordinator commit latency and failure pattern.** One coordinator writes metadata to the
catalog for the whole connector. When the catalog slows down or a commit fails, the only
evidence is log lines on whichever task happens to hold the coordinator role. There is no
time series to alert on, and no way to see whether commits are getting slower before they
start failing outright.

2. **Partial vs. full commits.** A timeout-driven partial commit
(`iceberg.control.commit.timeout-ms`) means the coordinator gave up waiting for some
worker's `DATA_COMPLETE`. Operationally that is a completely different event from a healthy
full commit, and today the two are indistinguishable from outside the process.

3. **Control-topic backlog.** The coordinator buffers `DATA_WRITTEN`/`DATA_COMPLETE` envelopes
in memory for the duration of a commit cycle. Unbounded growth here is the leading
indicator of the failure mode described in
[#16389](https://github.com/apache/iceberg/issues/16389) (no backpressure from coordinator
to worker), where a transient catalog outage degrades the connector for hours. There is
currently no signal for it at all.

4. **Where time goes on the worker.** `Worker.save()` covers record conversion plus the write
to object storage. Separating that from control-message handling is the difference between
"slow storage" and "slow commit pipeline" during an incident.

This proposal deliberately does *not* try to solve backpressure; it provides the
instrumentation that makes the problem in #16389 visible and any future fix measurable.

## Public Interfaces

This is the section that matters for the discussion — everything here is a compatibility
surface.

### JMX ObjectName layout

Metrics are registered through Kafka's own `org.apache.kafka.common.metrics.Metrics` with a
`JmxReporter`, so the resulting ObjectName follows Kafka's standard shape:

```
:type=[,=]*
```

with the domain fixed to `iceberg.kafka.connect`. The dotted form is deliberate: it sits
alphabetically alongside the `kafka.connect.*` beans an operator is already looking at in
jconsole, rather than in a separate hyphenated island.

Concrete ObjectNames:

| ObjectName | Cardinality |
| --- | --- |
| `iceberg.kafka.connect:type=worker-metrics,connector=,task=` | one per sink task |
| `iceberg.kafka.connect:type=coordinator-metrics,connector=,task=coordinator` | one per connector |
| `iceberg.kafka.connect:type=coordinator-metrics,connector=,task=coordinator,commit-mode=full` | one per connector |
| `iceberg.kafka.connect:type=coordinator-metrics,connector=,task=coordinator,commit-mode=partial` | one per connector |

### Tag schema

| Tag | Values | Notes |
| --- | --- | --- |
| `connector` | connector name | from `IcebergSinkConfig#connectorName()` |
| `task` | task id, or the literal `coordinator` | see open question **Q6** |
| `commit-mode` | `full` \| `partial` | only on the coordinator's commit timers; see **Q5** |

### Stat conventions

Every timing metric is published as a **pair** of cumulative stats rather than the more usual
avg/max:

| Suffix | Kafka stat | Meaning |
| --- | --- | --- |
| `-total` | `CumulativeSum` | cumulative total since task start |
| `-count` | `CumulativeCount` | number of recorded samples since task start |

Rationale: Kafka's `Avg`/`Max` are *sampled* stats bounded by `MetricConfig`'s window, which
defaults to 30s × 2 samples = 60s. The default Iceberg commit interval is 300s. An `Avg`/`Max`
over commit latency would therefore read `NaN` for roughly four minutes out of every five.
Tying the sample window to the commit interval was considered and rejected (see *Rejected
Alternatives*). A cumulative total/count pair is window-independent, and a scrape-to-scrape
rate or mean is trivially derivable by any TSDB — `rate(total)/rate(count)`.

**All durations are in microseconds.** Timing is captured with `System.nanoTime()` (not
`currentTimeMillis()`, which can go backwards under NTP slew and permanently corrupt a
`CumulativeSum`) and divided by 1_000. Milliseconds were rejected because integer division
floors sub-millisecond control-message handling to 0.

### Worker metrics

`type=worker-metrics,connector=,task=`

| Attribute | Type | Unit | Description |
| --- | --- | --- | --- |
| `save-time-total` | CumulativeSum | µs | Time spent in `Worker.save()` |
| `save-time-count` | CumulativeCount | calls | Number of `Worker.save()` calls |
| `data-files-written-total` | CumulativeSum | files | Files in successfully emitted `DATA_WRITTEN` events. **See Q3** |
| `data-complete-total` | CumulativeSum | events | Successfully emitted `DATA_COMPLETE` events |
| `channel-message-read-time-total` | CumulativeSum | µs | Avro-decode time for control messages |
| `channel-message-read-time-count` | CumulativeCount | messages | Control messages decoded |
| `channel-message-process-time-total` | CumulativeSum | µs | Dispatch time for control messages |
| `channel-message-process-time-count` | CumulativeCount | messages | Control messages dispatched |

### Coordinator metrics

`type=coordinator-metrics,connector=,task=coordinator`

| Attribute | Type | Unit | Description |
| --- | --- | --- | --- |
| `start-commit-total` | CumulativeSum | events | Successfully emitted `START_COMMIT` events. **See Q4** |
| `commit-complete-total` | CumulativeSum | events | Successfully emitted `COMMIT_COMPLETE` events. **See Q5** |
| `commit-buffer-size` | Gauge | envelopes | Current `CommitState.commitBuffer` size |
| `ready-buffer-size` | Gauge | envelopes | Current `CommitState.readyBuffer` size |
| `channel-message-read-time-{total,count}` | as above | µs / messages | Same channel timers as the worker |
| `channel-message-process-time-{total,count}` | as above | µs / messages | Same channel timers as the worker |

`type=coordinator-metrics,connector=,task=coordinator,commit-mode=`

| Attribute | Type | Unit | Description |
| --- | --- | --- | --- |
| `commit-time-total` | CumulativeSum | µs | Time in `Coordinator.commit()`, **successful or failed** |
| `commit-time-count` | CumulativeCount | commits | Commit attempts |

Note the commit timer is recorded from a `finally` block, so it deliberately spans failed
commits too — a commit that burns thirty seconds in the catalog and *then* fails is exactly the
sample an operator most wants to see. Splitting success from failure by tag is a possible
refinement; see **Q5**.

### Lifecycle

MBeans register when `WorkerMetrics`/`CoordinatorMetrics` are constructed and unregister on
`close()`. `close()` is idempotent, and the coordinator closes from both `terminate()` and
`stop()`: coordinator shutdown does not join the coordinator thread, so on re-election a
replacement coordinator can register the same connector-level ObjectName before the outgoing
one finishes, and a non-idempotent late `close()` would unregister the *replacement's* beans.

## Proposed Changes

Implementation shape (all within `org.apache.iceberg.connect.channel`):

- `ChannelMetrics` — abstract base owning the `Metrics` registry, the `JmxReporter`, the
sensor/gauge helpers, the tag builder, and the two channel timers every `Channel` feeds.
- `WorkerMetrics` / `CoordinatorMetrics` — subclasses adding their own sensors. A shared base
exists specifically so the two registries cannot drift in naming or unit conventions.
- `Channel` gains an abstract `getChannelMetrics()`, so the control-topic loop times decode and
dispatch once, in the base class, for both roles.
- Buffer gauges are lazy suppliers read at JMX poll time, so nothing is computed unless
something is scraping.

Two points worth flagging as decided rather than accidental:

- **The blocking `consumer.poll()` is not timed.** The coordinator polls with a 1s timeout, so
an idle poll would dominate any such metric, and the Kafka consumer already reports its own
poll metrics.
- **Counters are incremented after `send()` returns**, so they count events successfully
emitted, not attempted. This is the source of the naming concern in **Q4**.

Two correctness items raised in review are being fixed in the PR and are not up for
discussion here: the coordinator's `channel-message-process-time` currently absorbs a whole
inline commit (double-counting `commit-time-total` and going bimodal), and the buffer-size
gauges read `ArrayList.size()` from the JMX thread with no happens-before against the
coordinator thread.

## Compatibility, Deprecation, and Migration Plan

- **Purely additive.** No existing metric, config property, or public API changes.
- **No new config.** Metrics register when a task starts (see **Q8** if that should change).
- **No new dependency.** `org.apache.kafka.common.metrics` is already on the Connect runtime
classpath.
- **Overhead** is a few counters per task plus two lazily-evaluated gauges per connector; the
gauges do no work unless scraped.
- **Forward compatibility is the entire point of this proposal.** Once these attribute names
ship in a release, operators bind dashboards and alerts to the exact strings. JMX offers no
deprecation path: a rename does not warn, it just produces empty panels and alerts that stop
evaluating. Any rename after GA would need a full release cycle publishing both old and new
names, which is why the naming questions above should be closed *before* merge.

## Rejected Alternatives

1. **Sampled `Avg`/`Max` stats on the timers.** Rejected: the default 30s × 2-sample window is
five times shorter than the default 300s commit interval, so they read `NaN` most of the
time.
2. **Tying `MetricConfig`'s sample window to `iceberg.control.commit.interval-ms`.** Rejected:
the commit interval is operator-configurable and not precisely honoured (commit cycles drift
with catalog latency), so this couples the metric window to execution timing and produces a
window that is subtly wrong for everyone.
3. **Reusing Kafka Connect's own `SinkTaskMetricsGroup`.** Rejected: the Connect framework does
not expose a way for a plugin to add metrics to its registry, and writing into the
`kafka.connect` domain from a connector risks ObjectName collisions.
4. **Micrometer or Dropwizard Metrics.** Rejected: a new dependency in the connector's shaded
runtime for no functional gain over the metrics library already present.
5. **Logging only / an admin REST endpoint.** Rejected: neither is alertable, and JMX is what
Connect operators already scrape.
6. **Wall-clock (`System.currentTimeMillis()`) timing, and milliseconds as the unit.** Rejected:
NTP slew can produce negative deltas that permanently corrupt a `CumulativeSum`, and
millisecond integer division floors sub-millisecond control-message handling to 0.

## Test Plan

Covered by unit tests in the PR:

- MBeans register on construction under the exact expected ObjectName, and unregister on
`close()` — the latter matters because a task restart would otherwise hit
`InstanceAlreadyExistsException`.
- Attribute values read back through the platform `MBeanServer` (not through the internal
registry), so the assertions exercise the real JMX contract.
- `close()` is idempotent and does not unregister a replacement instance's beans.
- Gauges are read lazily at poll time.
- Partial and full commit timers do not pollute each other.

## Future Work

- The `coordinator-progress-*` metrics proposed in
[#16389](https://github.com/apache/iceberg/issues/16389) for coordinator/worker backpressure.
- Per-table commit metrics (the coordinator commits to N tables per cycle; per-table latency and
failure attribution would help isolate a single slow table).
- Surfacing the same signals through Iceberg's `MetricsReporter` (see **Q9**).

## References

- PR:
- Related proposal:
- KIP-410, as the model for a metrics-only proposal:

- Iceberg snapshot-summary naming precedent: `core/src/main/java/org/apache/iceberg/SnapshotSummary.java`

### Proposal document

_No response_

### Specifications

- [ ] Table
- [ ] View
- [ ] REST
- [ ] Puffin
- [ ] Encryption
- [x] Other

Start with PR #17025 and the linked discussion thread, then inspect ChannelMetrics, WorkerMetrics, CoordinatorMetrics, and Channel in org.apache.iceberg.connect.channel. The proposed work is complete in the PR; this issue's remaining outcome is community agreement on the JMX domain, metric names, tag keys, and open questions before merge.

Written by the indexing model from the issue text.