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