AndrewJSchofield commented on code in PR #19045:
URL: https://github.com/apache/kafka/pull/19045#discussion_r1973396006
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1265,19 +1289,7 @@ boolean canAcquireRecords() {
if (nextFetchOffset() != endOffset() + 1) {
return true;
}
-
- lock.readLock().lock();
- long numRecords;
- try {
- if (cachedState.isEmpty()) {
- numRecords = 0;
- } else {
- numRecords = this.endOffset - this.startOffset + 1;
- }
- } finally {
- lock.readLock().unlock();
- }
- return numRecords < maxInFlightMessages;
+ return numInflightRecords() < maxInFlightMessages;
Review Comment:
nit: `numInFlightRecords` for consistency.
##########
server/src/main/java/org/apache/kafka/server/share/metrics/SharePartitionMetrics.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.kafka.server.share.metrics;
+
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+/**
+ * SharePartitionMetrics is used to track the broker-side metrics for the
SharePartition.
+ */
+public class SharePartitionMetrics implements AutoCloseable {
+
+ public static final String INFLIGHT_MESSAGE_COUNT = "InFlightMessageCount";
Review Comment:
nit: Ought to be `IN_FLIGHT_` given the capitalization in the string version.
##########
server/src/main/java/org/apache/kafka/server/share/metrics/SharePartitionMetrics.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.kafka.server.share.metrics;
+
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+/**
+ * SharePartitionMetrics is used to track the broker-side metrics for the
SharePartition.
+ */
+public class SharePartitionMetrics implements AutoCloseable {
+
+ public static final String INFLIGHT_MESSAGE_COUNT = "InFlightMessageCount";
+ public static final String INFLIGHT_BATCH_COUNT = "InFlightBatchCount";
+
+ private static final String ACQUISITION_LOCK_TIMEOUT_PER_SEC =
"AcquisitionLockTimeoutPerSec";
+ private static final String INFLIGHT_BATCH_MESSAGE_COUNT =
"InFlightBatchMessageCount";
+ private static final String FETCH_LOCK_TIME_MS = "FetchLockTimeMs";
+ private static final String FETCH_LOCK_RATIO = "FetchLockRatio";
+
+ /**
+ * Metric for the rate of acquisition lock timeouts for records.
+ */
+ private final Meter acquisitionLockTimeoutPerSec;
+ /**
+ * Metric for the number of in-flight messages for the batch.
+ */
+ private final Histogram inFlightBatchMessageCount;
+ /**
+ * Metric for the time the fetch lock is held.
+ */
+ private final Histogram fetchLockTimeMs;
+ /**
+ * Metric for the ratio of fetch lock time to the total time.
+ */
+ private final Histogram fetchLockRatio;
+
+ private final Map<String, String> tags;
+ private final KafkaMetricsGroup metricsGroup;
+
+ public SharePartitionMetrics(String groupId, String topic, int partition) {
+ this.tags = Utils.mkMap(
+ Utils.mkEntry("group", Objects.requireNonNull(groupId)),
+ Utils.mkEntry("topic", Objects.requireNonNull(topic)),
+ Utils.mkEntry("partition", String.valueOf(partition))
+ );
+ this.metricsGroup = new KafkaMetricsGroup("kafka.server",
"SharePartitionMetrics");
+
+ this.acquisitionLockTimeoutPerSec = metricsGroup.newMeter(
+ ACQUISITION_LOCK_TIMEOUT_PER_SEC,
+ "acquisition lock timeout",
+ TimeUnit.SECONDS,
+ this.tags);
+
+ this.inFlightBatchMessageCount = metricsGroup.newHistogram(
+ INFLIGHT_BATCH_MESSAGE_COUNT,
+ true,
+ this.tags);
+
+ this.fetchLockTimeMs = metricsGroup.newHistogram(
+ FETCH_LOCK_TIME_MS,
+ true,
+ this.tags);
+
+ this.fetchLockRatio = metricsGroup.newHistogram(
+ FETCH_LOCK_RATIO,
+ true,
+ this.tags);
+ }
+
+ /**
+ * Register a gauge for the in-flight message count.
+ *
+ * @param messageCountSupplier The supplier for the in-flight message
count.
+ */
+ public void registerInflightMessageCount(Supplier<Long>
messageCountSupplier) {
Review Comment:
`registerInFlight...` please. And the next method 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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
