AndrewJSchofield commented on code in PR #18444:
URL: https://github.com/apache/kafka/pull/18444#discussion_r1909067548
##########
storage/src/main/java/org/apache/kafka/storage/log/metrics/BrokerTopicMetrics.java:
##########
@@ -75,8 +79,12 @@ private BrokerTopicMetrics(Optional<String> name, boolean
remoteStorageEnabled)
metricTypeMap.put(BYTES_REJECTED_PER_SEC, new
MeterWrapper(BYTES_REJECTED_PER_SEC, "bytes"));
metricTypeMap.put(FAILED_PRODUCE_REQUESTS_PER_SEC, new
MeterWrapper(FAILED_PRODUCE_REQUESTS_PER_SEC, "requests"));
metricTypeMap.put(FAILED_FETCH_REQUESTS_PER_SEC, new
MeterWrapper(FAILED_FETCH_REQUESTS_PER_SEC, "requests"));
+ metricTypeMap.put(FAILED_SHARE_FETCH_REQUESTS_PER_SEC, new
MeterWrapper(FAILED_SHARE_FETCH_REQUESTS_PER_SEC, "requests"));
+ metricTypeMap.put(FAILED_SHARE_ACKNOWLEDGEMENTS_REQUESTS_PER_SEC, new
MeterWrapper(FAILED_SHARE_ACKNOWLEDGEMENTS_REQUESTS_PER_SEC, "requests"));
metricTypeMap.put(TOTAL_PRODUCE_REQUESTS_PER_SEC, new
MeterWrapper(TOTAL_PRODUCE_REQUESTS_PER_SEC, "requests"));
metricTypeMap.put(TOTAL_FETCH_REQUESTS_PER_SEC, new
MeterWrapper(TOTAL_FETCH_REQUESTS_PER_SEC, "requests"));
+ metricTypeMap.put(TOTAL_SHARE_FETCH_REQUESTS_PER_SEC, new
MeterWrapper(TOTAL_SHARE_FETCH_REQUESTS_PER_SEC, "requests"));
+ metricTypeMap.put(TOTAL_SHARE_ACKNOWLEDGEMENTS_REQUESTS_PER_SEC, new
MeterWrapper(TOTAL_SHARE_ACKNOWLEDGEMENTS_REQUESTS_PER_SEC, "requests"));
Review Comment:
I think that this metric is missing from KIP-1103.
##########
share/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTest.java:
##########
@@ -87,4 +95,104 @@ public void testFilterErroneousTopicPartitions() {
assertTrue(result.isEmpty());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testMayBeCompleteWithErroneousTopicPartitions() {
Review Comment:
nit: This should be "Maybe".
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -361,24 +376,31 @@ public CompletableFuture<Map<TopicIdPartition,
ShareAcknowledgeResponseData.Part
}
});
- return mapAcknowledgementFutures(futuresMap);
+ return mapAcknowledgementFutures(futuresMap, Optional.empty());
}
- private CompletableFuture<Map<TopicIdPartition,
ShareAcknowledgeResponseData.PartitionData>>
mapAcknowledgementFutures(Map<TopicIdPartition, CompletableFuture<Throwable>>
futuresMap) {
+ private CompletableFuture<Map<TopicIdPartition,
ShareAcknowledgeResponseData.PartitionData>> mapAcknowledgementFutures(
+ Map<TopicIdPartition, CompletableFuture<Throwable>> futuresMap,
+ Optional<BiConsumer<Collection<String>, Boolean>> metricsHandler
Review Comment:
Personally, I think it would be neater to have an interface for the metrics
handler as opposed to using `BiConsumer<Collection<String>, Boolean>`
everywhere.
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -272,9 +285,11 @@ public CompletableFuture<Map<TopicIdPartition,
ShareAcknowledgeResponseData.Part
) {
log.trace("Acknowledge request for topicIdPartitions: {} with groupId:
{}",
acknowledgeTopics.keySet(), groupId);
- this.shareGroupMetrics.shareAcknowledgement();
+
brokerTopicStats.allTopicsStats().totalShareAcknowledgementRequestRate().mark();
Map<TopicIdPartition, CompletableFuture<Throwable>> futures = new
HashMap<>();
acknowledgeTopics.forEach((topicIdPartition,
acknowledgePartitionBatches) -> {
+ // Update share acknowledgement metrics.
+
brokerTopicStats.topicStats(topicIdPartition.topicPartition().topic()).totalShareAcknowledgementRequestRate().mark();
Review Comment:
Can't you just use `topicIdPartition.topic()`?
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -693,6 +717,38 @@ private static void removeSharePartitionFromCache(
}
}
+ /**
+ * The handler to update the failed share fetch request metrics.
+ *
+ * @return A BiConsumer that updates the failed share fetch request
metrics.
+ */
+ private BiConsumer<Collection<TopicIdPartition>, Boolean>
failedShareFetchMetricsHandler() {
+ return (topicIdPartitions, allTopicPartitionsFailed) -> {
+ // Update failed share fetch request metric.
+ topicIdPartitions.forEach(topicIdPartition ->
+
brokerTopicStats.topicStats(topicIdPartition.topicPartition().topic()).failedShareFetchRequestRate().mark());
+ if (allTopicPartitionsFailed) {
Review Comment:
Interesting. What you have done here is marked the all-topics metric if all
topic partitions failed, when I was really expecting you to mark it for every
topic partition which failed, rather than only when they all did.
If you look at `ReplicaManager.appendToLocalLog`, it marks once per failed
topic partition, not once when all topic partitions failed. I suspect this is
wrong and the boolean parameter is not necessary.
--
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
