You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "HenryCaiHaiying (via GitHub)" <gi...@apache.org> on 2026/07/01 01:00:04 UTC
[PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
HenryCaiHaiying opened a new pull request, #17025:
URL: https://github.com/apache/iceberg/pull/17025
Expose runtime visibility into the Iceberg Kafka Connect commit pipeline using kafka-clients' built-in Metrics framework with a JmxReporter (no new dependencies).
Two package-private classes own per-task and per-connector Metrics registries:
WorkerMetrics — tags {connector, task}
CoordinatorMetrics — tag {connector}
Exposed MBeans under the iceberg-kafka-connect-metrics domain:
type=worker-metrics
save-time-{avg,max,total}
consume-available-time-{avg,max,total}
data-written-total
data-complete-total
type=coordinator-metrics
commit-time-{avg,max,total}
consume-available-time-{avg,max,total}
start-commit-total
commit-complete-total
commit-buffer-size (gauge, lazy)
ready-buffer-size (gauge, lazy)
Wiring:
- Channel.consumeAvailable wraps its body in a timer and calls a new recordConsumeTime(long) hook overridden by Worker and Coordinator. The initial poll inside Channel.start() is timed too.
- Worker times save() and counts DATA_WRITTEN/DATA_COMPLETE emissions; closes its registry in stop().
- Coordinator times commit() (covering doCommit + endCurrentCommit on both success and failure paths), counts START_COMMIT/ COMMIT_COMPLETE emissions, and exposes lazy-supplier gauges backed by CommitState.commitBufferSize() / readyBufferSize(); closes its registry in terminate().
- CommitState gains commitBufferSize() / readyBufferSize() accessors.
Construction is hardened against partial-init MBean leaks: the Metrics registry is built into a local, sensors and gauges are registered inside a try block, and on RuntimeException the local registry is closed before rethrowing.
Tests: TestWorkerMetrics (4 tests), TestCoordinatorMetrics (5 tests) verify observable JMX behavior via the live MBeanServer. Registration assertions added to TestWorker and TestCoordinator. ChannelTestBase gains default config.connectorName()/taskId() stubs so existing Worker/Coordinator tests keep constructing without NPE inside the JMX tag map. TestCoordinator's coordinatorTest() helper now terminates and stops the Coordinator in a finally block to release the CoordinatorMetrics registry between tests.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3696518195
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -192,6 +205,7 @@ private void commit(boolean partialCommit) {
e);
} finally {
commitState.endCurrentCommit();
+ coordinatorMetrics.recordCommit(partialCommit, (System.nanoTime() - start) / 1_000L);
Review Comment:
The sensor is to record the total time spent on commit activities whether the final result is successful or not, I can update the sensor description to reflect this. But if you feel it is important to distinguish between the time spend on successful case vs failure cases, I can also add a tag to the sensor to distinguish between the two.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279164096
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -125,11 +128,16 @@ record -> {
// so increment the record offset by one
controlTopicOffsets.put(record.partition(), record.offset() + 1);
+ long readStart = System.nanoTime();
Event event = AvroUtil.decode(record.value());
+ getChannelMetrics().recordMessageRead((System.nanoTime() - readStart) / 1_000L);
if (event.groupId().equals(connectGroupId)) {
LOG.debug("Received event of type: {}", event.type().name());
- if (receive(new Envelope(event, record.partition(), record.offset()))) {
+ long processStart = System.nanoTime();
+ boolean handled = receive(new Envelope(event, record.partition(), record.offset()));
+ getChannelMetrics().recordMessageProcess((System.nanoTime() - processStart) / 1_000L);
Review Comment:
I don't see this is a conflict (or double counting) between 2 metrics. channel-message-process-time is on the outer parent function measuring the time spent on processing an incoming Kafka message, commit-time is on the child function measuring the time spent on commit. The parent function can potentially invoke the child function to process commit, when that happens the time spent on channel-message-process-time will increase as well. The 2 metrics are measuring their respective measurements, there is no specification dictating that any 2 time metrics need to be unique and non-overlapping. If we put in this constraint, any future metrics in a inner function needs to recursively check all parent functions to see whether there is time metrics on the outer layer and try to break up the code to make the 2 metrics non-overlapping in any combination of code paths, we are not going to be able to maintain the code like that.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3612505597
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -125,11 +128,17 @@ record -> {
// so increment the record offset by one
controlTopicOffsets.put(record.partition(), record.offset() + 1);
+ long readStart = System.nanoTime();
Event event = AvroUtil.decode(record.value());
+ getChannelMetrics().recordMessageRead((System.nanoTime() - readStart) / 1_000_000L);
Review Comment:
I can change the timer sensor to use microsecond as unit. But I think I will need to update all the time sensors to use microsecond: channel-message-read-time, channel-message-process-time, Coordinator's full-commit-time and partial-commit-time, Worker's save-time, otherwise it will be confusing and error-prone to the users.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562741497
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class WorkerMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerMetrics.class);
+ private static final String GROUP = "worker-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor saveTime;
+ private final Sensor consumeTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String taskId) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", taskId);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.saveTime =
+ createTimerSensor(newMetrics, "save-time", "Time spent in Worker.save() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (worker side) in ms",
+ tags);
+ this.dataWritten =
+ createCounterSensor(
+ newMetrics, "data-written", "Number of DATA_WRITTEN events emitted", tags);
+ this.dataComplete =
+ createCounterSensor(
+ newMetrics, "data-complete", "Number of DATA_COMPLETE events emitted", tags);
+ } catch (RuntimeException e) {
+ try {
+ newMetrics.close();
+ } catch (Exception suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+ this.metrics = newMetrics;
+ }
+
+ void recordSave(long elapsedMs) {
+ saveTime.record((double) elapsedMs);
+ }
+
+ void recordConsume(long elapsedMs) {
+ consumeTime.record((double) elapsedMs);
+ }
+
+ void incDataWritten(long count) {
+ if (count > 0) {
Review Comment:
I will remove the 'if' check so it can record 0 for dataWritten metric
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3696499797
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final Metrics metrics;
+ private final String group;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
Review Comment:
Good catch. But I don' think extend the sample window is useful here because 1. the customer can configure different commit interval; 2. the commit interval is not precise since there can be other code execution increase/decrease the commit interval; I don't want the sample window size coupled with the code execution logic.
In fact, I don't think I need the Avg and Max Stat for the timerSensor, it measures the avg/max within the 30 second window, it's not very meaningful. I already have the '-total' and '-count' CumulativeSum stats which is not affected by the sample window size, the user can easily extrapolate the rate or sum from these 2 numbers. So I will just remove the Avg/Max stats from the timerSensor.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3612602740
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,14 @@ void clearResponses() {
commitBuffer.clear();
}
+ int commitBufferSize() {
+ return commitBuffer.size();
Review Comment:
ArrayList.size() method will just return the internal size member variable, although it is not defined with volatile but there is no danger of reading an integer value half-way. The issue is on memory barrier visibility. So the JMX poll thread can read older stale value, it does not seem to me a serious issue for JMX reporting. For ArrayList.clear() and add(), they are updating 'size' variable either in the middle or towards the end of the method call, it still seems the danger is the other reader thread can read an old size variable value, not on reading an inconsistent list.
I think I am fine with adding AtomicInteger variables make the metric reading more accurate, it will add some code complexity on maintaining the consistency between several member variables within CommitState, let me know if you feel strongly about this one.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562731625
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -110,16 +118,25 @@ protected boolean receive(Envelope envelope) {
send(events, results.sourceOffsets());
+ workerMetrics.incDataWritten(results.writerResults().size());
Review Comment:
I will sum the dataFiles().size() and deleteFiles.size() to match the metrics file count description.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "AnatolyPopov (via GitHub)" <gi...@apache.org>.
AnatolyPopov commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3559049353
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor commitTime;
+ private final Sensor consumeTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.commitTime =
+ createTimerSensor(
+ newMetrics, "commit-time", "Time spent in Coordinator.commit() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (coordinator side) in ms",
+ tags);
+ this.startCommit =
+ createCounterSensor(
+ newMetrics, "start-commit", "Number of START_COMMIT events emitted", tags);
+ this.commitComplete =
+ createCounterSensor(
+ newMetrics, "commit-complete", "Number of COMMIT_COMPLETE events emitted", tags);
+
+ newMetrics.addMetric(
+ new MetricName(
+ "commit-buffer-size", GROUP, "Current size of CommitState.commitBuffer", tags),
+ (Gauge<Long>) (config, now) -> commitBufferSize.get());
Review Comment:
@HenryCaiHaiying unless you are going to make these instance variables volatile nothing will be changed I think. And it will just bring additional complexity because they will have to be properly updated.
I think commitBuffer.size() is safe enough because it returns a plain precomputed int value that is the same non-volatile int read and in the worst case it might be slightly stale.
So I would propose then either keep it as it is since it's a best effort metric and slightly stale values do not change much OR make volatile instance variables if we really want a memory barrier. But in the latter case they need to be always carefully updated.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3890720818
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,14 @@ void clearResponses() {
commitBuffer.clear();
}
+ int commitBufferSize() {
+ return commitBuffer.size();
Review Comment:
I flagged this back in round 2 and then approved thinking it got handled — but reading it again, the gauge still reads `commitBuffer.size()` straight off the `ArrayList` from the JMX poll thread while the coordinator thread mutates the list, no happens-before between them.
In practice `size()` is a single int read so worst case is a stale value, which is arguably fine for a monitoring gauge. But it's still technically a race, and it's the one thing from that round I'd rather close than leave implicit — an `AtomicInteger` bumped alongside the add/clear, or even just a comment documenting that a stale read is acceptable, would do it.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -125,11 +128,16 @@ record -> {
// so increment the record offset by one
controlTopicOffsets.put(record.partition(), record.offset() + 1);
+ long readStart = System.nanoTime();
Event event = AvroUtil.decode(record.value());
+ getChannelMetrics().recordMessageRead((System.nanoTime() - readStart) / 1_000L);
if (event.groupId().equals(connectGroupId)) {
LOG.debug("Received event of type: {}", event.type().name());
- if (receive(new Envelope(event, record.partition(), record.offset()))) {
+ long processStart = System.nanoTime();
+ boolean handled = receive(new Envelope(event, record.partition(), record.offset()));
+ getChannelMetrics().recordMessageProcess((System.nanoTime() - processStart) / 1_000L);
Review Comment:
Looking again at the coordinator side of this timer — on the coordinator, `receive()` for `DATA_COMPLETE` calls `commit(false)` inline once the commit is ready, so this process-time sample ends up including the whole catalog round-trip. That same duration is already recorded in `commit-time-total`, so a full commit lands in two metrics at once and `process-time-total` goes bimodal.
It's asymmetric too: the timeout-driven partial commit runs after `consumeAvailable()` returns and isn't counted here, only the full commit from the triggering `DATA_COMPLETE` is. I'd pull the `commit(false)` trigger out of `receive()` so the per-record timer only covers decode/dispatch, the way partial commit already works. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+ // Recorded from a finally block, so it covers commits that failed as well as ones that succeeded.
+ private static final String COMMIT_TIME_DESC =
+ "Time spent in Coordinator.commit() in microseconds, whether the commit succeeded or failed";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
Review Comment:
`start-commit-total` increments after `send()` succeeds, so it's really counting successful START_COMMIT sends, not commit initiations — if a send fails the commit is already in progress (the timeout path picks it up) but this counter stays put. The comment above captures that, but the name doesn't, and correlating start vs complete counts during a partition event is exactly when the gap bites.
Since the name is a public contract once operators wire it up, I'd float this on dev@ rather than just settle it in the PR — `start-commit-sent-total`, or a second counter at the initiation point, are both worth raising there. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+ // Recorded from a finally block, so it covers commits that failed as well as ones that succeeded.
+ private static final String COMMIT_TIME_DESC =
+ "Time spent in Coordinator.commit() in microseconds, whether the commit succeeded or failed";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
+ "Number of successfully emitted START_COMMIT events",
+ metricTags(connector, TASK, null));
+ this.commitComplete =
+ createCounterSensor(
+ "commit-complete",
Review Comment:
`commit-time` carries a `commit-mode=full/partial` tag but `commit-complete-total` is a single untagged counter for both — so the persistent completed-commit count can't separate timeout-driven partial commits from full ones, which are operationally quite different.
The tag schema is part of the contract operators build dashboards against, and adding a dimension after the fact reshuffles their queries, so I'd settle it now. Worth putting on dev@ with the other naming calls — I'd lean toward mirroring the `commit-mode` tag here, but it's a genuine fork. wdyt?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -344,47 +347,52 @@ private UUID coordinatorTest(
SinkTaskContext context = mock(SinkTaskContext.class);
Coordinator coordinator =
new Coordinator(catalog, config, ImmutableList.of(), clientFactory, context);
- coordinator.start();
-
- // init consumer after subscribe()
- initConsumer();
-
- coordinator.process();
-
- assertThat(producer.transactionCommitted()).isTrue();
- assertThat(producer.history()).hasSize(1);
-
- byte[] bytes = producer.history().get(0).value();
- Event commitRequest = AvroUtil.decode(bytes);
- assertThat(commitRequest.type()).isEqualTo(PayloadType.START_COMMIT);
-
- UUID commitId = ((StartCommit) commitRequest.payload()).commitId();
-
- Event commitResponse =
- new Event(
- config.connectGroupId(),
- new DataWritten(
- StructType.of(),
- commitId,
- TableReference.of("catalog", TableIdentifier.of("db", "tbl"), null),
- dataFiles,
- deleteFiles));
- bytes = AvroUtil.encode(commitResponse);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
-
- Event commitReady =
- new Event(
- config.connectGroupId(),
- new DataComplete(
- commitId, ImmutableList.of(new TopicPartitionOffset("topic", 1, 1L, ts))));
- bytes = AvroUtil.encode(commitReady);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", bytes));
-
- when(config.commitIntervalMs()).thenReturn(0);
-
- coordinator.process();
-
- return commitId;
+ try {
+ coordinator.start();
+
+ // init consumer after subscribe()
+ initConsumer();
+
+ coordinator.process();
+
+ assertThat(producer.transactionCommitted()).isTrue();
+ assertThat(producer.history()).hasSize(1);
+
+ byte[] bytes = producer.history().get(0).value();
+ Event commitRequest = AvroUtil.decode(bytes);
+ assertThat(commitRequest.type()).isEqualTo(PayloadType.START_COMMIT);
+
+ UUID commitId = ((StartCommit) commitRequest.payload()).commitId();
+
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db", "tbl"), null),
+ dataFiles,
+ deleteFiles));
+ bytes = AvroUtil.encode(commitResponse);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
+
+ Event commitReady =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new TopicPartitionOffset("topic", 1, 1L, ts))));
+ bytes = AvroUtil.encode(commitReady);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", bytes));
+
+ when(config.commitIntervalMs()).thenReturn(0);
+
+ coordinator.process();
+
+ return commitId;
+ } finally {
+ coordinator.terminate();
Review Comment:
Now that `CoordinatorMetrics` registers its MBeans in the constructor, the three retry tests up top — `testCommitBoundedRetry`, `testCommitCounterResetsOnSuccess`, `testCommitBoundedRetryWithMultipleThreads` — leak them: they build a `Coordinator`, let the `CommitFailedException` propagate, and return without ever hitting `terminate()`/`stop()`. `JmxReporter` silently re-registers so nothing throws, but the `Metrics`/`JmxReporter` never close and a later instance quietly steals the bean under the same `ObjectName`.
The same `try { … } finally { terminate(); stop(); }` you added here would cover them. Not urgent — happy to see it in a follow-up — but worth doing while the pattern's fresh. One small thing on this finally itself: if `terminate()` throws, `stop()` never runs, so I'd nest them as `try { terminate(); } finally { stop(); }`.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -110,16 +117,29 @@ protected boolean receive(Envelope envelope) {
send(events, results.sourceOffsets());
+ long fileCount =
+ results.writerResults().stream()
+ .mapToLong(result -> result.dataFiles().size() + result.deleteFiles().size())
Review Comment:
This sums `dataFiles().size() + deleteFiles().size()` into one `data-files-written-total`. Data and delete files are pretty different beasts — deletes only show up for upsert/CDC — so an operator watching ingest throughput sees inflated numbers (1 data + 1 equality-delete reads as 2).
This is really a naming-contract call: split into `data-files-written` / `delete-files-written`, or rename to `content-files-written` since `ContentFile` is the common supertype. I'd put that choice on dev@ before we commit to a name, since there's no clean way to rename it later without breaking dashboards.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3556896221
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -412,6 +427,8 @@ void terminate() {
}
} catch (InterruptedException e) {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
+ } finally {
+ coordinatorMetrics.close();
Review Comment:
Good catch, will override Coordinator#stop() to close the metrics as well.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562905455
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor commitTime;
+ private final Sensor consumeTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.commitTime =
Review Comment:
There are 2 different kinds of commits (partial vs full commit), it seems make sense to add a tag "mode", value 'partial' or 'full' for the metrics to differentiate time spent between the two.
Also it makes sense to count the number of times that partial/full commit is called, looks like WindowsCount Stat function can achieve this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#issuecomment-6071762523
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4326473144
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final String group;
+ private final Object closeLock = new Object();
+
+ // Nulled out by the first close() so a second close cannot unregister MBeans that a replacement
+ // instance (e.g. a newly elected coordinator in the same JVM) has since registered under the same
+ // ObjectName. Only written under closeLock; volatile so readers see the null without locking.
+ private volatile Metrics metrics;
Review Comment:
This protects against a second close, but not against a late *first* close, and the fenced path is exactly that case. By the time a fenced coordinator's thread notices and runs `stop()` → `close()`, the replacement has usually registered already. `JmxReporter.close()` unregisters by ObjectName without checking who owns it, so a new leader on the same worker loses its connector-level beans and the dashboards go blank with no error. The early close in `terminate()` only covers the handoff the committer drives.
I'd make ownership explicit: the last instance to register owns the name, and `close()` skips the JMX unregister once it no longer does. `testRepeatedCloseDoesNotUnregisterReplacement` should build the replacement before `old`'s first close. As written it only shows that the second close is a no-op. `testCloseAllowsReregistration` passes either way, since `reregister()` silently replaces the bean.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final String group;
+ private final Object closeLock = new Object();
+
+ // Nulled out by the first close() so a second close cannot unregister MBeans that a replacement
+ // instance (e.g. a newly elected coordinator in the same JVM) has since registered under the same
+ // ObjectName. Only written under closeLock; volatile so readers see the null without locking.
+ private volatile Metrics metrics;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.messageReadTime =
+ createTimerSensor(
+ "channel-message-read-time",
+ "Time to Avro-decode one control message in microseconds",
+ tags);
+ this.messageProcessTime =
+ createTimerSensor(
+ "channel-message-process-time",
+ "Time to process one control message in microseconds",
Review Comment:
fair point on the overlap. I called it a correctness issue last round, but nested timers are fine and I don't think restructuring is worth it. What's missing is saying what's inside the timer. On the coordinator, a full commit runs inside `receive(DATA_COMPLETE)`, and a partial commit triggered by the timeout doesn't, so nobody should add this to `commit-time`:
```suggestion
"Time to handle one control message in microseconds, including work it triggers"
+ " (a full commit on the coordinator; timeout-driven partial commits are not"
+ " included)",
```
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -110,16 +117,30 @@ protected boolean receive(Envelope envelope) {
send(events, results.sourceOffsets());
+ long dataFileCount =
+ results.writerResults().stream().mapToLong(result -> result.dataFiles().size()).sum();
+ long deleteFileCount =
+ results.writerResults().stream().mapToLong(result -> result.deleteFiles().size()).sum();
+ workerMetrics.incDataFilesWritten(dataFileCount);
+ workerMetrics.incDeleteFilesWritten(deleteFileCount);
+ workerMetrics.incDataComplete();
+
return true;
}
@Override
void stop() {
super.stop();
sinkWriter.close();
+ workerMetrics.close();
Review Comment:
`sinkWriter.close()` closes open writers, so it can throw. If it does, the beans stay registered until the next `Worker` silently replaces them. It's the same shape as the `terminate(); stop();` test fix from last round, so I'd close in a `finally` here and in `Coordinator.stop()`:
```java
try {
super.stop();
sinkWriter.close();
} finally {
workerMetrics.close();
}
```
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -199,7 +199,15 @@ private void processControlEvents() {
if (coordinatorThread != null && coordinatorThread.isTerminated()) {
if (coordinatorThread.isFenced()) {
LOG.warn("Coordinator on committer {} was fenced by a newer coordinator; clearing", taskId);
- coordinatorThread = null;
+ try {
+ // The thread has already exited and run coordinator.stop(), but only terminate() shuts
+ // down the coordinator's commit thread pool.
+ coordinatorThread.terminate();
Review Comment:
I'd move this commit into its own PR. It fixes a real leak (the fenced path never shut down `exec`). But `stop()` already closes the metrics on that path, so the metrics work doesn't need it, and a change to the fencing path deserves its own review and changelog line.
When it moves: `terminate()` throws `ConnectException` if it's interrupted in `awaitTermination`, and this catch logs and swallows it. Unless `terminate()` already re-sets the interrupt flag, I'd restore it here.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -82,36 +85,61 @@ public void testSave() {
when(sinkWriter.completeWrite()).thenReturn(sinkWriterResult);
Worker worker = new Worker(config, clientFactory, sinkWriter, context);
- worker.start();
-
- // init consumer after subscribe()
- initConsumer();
-
- // save a record
- Map<String, Object> value = ImmutableMap.of();
- SinkRecord rec = new SinkRecord(SRC_TOPIC_NAME, 0, null, "key", null, value, 0L);
- worker.save(ImmutableList.of(rec));
-
- UUID commitId = UUID.randomUUID();
- Event commitRequest = new Event(config.connectGroupId(), new StartCommit(commitId));
- byte[] bytes = AvroUtil.encode(commitRequest);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
-
- worker.process();
-
- assertThat(producer.history()).hasSize(2);
-
- Event event = AvroUtil.decode(producer.history().get(0).value());
- assertThat(event.payload().type()).isEqualTo(PayloadType.DATA_WRITTEN);
- DataWritten dataWritten = (DataWritten) event.payload();
- assertThat(dataWritten.commitId()).isEqualTo(commitId);
+ try {
+ worker.start();
+
+ // init consumer after subscribe()
+ initConsumer();
+
+ // save a record
+ Map<String, Object> value = ImmutableMap.of();
+ SinkRecord rec = new SinkRecord(SRC_TOPIC_NAME, 0, null, "key", null, value, 0L);
+ worker.save(ImmutableList.of(rec));
+
+ UUID commitId = UUID.randomUUID();
+ Event commitRequest = new Event(config.connectGroupId(), new StartCommit(commitId));
+ byte[] bytes = AvroUtil.encode(commitRequest);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
+
+ worker.process();
+
+ assertThat(producer.history()).hasSize(2);
+
+ Event event = AvroUtil.decode(producer.history().get(0).value());
+ assertThat(event.payload().type()).isEqualTo(PayloadType.DATA_WRITTEN);
+ DataWritten dataWritten = (DataWritten) event.payload();
+ assertThat(dataWritten.commitId()).isEqualTo(commitId);
+
+ event = AvroUtil.decode(producer.history().get(1).value());
+ assertThat(event.type()).isEqualTo(PayloadType.DATA_COMPLETE);
+ DataComplete dataComplete = (DataComplete) event.payload();
+ assertThat(dataComplete.commitId()).isEqualTo(commitId);
+ assertThat(dataComplete.assignments()).hasSize(1);
+ assertThat(dataComplete.assignments().get(0).offset()).isEqualTo(1L);
Review Comment:
No test checks the call sites. `TestChannel` and `TestControlTopicReplay` mock `ChannelMetrics`, and the metrics tests call `inc*` and `record*` directly. I'd assert the real `WorkerMetrics` counters here, using different data-file and delete-file counts so a swapped split would show up. I'd do the same for `commit-complete` by `commit-mode` in `coordinatorTest`. Those splits are what changed last round, so that's where a wiring slip would hide.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3557009135
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor commitTime;
+ private final Sensor consumeTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.commitTime =
+ createTimerSensor(
+ newMetrics, "commit-time", "Time spent in Coordinator.commit() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (coordinator side) in ms",
+ tags);
+ this.startCommit =
+ createCounterSensor(
+ newMetrics, "start-commit", "Number of START_COMMIT events emitted", tags);
+ this.commitComplete =
+ createCounterSensor(
+ newMetrics, "commit-complete", "Number of COMMIT_COMPLETE events emitted", tags);
+
+ newMetrics.addMetric(
+ new MetricName(
+ "commit-buffer-size", GROUP, "Current size of CommitState.commitBuffer", tags),
+ (Gauge<Long>) (config, now) -> commitBufferSize.get());
Review Comment:
I guess there is a danger on commitBuffer is not thread safe, calling commitBuffer.size() from another thread might return incorrect result.
I think I can solve this problem by adding 2 integer instance variables: commitBufferSize and readyBufferSize in CommitState object. Those size variables will be incremented each time the commitBuffer/readyBuffer is adding one element. Reading integer variables across threads should be atomic.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3567167629
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -125,11 +128,17 @@ record -> {
// so increment the record offset by one
controlTopicOffsets.put(record.partition(), record.offset() + 1);
+ long readStart = System.nanoTime();
Event event = AvroUtil.decode(record.value());
+ getChannelMetrics().recordMessageRead((System.nanoTime() - readStart) / 1_000_000L);
Review Comment:
The per-record timers I asked about last round are exactly what I wanted here — one snag though: `/ 1_000_000L` is integer division, so any decode faster than a millisecond floors to 0, and Avro-decoding a single control message almost always is. Same on `recordMessageProcess` at line 140.
So in practice these two read 0 nearly all the time. I'd record microseconds (`/ 1_000L`) and tweak the description, or pass the raw nanos as a double so avg/max stay meaningful.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -123,6 +129,7 @@ void process() {
Event event =
new Event(config.connectGroupId(), new StartCommit(commitState.currentCommitId()));
send(event);
+ coordinatorMetrics.incStartCommit();
Review Comment:
Small semantics question on the counter placement, in the same spirit as the count discussion last round: `incStartCommit()` sits after `send()`, so if `send()` throws the commit is already in progress but the counter never moves. Same shape in `Worker.receive()` — `incDataWritten`/`incDataComplete` run after `send()`, so a failed send means the files are already on disk but the counters show nothing.
If we're strictly counting "events successfully emitted", after-send is right and I'd just leave a note saying so. If we want "commits initiated" / "files written", I'd move them ahead of `send()` or into a finally. Either's defensible — I'd just pick one on purpose. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,14 @@ void clearResponses() {
commitBuffer.clear();
}
+ int commitBufferSize() {
+ return commitBuffer.size();
Review Comment:
This is the one thing from last round I still don't see addressed — the gauge suppliers read `commitBuffer.size()`/`readyBuffer.size()` from the JMX poll thread, but the buffers are plain `ArrayList`s mutated on the coordinator thread with no happens-before between the two.
Best case the gauge just lags a cycle, which is fine; worst case a `size()` read races an `add()` array-grow or a `clear()` that nulls elements before resetting the size field, which is a real (if rare) window. I'd back these with an `AtomicInteger` in `CommitState` bumped alongside the add/clear calls, or read them under the same lock the coordinator mutates under — either's fine, I just don't want the cross-thread `ArrayList.size()`. wdyt?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java:
##########
@@ -102,6 +102,8 @@ public void before() {
when(config.commitThreads()).thenReturn(1);
when(config.connectGroupId()).thenReturn(CONNECT_CONSUMER_GROUP_ID);
when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(config.connectorName()).thenReturn("test-connector");
+ when(config.taskId()).thenReturn("0");
Review Comment:
Now that `before()` pins a fixed connector/task and both the `Worker` and `Coordinator` constructors register MBeans, any test that builds one off this config and doesn't `stop()` it leaves that bean registered for the rest of the JVM — the next construction with the same name throws `InstanceAlreadyExistsException`, and JUnit gives us no ordering guarantee.
`coordinatorTest()` already got the right treatment with the `terminate()`/`stop()` finally. `testSave` in `TestWorker` and `testPartialCommitFailureMetric` in `TestCoordinator` still construct directly without it — I'd wrap those the same way.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3552788884
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinatorMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.lang.management.ManagementFactory;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import org.junit.jupiter.api.Test;
+
+public class TestCoordinatorMetrics {
+
+ private static final String CONNECTOR = "test-connector";
Review Comment:
Every test in here shares `CONNECTOR = "test-connector"`, which becomes the JMX `ObjectName` — so under JUnit5 parallel execution they collide on the same MBean and throw `InstanceAlreadyExistsException`, and this also overlaps the `"test-connector"` now set in `ChannelTestBase`.
I'd give each test a unique connector name (a UUID, or derive it from the test method) and thread it through the read helpers so they stay isolated. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class WorkerMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerMetrics.class);
+ private static final String GROUP = "worker-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor saveTime;
+ private final Sensor consumeTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String taskId) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", taskId);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.saveTime =
+ createTimerSensor(newMetrics, "save-time", "Time spent in Worker.save() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (worker side) in ms",
+ tags);
+ this.dataWritten =
+ createCounterSensor(
+ newMetrics, "data-written", "Number of DATA_WRITTEN events emitted", tags);
+ this.dataComplete =
+ createCounterSensor(
+ newMetrics, "data-complete", "Number of DATA_COMPLETE events emitted", tags);
+ } catch (RuntimeException e) {
+ try {
+ newMetrics.close();
+ } catch (Exception suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+ this.metrics = newMetrics;
+ }
+
+ void recordSave(long elapsedMs) {
+ saveTime.record((double) elapsedMs);
+ }
+
+ void recordConsume(long elapsedMs) {
+ consumeTime.record((double) elapsedMs);
+ }
+
+ void incDataWritten(long count) {
+ if (count > 0) {
Review Comment:
`incDataWritten` is guarded by `count > 0` but `incDataComplete` records unconditionally, so on empty poll cycles `data-complete-total` climbs past `data-written-total` — which reads like data loss to anyone watching the ratio. I'd guard both the same way (or record 0 for both) so they stay comparable.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,12 +59,19 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
+ this.workerMetrics =
Review Comment:
`WorkerMetrics` is constructed after the `Channel` super() has already stood up the producer/consumer/admin clients, so if this constructor throws, those clients leak (the producer's background sender thread in particular). I'd wrap it so a failure here calls `super.stop()` before rethrowing — same shape on the coordinator side.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -133,6 +140,11 @@ void process() {
}
}
+ @Override
+ protected void recordConsumeTime(long elapsedMs) {
+ coordinatorMetrics.recordConsume(elapsedMs);
Review Comment:
Two things pollute this timer on the coordinator side and I think they make it hard to read. The coordinator calls `consumeAvailable(Duration.ofSeconds(1))`, so an idle coordinator reports ~1000ms every cycle and a busy one reports ~1000ms plus a few — the poll wait dominates and swamps the signal. On top of that, `receive()` runs `commit(false)` synchronously inside this window, so `consume-available-time` double-counts the full commit latency that `commit-time` already captures.
The worker side is fine since it polls with `Duration.ZERO`. I'd either measure only the record-processing portion (exclude the blocking poll), or at minimum document that the coordinator value includes both the poll duration and commit time. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor commitTime;
+ private final Sensor consumeTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.commitTime =
+ createTimerSensor(
+ newMetrics, "commit-time", "Time spent in Coordinator.commit() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (coordinator side) in ms",
+ tags);
+ this.startCommit =
+ createCounterSensor(
+ newMetrics, "start-commit", "Number of START_COMMIT events emitted", tags);
+ this.commitComplete =
+ createCounterSensor(
+ newMetrics, "commit-complete", "Number of COMMIT_COMPLETE events emitted", tags);
+
+ newMetrics.addMetric(
+ new MetricName(
+ "commit-buffer-size", GROUP, "Current size of CommitState.commitBuffer", tags),
+ (Gauge<Long>) (config, now) -> commitBufferSize.get());
Review Comment:
`commitBuffer` and `readyBuffer` are plain `ArrayList`s mutated on `CoordinatorThread`, and these gauges read their size from the `JmxReporter` thread with no memory barrier — so the reported sizes can be stale or torn.
I'd back them with an `AtomicInteger` (or a volatile size updated under `CommitState`'s existing lock) rather than reading `size()` cross-thread. Thoughts?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -371,4 +376,25 @@ public void testCoordinatorCommittedOffsetValidation() {
assertThat(table.snapshots()).hasSize(2);
assertThat(table.currentSnapshot().summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
}
+
+ @Test
+ public void testCoordinatorRegistersJmxMetrics() throws Exception {
+ when(config.connectorName()).thenReturn("jmx-coord-connector");
+
+ SinkTaskContext taskContext = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory, taskContext);
+ try {
+ javax.management.MBeanServer server =
+ java.lang.management.ManagementFactory.getPlatformMBeanServer();
+ javax.management.ObjectName name =
+ new javax.management.ObjectName(
+ "iceberg-kafka-connect-metrics:type=coordinator-metrics,"
+ + "connector=jmx-coord-connector");
+ assertThat(server.queryNames(name, null)).isNotEmpty();
Review Comment:
This asserts the MBeans are present while the coordinator is alive but never that they're gone after close — so a regression that leaks the registration would pass here and only surface as `InstanceAlreadyExistsException` on restart. The dedicated unit tests check both sides; I'd add a `queryNames(...).isEmpty()` assertion after the `finally` here (and in the `TestWorker` equivalent) too.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class WorkerMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerMetrics.class);
+ private static final String GROUP = "worker-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor saveTime;
+ private final Sensor consumeTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String taskId) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", taskId);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.saveTime =
+ createTimerSensor(newMetrics, "save-time", "Time spent in Worker.save() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (worker side) in ms",
+ tags);
+ this.dataWritten =
+ createCounterSensor(
+ newMetrics, "data-written", "Number of DATA_WRITTEN events emitted", tags);
+ this.dataComplete =
+ createCounterSensor(
+ newMetrics, "data-complete", "Number of DATA_COMPLETE events emitted", tags);
+ } catch (RuntimeException e) {
+ try {
+ newMetrics.close();
+ } catch (Exception suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+ this.metrics = newMetrics;
+ }
+
+ void recordSave(long elapsedMs) {
+ saveTime.record((double) elapsedMs);
+ }
+
+ void recordConsume(long elapsedMs) {
+ consumeTime.record((double) elapsedMs);
+ }
+
+ void incDataWritten(long count) {
+ if (count > 0) {
+ dataWritten.record((double) count);
+ }
+ }
+
+ void incDataComplete() {
+ dataComplete.record(1);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing WorkerMetrics", e);
+ }
+ }
+
+ private Sensor createTimerSensor(
Review Comment:
`createTimerSensor`/`createCounterSensor` are byte-for-byte identical between `WorkerMetrics` and `CoordinatorMetrics`, and `NAMESPACE` is duplicated across both (plus the test `ObjectName` strings).
Could we lift the helpers and the constant into a small shared base or util? Not blocking — it's just going to drift otherwise, and a rename shouldn't have to touch four places.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
Review Comment:
Minor, but the JMX domain uses hyphens (`iceberg-kafka-connect-metrics`) rather than Kafka's dotted reverse-DNS convention, so it shows up as a flat node in jconsole next to the dotted `kafka.connect` beans. `iceberg.kafka.connect` would sit more naturally alongside them. Your call.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -412,6 +427,8 @@ void terminate() {
}
} catch (InterruptedException e) {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
+ } finally {
+ coordinatorMetrics.close();
Review Comment:
`awaitTermination` drains the commit executor, but `commit()` — and its `finally` that calls `recordCommit` — runs on `CoordinatorThread`, not the pool. So this `close()` can land while `CoordinatorThread` is still in that `finally`, recording against a closed `Metrics`.
A volatile `closed` guard on the `record*` methods, or closing metrics only after the loop exits, would shut the window. Worth a line documenting that `terminate()` owns the metrics lifecycle either way.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -371,4 +376,25 @@ public void testCoordinatorCommittedOffsetValidation() {
assertThat(table.snapshots()).hasSize(2);
assertThat(table.currentSnapshot().summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
}
+
+ @Test
+ public void testCoordinatorRegistersJmxMetrics() throws Exception {
+ when(config.connectorName()).thenReturn("jmx-coord-connector");
+
+ SinkTaskContext taskContext = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory, taskContext);
+ try {
+ javax.management.MBeanServer server =
Review Comment:
These use fully-qualified `javax.management.*` / `java.lang.management.*` inline while the dedicated unit test files import them — I'd add the imports and use simple names here (and in `TestWorker`) for consistency.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -110,16 +118,25 @@ protected boolean receive(Envelope envelope) {
send(events, results.sourceOffsets());
+ workerMetrics.incDataWritten(results.writerResults().size());
Review Comment:
`results.writerResults()` is one entry per table, not per file, so for a single-table connector this counter is effectively 1 per commit regardless of how much data actually moved — which defeats using it to spot stalls. The description "Number of DATA_WRITTEN events emitted" is only accidentally accurate.
I'd either rename it to something like `data-written-tables` and fix the description, or sum `dataFiles().size() + deleteFiles().size()` across the results for a real file count. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -117,27 +117,36 @@ protected void send(List<Event> events, Map<TopicPartition, Offset> sourceOffset
protected abstract boolean receive(Envelope envelope);
protected void consumeAvailable(Duration pollDuration) {
- ConsumerRecords<String, byte[]> records = consumer.poll(pollDuration);
- while (!records.isEmpty()) {
- records.forEach(
- record -> {
- // the consumer stores the offsets that corresponds to the next record to consume,
- // so increment the record offset by one
- controlTopicOffsets.put(record.partition(), record.offset() + 1);
-
- Event event = AvroUtil.decode(record.value());
-
- if (event.groupId().equals(connectGroupId)) {
- LOG.debug("Received event of type: {}", event.type().name());
- if (receive(new Envelope(event, record.partition(), record.offset()))) {
- LOG.info("Handled event of type: {}", event.type().name());
+ long start = System.currentTimeMillis();
Review Comment:
`System.currentTimeMillis()` measures wall-clock, so an NTP or DST slew mid-commit can make `now - start` negative. That negative flows into `CumulativeSum` and `Max`, which permanently corrupts `commit-time-total` and the high-water mark for the rest of the task's life.
I'd switch both captures at all three sites (here, `Coordinator.commit`, `Worker.save`) to `System.nanoTime()` and divide by `1_000_000L`. wdyt?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3557105704
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -371,4 +376,25 @@ public void testCoordinatorCommittedOffsetValidation() {
assertThat(table.snapshots()).hasSize(2);
assertThat(table.currentSnapshot().summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
}
+
+ @Test
+ public void testCoordinatorRegistersJmxMetrics() throws Exception {
+ when(config.connectorName()).thenReturn("jmx-coord-connector");
+
+ SinkTaskContext taskContext = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory, taskContext);
+ try {
+ javax.management.MBeanServer server =
Review Comment:
Will do.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3787822615
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final Metrics metrics;
+ private final String group;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.messageReadTime =
+ createTimerSensor(
+ "channel-message-read-time",
+ "Time to Avro-decode one control message in microseconds",
+ tags);
+ this.messageProcessTime =
+ createTimerSensor(
+ "channel-message-process-time",
+ "Time to process one control message in microseconds",
+ tags);
+ } catch (RuntimeException e) {
+ closeQuietly(e);
+ throw e;
+ }
+ }
+
+ void recordMessageRead(long elapsedMs) {
+ messageReadTime.record((double) elapsedMs);
+ }
+
+ void recordMessageProcess(long elapsedMs) {
+ messageProcessTime.record((double) elapsedMs);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing {}", getClass().getSimpleName(), e);
+ }
+ }
+
+ /**
+ * Closes the registry while an in-flight constructor failure is unwinding, to avoid MBean leaks.
+ */
+ protected void closeQuietly(RuntimeException failure) {
+ try {
+ metrics.close();
+ } catch (Exception suppressed) {
+ failure.addSuppressed(suppressed);
+ }
+ }
+
+ /** Registers a lazily-evaluated gauge that reads {@code supplier} each time JMX polls it. */
+ protected void addGauge(
+ String name, String description, Map<String, String> tags, Supplier<Long> supplier) {
+ metrics.addMetric(
+ new MetricName(name, group, description, tags),
+ (Gauge<Long>) (config, now) -> supplier.get());
+ }
+
+ protected Sensor createTimerSensor(
+ String baseName, String description, Map<String, String> tags) {
+ Sensor sensor = metrics.sensor(sensorName(baseName, tags));
+ sensor.add(new MetricName(baseName + "-avg", group, description + " (avg)", tags), new Avg());
+ sensor.add(new MetricName(baseName + "-max", group, description + " (max)", tags), new Max());
+ sensor.add(
+ new MetricName(baseName + "-total", group, description + " (total)", tags),
+ new CumulativeSum());
+ sensor.add(
+ new MetricName(baseName + "-count", group, description + " (count)", tags),
+ new CumulativeCount());
+ return sensor;
+ }
+
+ protected Sensor createCounterSensor(
+ String baseName, String description, Map<String, String> tags) {
+ Sensor sensor = metrics.sensor(sensorName(baseName, tags));
+ sensor.add(new MetricName(baseName + "-total", group, description, tags), new CumulativeSum());
+ return sensor;
+ }
+
+ /**
+ * Builds the JMX tag map shared by worker and coordinator metrics. {@code commitMode} is only set
+ * on the coordinator's per-commit sensors ({@code full}/{@code partial}); pass {@code null} to
+ * omit it.
+ */
+ static Map<String, String> metricTags(String connector, String task, String commitMode) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", task);
+ if (commitMode != null) {
+ tags.put("commitMode", commitMode);
Review Comment:
Yes commit-mode kebab-case makes sense.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562567814
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor commitTime;
+ private final Sensor consumeTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.commitTime =
+ createTimerSensor(
+ newMetrics, "commit-time", "Time spent in Coordinator.commit() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (coordinator side) in ms",
+ tags);
+ this.startCommit =
+ createCounterSensor(
+ newMetrics, "start-commit", "Number of START_COMMIT events emitted", tags);
+ this.commitComplete =
+ createCounterSensor(
+ newMetrics, "commit-complete", "Number of COMMIT_COMPLETE events emitted", tags);
+
+ newMetrics.addMetric(
+ new MetricName(
+ "commit-buffer-size", GROUP, "Current size of CommitState.commitBuffer", tags),
+ (Gauge<Long>) (config, now) -> commitBufferSize.get());
Review Comment:
Agree. ArrayList.size() is return the value of its own instance integer variable `size` which is fine for our case (might be a little stale but atomic) and both commitBuffer and readyBuffer is not resetting to null in the class so we will not hit NPE either.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562764595
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -371,4 +376,25 @@ public void testCoordinatorCommittedOffsetValidation() {
assertThat(table.snapshots()).hasSize(2);
assertThat(table.currentSnapshot().summary()).containsEntry(OFFSETS_SNAPSHOT_PROP, "{\"0\":7}");
}
+
+ @Test
+ public void testCoordinatorRegistersJmxMetrics() throws Exception {
+ when(config.connectorName()).thenReturn("jmx-coord-connector");
+
+ SinkTaskContext taskContext = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory, taskContext);
+ try {
+ javax.management.MBeanServer server =
+ java.lang.management.ManagementFactory.getPlatformMBeanServer();
+ javax.management.ObjectName name =
+ new javax.management.ObjectName(
+ "iceberg-kafka-connect-metrics:type=coordinator-metrics,"
+ + "connector=jmx-coord-connector");
+ assertThat(server.queryNames(name, null)).isNotEmpty();
Review Comment:
Will do
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3556912262
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -412,6 +427,8 @@ void terminate() {
}
} catch (InterruptedException e) {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
+ } finally {
+ coordinatorMetrics.close();
Review Comment:
I think I am going to override Coordinator#stop() to close the metrics object instead of closing metrics in Coordinator.terminate(). Coordinator#stop() is called at the end of Coordinator's run loop(), this way there is no conflict with Coordinator recording some metrics on a metrics object already closed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "AnatolyPopov (via GitHub)" <gi...@apache.org>.
AnatolyPopov commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3520824781
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -412,6 +427,8 @@ void terminate() {
}
} catch (InterruptedException e) {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
+ } finally {
+ coordinatorMetrics.close();
Review Comment:
I wonder if we need to overwrite `stop` method and close the metrics from there as well, similar to `Worker#stop()`. I believe that covers some more places where metrics needs to be closed.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -412,6 +427,8 @@ void terminate() {
}
} catch (InterruptedException e) {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
+ } finally {
+ coordinatorMetrics.close();
Review Comment:
You are right but as far as I can see the only thing that might happen during recording against closed metrics object is that the update will be lost, or am I missing anything? There does not seem to be any error thrown in this case.
If my understanding is correct then the guard seems to be an overkill since a lost metrics update does not seem to matter during shutdown?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class WorkerMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerMetrics.class);
+ private static final String GROUP = "worker-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor saveTime;
+ private final Sensor consumeTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String taskId) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", taskId);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.saveTime =
+ createTimerSensor(newMetrics, "save-time", "Time spent in Worker.save() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (worker side) in ms",
+ tags);
+ this.dataWritten =
+ createCounterSensor(
+ newMetrics, "data-written", "Number of DATA_WRITTEN events emitted", tags);
+ this.dataComplete =
+ createCounterSensor(
+ newMetrics, "data-complete", "Number of DATA_COMPLETE events emitted", tags);
+ } catch (RuntimeException e) {
+ try {
+ newMetrics.close();
+ } catch (Exception suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+ this.metrics = newMetrics;
+ }
+
+ void recordSave(long elapsedMs) {
+ saveTime.record((double) elapsedMs);
+ }
+
+ void recordConsume(long elapsedMs) {
+ consumeTime.record((double) elapsedMs);
+ }
+
+ void incDataWritten(long count) {
+ if (count > 0) {
+ dataWritten.record((double) count);
+ }
+ }
+
+ void incDataComplete() {
+ dataComplete.record(1);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing WorkerMetrics", e);
+ }
+ }
+
+ private Sensor createTimerSensor(
Review Comment:
Agree with this one. Also I foresee the addition of more JMX metrics for the connector, so this utils might be useful in future.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor commitTime;
+ private final Sensor consumeTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.commitTime =
+ createTimerSensor(
+ newMetrics, "commit-time", "Time spent in Coordinator.commit() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (coordinator side) in ms",
+ tags);
+ this.startCommit =
+ createCounterSensor(
+ newMetrics, "start-commit", "Number of START_COMMIT events emitted", tags);
+ this.commitComplete =
+ createCounterSensor(
+ newMetrics, "commit-complete", "Number of COMMIT_COMPLETE events emitted", tags);
+
+ newMetrics.addMetric(
+ new MetricName(
+ "commit-buffer-size", GROUP, "Current size of CommitState.commitBuffer", tags),
+ (Gauge<Long>) (config, now) -> commitBufferSize.get());
Review Comment:
I'd add my 2 cents here: I do not think size can be torn here since int reads are atomic according to JLS. And does it make sense to add more synchronization here for a best-effort gauge metric? I do not think a slightly stale value matters much here.
Also which particular lock you mean by CommitState's existing lock?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -113,4 +113,24 @@ public void testSave() {
assertThat(dataComplete.assignments().get(0).offset()).isEqualTo(1L);
}
}
+
+ @Test
+ public void testWorkerRegistersJmxMetrics() throws Exception {
+ when(config.connectorName()).thenReturn("jmx-test-connector");
+ when(config.taskId()).thenReturn("3");
+
+ SinkTaskContext taskContext = mock(SinkTaskContext.class);
+ Worker worker = new Worker(config, clientFactory, mock(SinkWriter.class), taskContext);
+ try {
+ javax.management.MBeanServer server =
Review Comment:
Could you replace all these FQNs with imports instead?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#issuecomment-5580330105
@laskoviymishka I think start a discussion thread in the community is a good idea. I have started an email discussion thread [here](https://lists.apache.org/thread/n30rqprw6p6wmsqxvot1k4k4j31j2pt6), I also started a [proposal doc](https://github.com/apache/iceberg/issues/18007) on GitHub issues (pretty much followed the Kafka KIP format).
Will address individual review comments a little bit later.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3612458381
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -123,6 +129,7 @@ void process() {
Event event =
new Event(config.connectGroupId(), new StartCommit(commitState.currentCommitId()));
send(event);
+ coordinatorMetrics.incStartCommit();
Review Comment:
I want to count 'events successfully emitted' and will leave a note with the metric description. This way it is consistent with the logging of 'Coordinator {} initiated commit {}' on the next line. Same for Worker.receive()
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3556970555
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CoordinatorMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorMetrics.class);
+ private static final String GROUP = "coordinator-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
Review Comment:
Yes, I looked at the Kafka Connect's convention for MetricsContext, `iceberg.kafka.connect` would fit better.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562880424
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -133,6 +140,11 @@ void process() {
}
}
+ @Override
+ protected void recordConsumeTime(long elapsedMs) {
+ coordinatorMetrics.recordConsume(elapsedMs);
Review Comment:
I will remove the consume-available-time sensor and replace it with 2 smaller time-sensors: channel-message-read-time, channel-message-process-time
There are several steps during consumeAvailable():
1. consumer.poll() -- Kafka consumer has its own metrics covering this portion
2. Avro deserialize the message -- covered by the new sensor channel-message-read-time
3. Process the event/message and generate the output:
a) On Coordinator, the big time chunk is on commit(). -- We have the commit-time metrics for this, so we don't have count it again in channel-message-process-time (but I am OK with doing the double counting here since commit() can be called from partial commit path as well)
b) On Worker, the big time chunk is on preparing the DataWritten and DataComplete events/messages. -- covered by the new sensor channel-message-process-time
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279358911
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -344,47 +347,52 @@ private UUID coordinatorTest(
SinkTaskContext context = mock(SinkTaskContext.class);
Coordinator coordinator =
new Coordinator(catalog, config, ImmutableList.of(), clientFactory, context);
- coordinator.start();
-
- // init consumer after subscribe()
- initConsumer();
-
- coordinator.process();
-
- assertThat(producer.transactionCommitted()).isTrue();
- assertThat(producer.history()).hasSize(1);
-
- byte[] bytes = producer.history().get(0).value();
- Event commitRequest = AvroUtil.decode(bytes);
- assertThat(commitRequest.type()).isEqualTo(PayloadType.START_COMMIT);
-
- UUID commitId = ((StartCommit) commitRequest.payload()).commitId();
-
- Event commitResponse =
- new Event(
- config.connectGroupId(),
- new DataWritten(
- StructType.of(),
- commitId,
- TableReference.of("catalog", TableIdentifier.of("db", "tbl"), null),
- dataFiles,
- deleteFiles));
- bytes = AvroUtil.encode(commitResponse);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
-
- Event commitReady =
- new Event(
- config.connectGroupId(),
- new DataComplete(
- commitId, ImmutableList.of(new TopicPartitionOffset("topic", 1, 1L, ts))));
- bytes = AvroUtil.encode(commitReady);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", bytes));
-
- when(config.commitIntervalMs()).thenReturn(0);
-
- coordinator.process();
-
- return commitId;
+ try {
+ coordinator.start();
+
+ // init consumer after subscribe()
+ initConsumer();
+
+ coordinator.process();
+
+ assertThat(producer.transactionCommitted()).isTrue();
+ assertThat(producer.history()).hasSize(1);
+
+ byte[] bytes = producer.history().get(0).value();
+ Event commitRequest = AvroUtil.decode(bytes);
+ assertThat(commitRequest.type()).isEqualTo(PayloadType.START_COMMIT);
+
+ UUID commitId = ((StartCommit) commitRequest.payload()).commitId();
+
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db", "tbl"), null),
+ dataFiles,
+ deleteFiles));
+ bytes = AvroUtil.encode(commitResponse);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
+
+ Event commitReady =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new TopicPartitionOffset("topic", 1, 1L, ts))));
+ bytes = AvroUtil.encode(commitReady);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", bytes));
+
+ when(config.commitIntervalMs()).thenReturn(0);
+
+ coordinator.process();
+
+ return commitId;
+ } finally {
+ coordinator.terminate();
Review Comment:
Good catch. I will add try/catch for all other tests (looks like there are 6 tests creating Coordinator objects), and do another try/catch for terminate() as well.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279306817
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+ // Recorded from a finally block, so it covers commits that failed as well as ones that succeeded.
+ private static final String COMMIT_TIME_DESC =
+ "Time spent in Coordinator.commit() in microseconds, whether the commit succeeded or failed";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
Review Comment:
I can use the name 'start-commit-sent-total'
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#issuecomment-5061573790
@HenryCaiHaiying - need to resolve conflict.
@AnatolyPopov - will wait for your approval before merge.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#issuecomment-4849323713
This is related to issue: https://github.com/apache/iceberg/issues/17026
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3787973258
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -440,4 +455,13 @@ void terminate() {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
}
}
+
+ @Override
+ void stop() {
+ // stop() runs at the very end of the CoordinatorThread run loop, after process() (and any
+ // commit() recording) can no longer fire, so closing the registry here avoids racing a
+ // record* call against a closed Metrics instance.
+ super.stop();
Review Comment:
Instead of waiting between terminate() and stop() which is precarious, I will also call 'coordinatorMetrics.close();' during terminate(). And in order to prevent 'metric.close()' being called twice, I will just set the metrics to null on the first close() and the second call will check metrics variable. And make metrics instance variable volatile and a synchronized protection too.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3787881865
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final Metrics metrics;
+ private final String group;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.messageReadTime =
+ createTimerSensor(
+ "channel-message-read-time",
+ "Time to Avro-decode one control message in microseconds",
+ tags);
+ this.messageProcessTime =
+ createTimerSensor(
+ "channel-message-process-time",
+ "Time to process one control message in microseconds",
+ tags);
+ } catch (RuntimeException e) {
+ closeQuietly(e);
+ throw e;
+ }
+ }
+
+ void recordMessageRead(long elapsedMs) {
+ messageReadTime.record((double) elapsedMs);
+ }
+
+ void recordMessageProcess(long elapsedMs) {
+ messageProcessTime.record((double) elapsedMs);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing {}", getClass().getSimpleName(), e);
+ }
+ }
+
+ /**
+ * Closes the registry while an in-flight constructor failure is unwinding, to avoid MBean leaks.
+ */
+ protected void closeQuietly(RuntimeException failure) {
+ try {
+ metrics.close();
+ } catch (Exception suppressed) {
+ failure.addSuppressed(suppressed);
+ }
+ }
+
+ /** Registers a lazily-evaluated gauge that reads {@code supplier} each time JMX polls it. */
+ protected void addGauge(
+ String name, String description, Map<String, String> tags, Supplier<Long> supplier) {
+ metrics.addMetric(
+ new MetricName(name, group, description, tags),
+ (Gauge<Long>) (config, now) -> supplier.get());
+ }
+
+ protected Sensor createTimerSensor(
+ String baseName, String description, Map<String, String> tags) {
+ Sensor sensor = metrics.sensor(sensorName(baseName, tags));
+ sensor.add(new MetricName(baseName + "-avg", group, description + " (avg)", tags), new Avg());
+ sensor.add(new MetricName(baseName + "-max", group, description + " (max)", tags), new Max());
+ sensor.add(
+ new MetricName(baseName + "-total", group, description + " (total)", tags),
+ new CumulativeSum());
+ sensor.add(
+ new MetricName(baseName + "-count", group, description + " (count)", tags),
+ new CumulativeCount());
Review Comment:
I will change the description " (count)" to " (number of sampled points)", " (total)" to " (cumulative total)" to make it easier to read/understand.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279318994
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+ // Recorded from a finally block, so it covers commits that failed as well as ones that succeeded.
+ private static final String COMMIT_TIME_DESC =
+ "Time spent in Coordinator.commit() in microseconds, whether the commit succeeded or failed";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
Review Comment:
And update the proposal doc.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279289500
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,14 @@ void clearResponses() {
commitBuffer.clear();
}
+ int commitBufferSize() {
+ return commitBuffer.size();
Review Comment:
If you are worried about stale reads, I can add in a new volatile int variable (commitBufferSize) to go together with commitBuffer updates. There is some performance overhead going to be added because the memory barrier flush each time the commit buffer is changed, but it is probably negligible. I think volatile is a little bit faster than AtomicInteger (they both need to do a memory barrier flush, but volatile doesn't need to do a compare-and-swap).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#issuecomment-6072023181
Still working on it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279336056
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -110,16 +117,29 @@ protected boolean receive(Envelope envelope) {
send(events, results.sourceOffsets());
+ long fileCount =
+ results.writerResults().stream()
+ .mapToLong(result -> result.dataFiles().size() + result.deleteFiles().size())
Review Comment:
I think I would just split into 2 counters: data-files-written and delete-files-written and update the proposal doc. By the way, I haven't see anyone replied to my original email thread on dev@ email alias nor commenting on the proposal doc.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562770908
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinatorMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.lang.management.ManagementFactory;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import org.junit.jupiter.api.Test;
+
+public class TestCoordinatorMetrics {
+
+ private static final String CONNECTOR = "test-connector";
Review Comment:
Adding UUID to connector name is a good idea.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3612449173
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java:
##########
@@ -102,6 +102,8 @@ public void before() {
when(config.commitThreads()).thenReturn(1);
when(config.connectGroupId()).thenReturn(CONNECT_CONSUMER_GROUP_ID);
when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(config.connectorName()).thenReturn("test-connector");
+ when(config.taskId()).thenReturn("0");
Review Comment:
Good catch, I will add stop()/terminate() for testSave() and testPartialCommitFailureMetric()
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3616983140
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,14 @@ void clearResponses() {
commitBuffer.clear();
}
+ int commitBufferSize() {
+ return commitBuffer.size();
Review Comment:
I also traced the Kafka Metrics code a little bit. The CommitBufferSize is a gauge eventually added as KafkaMetric to the registry. KafkaMetric value is synchronize protected on get: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java#L81.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "AnatolyPopov (via GitHub)" <gi...@apache.org>.
AnatolyPopov commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3683052457
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final Metrics metrics;
+ private final String group;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
Review Comment:
I think the default time window with default metric config is 30 seconds + default number of samples is 2. So by default it will cover only 1 minute out 5 min default commit interval. So commit-time-avg/commit-time-max metrics will be showing NaN during 4 out of 5 min commit interval since the obsolete samples are purged when the metrics are read and 30 sec * 2 samples is expired. I think the time window should be tied to commit interval.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -192,6 +205,7 @@ private void commit(boolean partialCommit) {
e);
} finally {
commitState.endCurrentCommit();
+ coordinatorMetrics.recordCommit(partialCommit, (System.nanoTime() - start) / 1_000L);
Review Comment:
Here we are counting together both successful and unsuccessful commits, correct? Should they be separate?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Map;
+import org.apache.kafka.common.metrics.Sensor;
+
+class WorkerMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "worker-metrics";
+
+ private final Sensor saveTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String task) {
+ super(GROUP, connector, task);
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.saveTime =
+ createTimerSensor("save-time", "Time spent in Worker.save() in microseconds", tags);
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the files are already on disk.
+ this.dataWritten =
+ createCounterSensor(
+ "data-written",
Review Comment:
I think in other places in Iceberg the naming convention is slightly different like `added-data-files`, can we do similar thing here or am I misunderstanding something? For example `data-files-written-total` and maybe even separating delete files into separate counter
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -440,4 +455,13 @@ void terminate() {
throw new ConnectException("Interrupted while waiting for coordinator shutdown", e);
}
}
+
+ @Override
+ void stop() {
+ // stop() runs at the very end of the CoordinatorThread run loop, after process() (and any
+ // commit() recording) can no longer fire, so closing the registry here avoids racing a
+ // record* call against a closed Metrics instance.
+ super.stop();
Review Comment:
Could coordinator re-election cause the old coordinator to unregister the replacement MBeans here? stopCoordinator() calls terminate() without waiting for CoordinatorThread to reach this stop(). A newly elected coordinator in the same Connect JVM can therefore register the same connector-level ObjectName first. Kafka’s JmxReporter replaces the existing MBean during registration, but the old reporter still retains that ObjectName and unregisters the replacement when it eventually closes. The replacement registered successfully, but closing the first instance removed it. Should coordinator shutdown wait for the old thread to finish?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final Metrics metrics;
+ private final String group;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.messageReadTime =
+ createTimerSensor(
+ "channel-message-read-time",
+ "Time to Avro-decode one control message in microseconds",
+ tags);
+ this.messageProcessTime =
+ createTimerSensor(
+ "channel-message-process-time",
+ "Time to process one control message in microseconds",
+ tags);
+ } catch (RuntimeException e) {
+ closeQuietly(e);
+ throw e;
+ }
+ }
+
+ void recordMessageRead(long elapsedMs) {
+ messageReadTime.record((double) elapsedMs);
+ }
+
+ void recordMessageProcess(long elapsedMs) {
+ messageProcessTime.record((double) elapsedMs);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing {}", getClass().getSimpleName(), e);
+ }
+ }
+
+ /**
+ * Closes the registry while an in-flight constructor failure is unwinding, to avoid MBean leaks.
+ */
+ protected void closeQuietly(RuntimeException failure) {
+ try {
+ metrics.close();
+ } catch (Exception suppressed) {
+ failure.addSuppressed(suppressed);
+ }
+ }
+
+ /** Registers a lazily-evaluated gauge that reads {@code supplier} each time JMX polls it. */
+ protected void addGauge(
+ String name, String description, Map<String, String> tags, Supplier<Long> supplier) {
+ metrics.addMetric(
+ new MetricName(name, group, description, tags),
+ (Gauge<Long>) (config, now) -> supplier.get());
+ }
+
+ protected Sensor createTimerSensor(
+ String baseName, String description, Map<String, String> tags) {
+ Sensor sensor = metrics.sensor(sensorName(baseName, tags));
+ sensor.add(new MetricName(baseName + "-avg", group, description + " (avg)", tags), new Avg());
+ sensor.add(new MetricName(baseName + "-max", group, description + " (max)", tags), new Max());
+ sensor.add(
+ new MetricName(baseName + "-total", group, description + " (total)", tags),
+ new CumulativeSum());
+ sensor.add(
+ new MetricName(baseName + "-count", group, description + " (count)", tags),
+ new CumulativeCount());
Review Comment:
The description is borrowed from the sensor description, right? So we will get something like
`Time to process one control message in microseconds (count)` if I understand correctly. And this looks a bit unclear what is the actual unit of measurement if read just the description without looking into the code.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor(
+ "commit-time",
+ "Time spent in Coordinator.commit() in microseconds",
+ metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor(
+ "commit-time",
+ "Time spent in Coordinator.commit() in microseconds",
+ metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
+ "Number of successfully emitted START_COMMIT events",
+ metricTags(connector, TASK, null));
+ this.commitComplete =
+ createCounterSensor(
+ "commit-complete",
+ "Number of successfully emitted COMMIT_COMPLETE events",
+ metricTags(connector, TASK, null));
+
+ addGauge(
+ "commit-buffer-size",
+ "Current size of CommitState.commitBuffer",
+ metricTags(connector, TASK, null),
+ commitBufferSize);
+ addGauge(
+ "ready-buffer-size",
+ "Current size of CommitState.readyBuffer",
+ metricTags(connector, TASK, null),
+ readyBufferSize);
+ } catch (RuntimeException e) {
+ closeQuietly(e);
+ throw e;
+ }
+ }
+
+ void recordCommit(boolean partialCommit, long elapsedMs) {
Review Comment:
I think the at least the argument name is misleading since we seem to record microseconds, not milliseconds, right?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/ChannelMetrics.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Gauge;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeCount;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base for the per-task/per-connector JMX metric registries. Owns the {@link Metrics} registry, the
+ * sensor-creation helpers shared by the worker and coordinator, and the two channel-level timers
+ * that every {@link Channel} feeds while draining the control topic.
+ */
+abstract class ChannelMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelMetrics.class);
+
+ /** JMX domain; dotted to sit alongside Kafka's own {@code kafka.connect.*} beans in jconsole. */
+ static final String NAMESPACE = "iceberg.kafka.connect";
+
+ private final Metrics metrics;
+ private final String group;
+
+ private final Sensor messageReadTime;
+ private final Sensor messageProcessTime;
+
+ ChannelMetrics(String group, String connector, String task) {
+ this.group = group;
+ this.metrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.messageReadTime =
+ createTimerSensor(
+ "channel-message-read-time",
+ "Time to Avro-decode one control message in microseconds",
+ tags);
+ this.messageProcessTime =
+ createTimerSensor(
+ "channel-message-process-time",
+ "Time to process one control message in microseconds",
+ tags);
+ } catch (RuntimeException e) {
+ closeQuietly(e);
+ throw e;
+ }
+ }
+
+ void recordMessageRead(long elapsedMs) {
+ messageReadTime.record((double) elapsedMs);
+ }
+
+ void recordMessageProcess(long elapsedMs) {
+ messageProcessTime.record((double) elapsedMs);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing {}", getClass().getSimpleName(), e);
+ }
+ }
+
+ /**
+ * Closes the registry while an in-flight constructor failure is unwinding, to avoid MBean leaks.
+ */
+ protected void closeQuietly(RuntimeException failure) {
+ try {
+ metrics.close();
+ } catch (Exception suppressed) {
+ failure.addSuppressed(suppressed);
+ }
+ }
+
+ /** Registers a lazily-evaluated gauge that reads {@code supplier} each time JMX polls it. */
+ protected void addGauge(
+ String name, String description, Map<String, String> tags, Supplier<Long> supplier) {
+ metrics.addMetric(
+ new MetricName(name, group, description, tags),
+ (Gauge<Long>) (config, now) -> supplier.get());
+ }
+
+ protected Sensor createTimerSensor(
+ String baseName, String description, Map<String, String> tags) {
+ Sensor sensor = metrics.sensor(sensorName(baseName, tags));
+ sensor.add(new MetricName(baseName + "-avg", group, description + " (avg)", tags), new Avg());
+ sensor.add(new MetricName(baseName + "-max", group, description + " (max)", tags), new Max());
+ sensor.add(
+ new MetricName(baseName + "-total", group, description + " (total)", tags),
+ new CumulativeSum());
+ sensor.add(
+ new MetricName(baseName + "-count", group, description + " (count)", tags),
+ new CumulativeCount());
+ return sensor;
+ }
+
+ protected Sensor createCounterSensor(
+ String baseName, String description, Map<String, String> tags) {
+ Sensor sensor = metrics.sensor(sensorName(baseName, tags));
+ sensor.add(new MetricName(baseName + "-total", group, description, tags), new CumulativeSum());
+ return sensor;
+ }
+
+ /**
+ * Builds the JMX tag map shared by worker and coordinator metrics. {@code commitMode} is only set
+ * on the coordinator's per-commit sensors ({@code full}/{@code partial}); pass {@code null} to
+ * omit it.
+ */
+ static Map<String, String> metricTags(String connector, String task, String commitMode) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", task);
+ if (commitMode != null) {
+ tags.put("commitMode", commitMode);
Review Comment:
Should it be commit-mode or similar in kebab-case?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#issuecomment-5066641850
@laskoviymishka conflicts are resolved.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562757580
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.KafkaMetricsContext;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class WorkerMetrics implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerMetrics.class);
+ private static final String GROUP = "worker-metrics";
+ private static final String NAMESPACE = "iceberg-kafka-connect-metrics";
+
+ private final Metrics metrics;
+ private final Sensor saveTime;
+ private final Sensor consumeTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String taskId) {
+ Map<String, String> tags = new LinkedHashMap<>();
+ tags.put("connector", connector);
+ tags.put("task", taskId);
+
+ Metrics newMetrics =
+ new Metrics(
+ new MetricConfig(),
+ Collections.singletonList(new JmxReporter()),
+ Time.SYSTEM,
+ new KafkaMetricsContext(NAMESPACE));
+ try {
+ this.saveTime =
+ createTimerSensor(newMetrics, "save-time", "Time spent in Worker.save() in ms", tags);
+ this.consumeTime =
+ createTimerSensor(
+ newMetrics,
+ "consume-available-time",
+ "Time spent in Channel.consumeAvailable() (worker side) in ms",
+ tags);
+ this.dataWritten =
+ createCounterSensor(
+ newMetrics, "data-written", "Number of DATA_WRITTEN events emitted", tags);
+ this.dataComplete =
+ createCounterSensor(
+ newMetrics, "data-complete", "Number of DATA_COMPLETE events emitted", tags);
+ } catch (RuntimeException e) {
+ try {
+ newMetrics.close();
+ } catch (Exception suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+ this.metrics = newMetrics;
+ }
+
+ void recordSave(long elapsedMs) {
+ saveTime.record((double) elapsedMs);
+ }
+
+ void recordConsume(long elapsedMs) {
+ consumeTime.record((double) elapsedMs);
+ }
+
+ void incDataWritten(long count) {
+ if (count > 0) {
+ dataWritten.record((double) count);
+ }
+ }
+
+ void incDataComplete() {
+ dataComplete.record(1);
+ }
+
+ @Override
+ public void close() {
+ try {
+ metrics.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing WorkerMetrics", e);
+ }
+ }
+
+ private Sensor createTimerSensor(
Review Comment:
Agree. I can create a superclass ChannelMetrics to host those common methods.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3562760367
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -113,4 +113,24 @@ public void testSave() {
assertThat(dataComplete.assignments().get(0).offset()).isEqualTo(1L);
}
}
+
+ @Test
+ public void testWorkerRegistersJmxMetrics() throws Exception {
+ when(config.connectorName()).thenReturn("jmx-test-connector");
+ when(config.taskId()).thenReturn("3");
+
+ SinkTaskContext taskContext = mock(SinkTaskContext.class);
+ Worker worker = new Worker(config, clientFactory, mock(SinkWriter.class), taskContext);
+ try {
+ javax.management.MBeanServer server =
Review Comment:
Will do.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3556915638
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -117,27 +117,36 @@ protected void send(List<Event> events, Map<TopicPartition, Offset> sourceOffset
protected abstract boolean receive(Envelope envelope);
protected void consumeAvailable(Duration pollDuration) {
- ConsumerRecords<String, byte[]> records = consumer.poll(pollDuration);
- while (!records.isEmpty()) {
- records.forEach(
- record -> {
- // the consumer stores the offsets that corresponds to the next record to consume,
- // so increment the record offset by one
- controlTopicOffsets.put(record.partition(), record.offset() + 1);
-
- Event event = AvroUtil.decode(record.value());
-
- if (event.groupId().equals(connectGroupId)) {
- LOG.debug("Received event of type: {}", event.type().name());
- if (receive(new Envelope(event, record.partition(), record.offset()))) {
- LOG.info("Handled event of type: {}", event.type().name());
+ long start = System.currentTimeMillis();
Review Comment:
Good catch, will change to System.nanoTime().
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3557099190
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,12 +59,19 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
+ this.workerMetrics =
Review Comment:
This seems over engineering. When the Worker object (or Coordinator object) cannot be constructed, the exception thrown chain will eventually fail the KafkaConnect's task: https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java#L251
log.error("{} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted", this, t);
If we want to be over cautious, the Channel's constructor also has the same vulnerability, the consumer object and the admin object is created after the producer object. If the consumer object or admin object creation has an exception we will also have a unused producer object left around.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279319977
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+ // Recorded from a finally block, so it covers commits that failed as well as ones that succeeded.
+ private static final String COMMIT_TIME_DESC =
+ "Time spent in Coordinator.commit() in microseconds, whether the commit succeeded or failed";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor("commit-time", COMMIT_TIME_DESC, metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
+ "Number of successfully emitted START_COMMIT events",
+ metricTags(connector, TASK, null));
+ this.commitComplete =
+ createCounterSensor(
+ "commit-complete",
Review Comment:
Yes I can add the commit-mode tag here and update the proposal doc.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3787893872
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/WorkerMetrics.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.Map;
+import org.apache.kafka.common.metrics.Sensor;
+
+class WorkerMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "worker-metrics";
+
+ private final Sensor saveTime;
+ private final Sensor dataWritten;
+ private final Sensor dataComplete;
+
+ WorkerMetrics(String connector, String task) {
+ super(GROUP, connector, task);
+ Map<String, String> tags = metricTags(connector, task, null);
+ try {
+ this.saveTime =
+ createTimerSensor("save-time", "Time spent in Worker.save() in microseconds", tags);
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the files are already on disk.
+ this.dataWritten =
+ createCounterSensor(
+ "data-written",
Review Comment:
I will change the metric base name from "data-written" to "data-files-written", the "-total" will be added inside createCounterSensor when an inner metric (-total) is created.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Add JMX metrics for commit pipeline [iceberg]
Posted by "HenryCaiHaiying (via GitHub)" <gi...@apache.org>.
HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3696658307
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CoordinatorMetrics.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect.channel;
+
+import java.util.function.Supplier;
+import org.apache.kafka.common.metrics.Sensor;
+
+class CoordinatorMetrics extends ChannelMetrics {
+
+ private static final String GROUP = "coordinator-metrics";
+ // The coordinator is a single per-connector task, so it reports a fixed task tag.
+ private static final String TASK = "coordinator";
+ private static final String FULL = "full";
+ private static final String PARTIAL = "partial";
+
+ // Commit timers are tagged by commitMode (partial vs full) so the two paths stay separable.
+ private final Sensor fullCommitTime;
+ private final Sensor partialCommitTime;
+ private final Sensor startCommit;
+ private final Sensor commitComplete;
+
+ CoordinatorMetrics(
+ String connector, Supplier<Long> commitBufferSize, Supplier<Long> readyBufferSize) {
+ super(GROUP, connector, TASK);
+ try {
+ this.fullCommitTime =
+ createTimerSensor(
+ "commit-time",
+ "Time spent in Coordinator.commit() in microseconds",
+ metricTags(connector, TASK, FULL));
+ this.partialCommitTime =
+ createTimerSensor(
+ "commit-time",
+ "Time spent in Coordinator.commit() in microseconds",
+ metricTags(connector, TASK, PARTIAL));
+ // Counters are bumped only after send() succeeds, so they count events successfully emitted;
+ // a failed send leaves them unmoved even though the commit is already in progress.
+ this.startCommit =
+ createCounterSensor(
+ "start-commit",
+ "Number of successfully emitted START_COMMIT events",
+ metricTags(connector, TASK, null));
+ this.commitComplete =
+ createCounterSensor(
+ "commit-complete",
+ "Number of successfully emitted COMMIT_COMPLETE events",
+ metricTags(connector, TASK, null));
+
+ addGauge(
+ "commit-buffer-size",
+ "Current size of CommitState.commitBuffer",
+ metricTags(connector, TASK, null),
+ commitBufferSize);
+ addGauge(
+ "ready-buffer-size",
+ "Current size of CommitState.readyBuffer",
+ metricTags(connector, TASK, null),
+ readyBufferSize);
+ } catch (RuntimeException e) {
+ closeQuietly(e);
+ throw e;
+ }
+ }
+
+ void recordCommit(boolean partialCommit, long elapsedMs) {
Review Comment:
Will rename the variable to `elapsedMicros`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org