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: [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]
