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