You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2024/12/13 18:45:17 UTC
[PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
smjn opened a new pull request, #18174:
URL: https://github.com/apache/kafka/pull/18174
(no comment)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #18174:
URL: https://github.com/apache/kafka/pull/18174
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887252720
##########
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:
The serves as proof of the record pruner working from the monitoring perspective. This is capturing the value of the offset for which the delete call was successful.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249722
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -46,6 +48,8 @@ public class ShareCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String SHARE_COORDINATOR_WRITE_SENSOR_NAME = "ShareCoordinatorWrite";
public static final String SHARE_COORDINATOR_WRITE_LATENCY_SENSOR_NAME = "ShareCoordinatorWriteLatency";
+ public static final String SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME = "ShareCoordinatorStateTopicPruneSensorName";
+ private Map<TopicPartition, ShareGroupPruneMetrics> pruneMetrics = new HashMap<>();
Review Comment:
This will be called from single threaded context
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887252720
##########
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:
The serves as proof of the record pruner working from the monitoring perspective. On the off chance that we suspect that the pruner has deleted overwhelming amount of data - this could prove useful data. Secondly, in share coordinator terminology this is just an offset and we are not privy to what the core replicaManger and partition implementation use.
This is capturing the value of the offset for which the delete call was successful. Further details can be ascertained from the broker logs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887909112
##########
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:
No we needn't remove metrics explicitly since `removeSensor` method in `Metrics.java` https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java#L442 manages that internally.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249392
##########
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:
Checking the code I did not see that in success case. But suppose we ask to prune an offset which is lower than low watermark - we could get that result.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249722
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -46,6 +48,8 @@ public class ShareCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String SHARE_COORDINATOR_WRITE_SENSOR_NAME = "ShareCoordinatorWrite";
public static final String SHARE_COORDINATOR_WRITE_LATENCY_SENSOR_NAME = "ShareCoordinatorWriteLatency";
+ public static final String SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME = "ShareCoordinatorStateTopicPruneSensorName";
+ private Map<TopicPartition, ShareGroupPruneMetrics> pruneMetrics = new HashMap<>();
Review Comment:
This will be called from single threaded context. There is a single instance of share coordinator
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887904112
##########
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:
It is not a question of what we need to rely on - it is what the pruner is sending to the replicaManager. This is the correct information we want to capture. Capturing of low water mark is already present.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887170307
##########
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:
replica manager's `deleteRecords` method respond with `DeleteRecordsPartitionResult` which contains new `lowWaterMark`. Can there be a case when `lowWaterMark` from replica manager is different than from offset requested?
##########
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 essentially the metric is capturing the lowWaterMark for topic partition, hence what's the relevance of calling it `pruned`? Also what monitoring importance this metric captures?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -46,6 +48,8 @@ public class ShareCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String SHARE_COORDINATOR_WRITE_SENSOR_NAME = "ShareCoordinatorWrite";
public static final String SHARE_COORDINATOR_WRITE_LATENCY_SENSOR_NAME = "ShareCoordinatorWriteLatency";
+ public static final String SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME = "ShareCoordinatorStateTopicPruneSensorName";
+ private Map<TopicPartition, ShareGroupPruneMetrics> pruneMetrics = new HashMap<>();
Review Comment:
Should it be thread safe? Can recordPrune be called from multiple threads?
##########
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:
Do we really need this or can rely on `metrics.getSensor(name)` as internally a map already exists there?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249392
##########
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:
Checking the code I did not see that in success case. But suppose we ask to prune an offset which is lower than low watermark - we could get that result. However, no delete will happen.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249722
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -46,6 +48,8 @@ public class ShareCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String SHARE_COORDINATOR_WRITE_SENSOR_NAME = "ShareCoordinatorWrite";
public static final String SHARE_COORDINATOR_WRITE_LATENCY_SENSOR_NAME = "ShareCoordinatorWriteLatency";
+ public static final String SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME = "ShareCoordinatorStateTopicPruneSensorName";
+ private Map<TopicPartition, ShareGroupPruneMetrics> pruneMetrics = new HashMap<>();
Review Comment:
This will be called from single threaded context. There is a single instance of share coordinator running a 5 minute periodic job that will call delete. Will change to `concurrent`, nevertheless.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249722
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/metrics/ShareCoordinatorMetrics.java:
##########
@@ -46,6 +48,8 @@ public class ShareCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String SHARE_COORDINATOR_WRITE_SENSOR_NAME = "ShareCoordinatorWrite";
public static final String SHARE_COORDINATOR_WRITE_LATENCY_SENSOR_NAME = "ShareCoordinatorWriteLatency";
+ public static final String SHARE_COORDINATOR_STATE_TOPIC_PRUNE_SENSOR_NAME = "ShareCoordinatorStateTopicPruneSensorName";
+ private Map<TopicPartition, ShareGroupPruneMetrics> pruneMetrics = new HashMap<>();
Review Comment:
This will be called from single threaded context. There is a single instance of share coordinator. Will change to `concurrent`, nevertheless.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887908162
##########
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:
Like the case I mentioned our last pruned offset could be different we what the low water mark is (if we send something smaller). Current `Partition.scala` code simply swallows this condition with no specific indication.
Another point is that our `last-pruned-offset` is dual purposed. It is also giving the indication that the pruner is working, which the other offsets do not provide.
This metric provides pruners perspective, low water mark/LogStartOffset provide replicamanager/partition/Unified log perspectives.
If however, this information is appears to be derivative, then we needn't make any changes and this PR can be closed as LogStartOffset is being tracked.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887233880
##########
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:
@apoorvmittal10
We can use that but we would anyway need a set or map to store sensor names since we need to remove the sensors from metrics in the `close` method. This is necessary as we do not know beforehand how many tps are there.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887252720
##########
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:
The serves as proof of the record pruner working from the monitoring perspective. On the off chance that we suspect that the pruner has deleted overwhelming amount of data - this could prove useful. This is capturing the value of the offset for which the delete call was successful.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887249392
##########
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:
Checking the code I did not see that in success case.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887252720
##########
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:
The serves as proof of the record pruner working from the monitoring perspective. On the off chance that we suspect that the pruner has deleted overwhelming amount of data - this could prove useful data. This is capturing the value of the offset for which the delete call was successful. Further details can be ascertained from the broker logs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887908162
##########
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:
Like the case I mentioned our last pruned offset could be different we what the low water mark is (if we send something smaller). Current `Partition.scala` code simply swallows this condition with no specific indication.
This metric provides pruners perspective, low water mark/LogStartOffset provide replicamanager/partition/Unified log perspectives.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887908162
##########
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:
Like the case I mentioned our last pruned offset could be different we what the low water mark is (if we send something smaller). Current `Partition.scala` code simply swallows this condition with no specific indication.
This metric provides pruners perspective, low water mark/LogStartOffset provide replicamanager/partition/Unified log perspectives.
If however, this information is appears to be derivative. Then we needn't make any changes and this PR can be closed as LogStartOffset is being tracked.
##########
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:
Like the case I mentioned our last pruned offset could be different we what the low water mark is (if we send something smaller). Current `Partition.scala` code simply swallows this condition with no specific indication.
This metric provides pruners perspective, low water mark/LogStartOffset provide replicamanager/partition/Unified log perspectives.
If however, this information is appears to be derivative, then we needn't make any changes and this PR can be closed as LogStartOffset is being tracked.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18174:
URL: https://github.com/apache/kafka/pull/18174#issuecomment-2547529071
> Thanks for the details, I have some more questions to understand the usage of metric.
@apoorvmittal10
Thanks for next review iteration. Responded to comments.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #18174:
URL: https://github.com/apache/kafka/pull/18174#issuecomment-2546298263
> Thanks for the PR, I have taken a pass and have some comments.
@apoorvmittal10
Thanks for the review. Incorporated some changes - responded to others.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1887252720
##########
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:
The serves as proof of the record pruner working from the monitoring perspective. On the off chance that we suspect that the pruner has deleted overwhelming amount of data - this could prove useful data. This is capturing the value of the offset for which the delete call was successful.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18232: Add share group state topic prune metrics. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18174:
URL: https://github.com/apache/kafka/pull/18174#discussion_r1922553367
##########
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:
I think perhaps the values will match in general. However, I think it's helpful to have this metric in the share-coordinator group.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org