Skip to content

MINOR: Clean up streams metric sensors #9696

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 3 commits into from
Dec 9, 2020
Merged

MINOR: Clean up streams metric sensors #9696

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7033d1b
Updating adding sensors
Dec 4, 2020
e7dac98
Removing duplicate code
Dec 8, 2020
8c7b912
minor cleanup
Dec 8, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 25 additions & 37 deletions ...rc/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,6 @@
import java.util.LinkedList;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -220,11 +219,12 @@ public final Sensor clientLevelSensor(final String sensorName,
final Sensor... parents) {
synchronized (clientLevelSensors) {
final String fullSensorName = CLIENT_LEVEL_GROUP + SENSOR_NAME_DELIMITER + sensorName;
return Optional.ofNullable(metrics.getSensor(fullSensorName))
.orElseGet(() -> {
clientLevelSensors.push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
});
final Sensor sensor = metrics.getSensor(fullSensorName);
if (sensor == null) {
clientLevelSensors.push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
}
return sensor;
}
}

Expand All @@ -234,12 +234,7 @@ public final Sensor threadLevelSensor(final String threadId,
final Sensor... parents) {
final String key = threadSensorPrefix(threadId);
synchronized (threadLevelSensors) {
final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
return Optional.ofNullable(metrics.getSensor(fullSensorName))
.orElseGet(() -> {
threadLevelSensors.computeIfAbsent(key, ignored -> new LinkedList<>()).push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
});
return getSensors(threadLevelSensors, sensorName, key, recordingLevel, parents);
}
}

Expand Down Expand Up @@ -327,12 +322,7 @@ public final Sensor taskLevelSensor(final String threadId,
final Sensor... parents) {
final String key = taskSensorPrefix(threadId, taskId);
synchronized (taskLevelSensors) {
final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
return Optional.ofNullable(metrics.getSensor(fullSensorName))
.orElseGet(() -> {
taskLevelSensors.computeIfAbsent(key, ignored -> new LinkedList<>()).push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
});
return getSensors(taskLevelSensors, sensorName, key, recordingLevel, parents);
}
}

Expand All @@ -359,12 +349,7 @@ public Sensor nodeLevelSensor(final String threadId,
final Sensor... parents) {
final String key = nodeSensorPrefix(threadId, taskId, processorNodeName);
synchronized (nodeLevelSensors) {
final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
return Optional.ofNullable(metrics.getSensor(fullSensorName))
.orElseGet(() -> {
nodeLevelSensors.computeIfAbsent(key, ignored -> new LinkedList<>()).push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
});
return getSensors(nodeLevelSensors, sensorName, key, recordingLevel, parents);
}
}

Expand Down Expand Up @@ -393,12 +378,7 @@ public Sensor cacheLevelSensor(final String threadId,
final Sensor... parents) {
final String key = cacheSensorPrefix(threadId, taskName, storeName);
synchronized (cacheLevelSensors) {
final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
return Optional.ofNullable(metrics.getSensor(fullSensorName))
.orElseGet(() -> {
cacheLevelSensors.computeIfAbsent(key, ignored -> new LinkedList<>()).push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
});
return getSensors(cacheLevelSensors, sensorName, key, recordingLevel, parents);
}
}

Expand Down Expand Up @@ -437,20 +417,14 @@ public final Sensor storeLevelSensor(final String taskId,
final RecordingLevel recordingLevel,
final Sensor... parents) {
final String key = storeSensorPrefix(Thread.currentThread().getName(), taskId, storeName);
final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
final Sensor sensor = metrics.getSensor(fullSensorName);
if (sensor == null) {
// since the keys in the map storeLevelSensors contain the name of the current thread and threads only
// access keys in which their name is contained, the value in the maps do not need to be thread safe
// and we can use a LinkedList here.
// TODO: In future, we could use thread local maps since each thread will exclusively access the set of keys
// that contain its name. Similar is true for the other metric levels. Thread-level metrics need some
// special attention, since they are created before the thread is constructed. The creation of those
// metrics could be moved into the run() method of the thread.
storeLevelSensors.computeIfAbsent(key, ignored -> new LinkedList<>()).push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
}
return sensor;
return getSensors(storeLevelSensors, sensorName, key, recordingLevel, parents);
}

public <T> void addStoreLevelMutableMetric(final String taskId,
Expand Down Expand Up @@ -926,6 +900,20 @@ public static <T> T maybeMeasureLatency(final Supplier<T> actionToMeasure,
}
}

private Sensor getSensors(final Map<String, Deque<String>> sensors,
final String sensorName,
final String key,
final RecordingLevel recordingLevel,
final Sensor... parents) {
final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;
final Sensor sensor = metrics.getSensor(fullSensorName);
if (sensor == null) {
sensors.computeIfAbsent(key, ignored -> new LinkedList<>()).push(fullSensorName);
return metrics.sensor(fullSensorName, recordingLevel, parents);
}
return sensor;
}

/**
* Deletes a sensor and its parents, if any
*/
Expand Down