HenryCaiHaiying opened a new issue, #18007: URL: https://github.com/apache/iceberg/issues/18007
### 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 _TBD — link the lists.apache.org thread here once started._ ## 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: ``` <domain>:type=<group>[,<tag>=<value>]* ``` 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=<connector>,task=<taskId>` | one per sink task | | `iceberg.kafka.connect:type=coordinator-metrics,connector=<connector>,task=coordinator` | one per connector | | `iceberg.kafka.connect:type=coordinator-metrics,connector=<connector>,task=coordinator,commit-mode=full` | one per connector | | `iceberg.kafka.connect:type=coordinator-metrics,connector=<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=<connector>,task=<taskId>` | 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=<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=<connector>,task=coordinator,commit-mode=<full\|partial>` | 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: <https://github.com/apache/iceberg/pull/17025> - Related proposal: <https://github.com/apache/iceberg/issues/16389> - KIP-410, as the model for a metrics-only proposal: <https://cwiki.apache.org/confluence/spaces/KAFKA/pages/97555418/KIP-410+Add+metric+for+request+handler+thread+pool+utilization+by+request+type> - 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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
