cadonna commented on a change in pull request #9177:
URL: https://github.com/apache/kafka/pull/9177#discussion_r472276615
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
##########
@@ -415,9 +416,40 @@ public final Sensor storeLevelSensor(final String threadId,
}
}
- public final void removeAllStoreLevelSensors(final String threadId,
- final String taskId,
- final String storeName) {
+ public <T> void addStoreLevelMutableMetric(final String threadId,
+ final String taskId,
+ final String metricsScope,
+ final String storeName,
+ final String name,
+ final String description,
+ final RecordingLevel
recordingLevel,
+ final Gauge<T> valueProvider) {
+ final MetricName metricName = metrics.metricName(
+ name,
+ STATE_STORE_LEVEL_GROUP,
+ description,
+ storeLevelTagMap(threadId, taskId, metricsScope, storeName)
+ );
+ if (metrics.metric(metricName) == null) {
+ final MetricConfig metricConfig = new
MetricConfig().recordLevel(recordingLevel);
+ final String key = storeSensorPrefix(threadId, taskId, storeName);
+ synchronized (storeLevelMetrics) {
+ metrics.addMetric(metricName, metricConfig, valueProvider);
+ storeLevelMetrics.computeIfAbsent(key, ignored -> new
LinkedList<>()).push(metricName);
+ }
+ }
+ }
+
+ public final void removeAllStoreLevelSensorsAndMetrics(final String
threadId,
+ final String taskId,
+ final String
storeName) {
+ removeAllStoreLevelMetrics(threadId, taskId, storeName);
+ removeAllStoreLevelSensors(threadId, taskId, storeName);
+ }
Review comment:
Yes, we need to synchronize. At least, we have to ensure that lines 411
and 438 are thread-safe. Then, if we do not want to have duplicates in
`storeLevelSensors` we should ensure to have a lock between line 409 to 411.
Between line 434 and 439, we need to ensure that the removal of all store level
metrics completed otherwise it could happen that we find a store level metric
that would prevent the addition of a metric but then the earlier found metric
would be removed during the remainder of the removal process. Similar is true
for the store level sensors.
It is true that we always remove all of both collections together, but we do
not add metric names and sensor names to both collections together.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
