apoorvmittal10 commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887340541
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -317,6 +317,10 @@ private CompletableFuture<Void>
performRecordPruning(TopicPartition tp) {
fut.completeExceptionally(exp);
return;
}
+ shareCoordinatorMetrics.recordPrune(
+ off,
Review Comment:
Yeah, so should we rely on lowWaterMark from Replica Manager as that will
capture the correct information?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -153,4 +158,31 @@ public void record(String sensorName) {
globalSensors.get(sensorName).record();
}
}
+
+ public void recordPrune(double value, TopicPartition tp) {
+ pruneMetrics.computeIfAbsent(tp, k -> new ShareGroupPruneMetrics(tp))
+ .pruneSensor.record(value);
+ }
+
+ private class ShareGroupPruneMetrics {
+ private final Sensor pruneSensor;
+
+ ShareGroupPruneMetrics(TopicPartition tp) {
+ String sensorNameSuffix = tp.toString();
+ Map<String, String> tags = Map.of(
+ "topic", tp.topic(),
+ "partition", Integer.toString(tp.partition())
+ );
+
+ pruneSensor =
metrics.sensor(SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME +
sensorNameSuffix);
+
+ pruneSensor.add(
+ metrics.metricName("last-pruned-offset",
+ METRICS_GROUP,
+ "The offset at which the share-group state topic was last
pruned.",
+ tags),
+ new Value()
+ );
+ }
+ }
Review Comment:
Yeah so we just need assigned partitions for the repective broker which can
be iterated. Also how the metric will be removed when a partition moves from
one broker to another, shouldn't we add handling for that?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -153,4 +158,31 @@ public void record(String sensorName) {
globalSensors.get(sensorName).record();
}
}
+
+ public void recordPrune(double value, TopicPartition tp) {
+ pruneMetrics.computeIfAbsent(tp, k -> new ShareGroupPruneMetrics(tp))
+ .pruneSensor.record(value);
+ }
+
+ private class ShareGroupPruneMetrics {
+ private final Sensor pruneSensor;
+
+ ShareGroupPruneMetrics(TopicPartition tp) {
+ String sensorNameSuffix = tp.toString();
+ Map<String, String> tags = Map.of(
+ "topic", tp.topic(),
+ "partition", Integer.toString(tp.partition())
+ );
+
+ pruneSensor =
metrics.sensor(SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME +
sensorNameSuffix);
+
+ pruneSensor.add(
+ metrics.metricName("last-pruned-offset",
Review Comment:
So how this metric differs from existing:
`kafka.log:type=Log,name=LogStartOffset,topic=__share_group_state,partition=<partition_num>`
<img width="895" alt="Screenshot 2024-12-17 at 00 11 17"
src="https://github.com/user-attachments/assets/28a90c78-cb39-48f1-ab73-df34a9ff5205";
/>
<img width="1216" alt="Screenshot 2024-12-17 at 00 12 21"
src="https://github.com/user-attachments/assets/f8e8885d-09e8-4a17-b881-0811016d5475";
/>
--
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]
