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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
