Skip to content

KAFKA-7660: fix parentSensors memory leak #5953

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
guozhangwang merged 4 commits into from
Nov 29, 2018
Merged

KAFKA-7660: fix parentSensors memory leak #5953

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
efabc14
KAFKA-7660: fix parentSensors memory leak
vvcephei Nov 27, 2018
9a6456a
unit test
vvcephei Nov 27, 2018
e68167c
clean up and test creation and removal of x-Level sensors
vvcephei Nov 28, 2018
4db2541
cr comments
vvcephei Nov 28, 2018
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
2 changes: 2 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,9 +234,11 @@ private static Sensor createTaskAndNodeLatencyAndThroughputSensors(final String
final Sensor parent = metrics.taskLevelSensor(taskName, operation, Sensor.RecordingLevel.DEBUG);
addAvgMaxLatency(parent, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);
addInvocationRateAndCount(parent, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);

final Sensor sensor = metrics.nodeLevelSensor(taskName, processorNodeName, operation, Sensor.RecordingLevel.DEBUG, parent);
addAvgMaxLatency(sensor, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);
addInvocationRateAndCount(sensor, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);

return sensor;
}
}
Expand Down
40 changes: 20 additions & 20 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 @@ -115,11 +115,9 @@ public final Sensor taskLevelSensor(final String taskName,
public final void removeAllTaskLevelSensors(final String taskName) {
final String key = taskSensorPrefix(taskName);
synchronized (taskLevelSensors) {
if (taskLevelSensors.containsKey(key)) {
while (!taskLevelSensors.get(key).isEmpty()) {
metrics.removeSensor(taskLevelSensors.get(key).pop());
}
taskLevelSensors.remove(key);
final Deque<String> sensors = taskLevelSensors.remove(key);
while (sensors != null && !sensors.isEmpty()) {
metrics.removeSensor(sensors.pop());
}
}
}
Expand Down Expand Up @@ -152,10 +150,9 @@ public Sensor nodeLevelSensor(final String taskName,
public final void removeAllNodeLevelSensors(final String taskName, final String processorNodeName) {
final String key = nodeSensorPrefix(taskName, processorNodeName);
synchronized (nodeLevelSensors) {
if (nodeLevelSensors.containsKey(key)) {
while (!nodeLevelSensors.get(key).isEmpty()) {
metrics.removeSensor(nodeLevelSensors.get(key).pop());
}
final Deque<String> sensors = nodeLevelSensors.remove(key);
while (sensors != null && !sensors.isEmpty()) {
metrics.removeSensor(sensors.pop());
}
}
}
Expand Down Expand Up @@ -188,11 +185,9 @@ public final Sensor cacheLevelSensor(final String taskName,
public final void removeAllCacheLevelSensors(final String taskName, final String cacheName) {
final String key = cacheSensorPrefix(taskName, cacheName);
synchronized (cacheLevelSensors) {
if (cacheLevelSensors.containsKey(key)) {
while (!cacheLevelSensors.get(key).isEmpty()) {
metrics.removeSensor(cacheLevelSensors.get(key).pop());
}
cacheLevelSensors.remove(key);
final Deque<String> strings = cacheLevelSensors.remove(key);
while (strings != null && !strings.isEmpty()) {
metrics.removeSensor(strings.pop());
}
}
}
Expand Down Expand Up @@ -225,11 +220,9 @@ public final Sensor storeLevelSensor(final String taskName,
public final void removeAllStoreLevelSensors(final String taskName, final String storeName) {
final String key = storeSensorPrefix(taskName, storeName);
synchronized (storeLevelSensors) {
if (storeLevelSensors.containsKey(key)) {
while (!storeLevelSensors.get(key).isEmpty()) {
metrics.removeSensor(storeLevelSensors.get(key).pop());
}
storeLevelSensors.remove(key);
final Deque<String> sensors = storeLevelSensors.remove(key);
while (sensors != null && !sensors.isEmpty()) {
metrics.removeSensor(sensors.pop());
}
}
}
Expand Down Expand Up @@ -413,12 +406,19 @@ public void removeSensor(final Sensor sensor) {
Objects.requireNonNull(sensor, "Sensor is null");
metrics.removeSensor(sensor.name());

final Sensor parent = parentSensors.get(sensor);
final Sensor parent = parentSensors.remove(sensor);

@mjsax mjsax Nov 27, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nice catch!

if (parent != null) {
metrics.removeSensor(parent.name());
}
}

/**
* Visible for testing
*/
Map<Sensor, Sensor> parentSensors() {
return Collections.unmodifiableMap(parentSensors);
}

private static String groupNameFromScope(final String scopeName) {
return "stream-" + scopeName + "-metrics";
}
Expand Down
89 changes: 76 additions & 13 deletions ...sor/internals/StreamsMetricsImplTest.java → ...rnals/metrics/StreamsMetricsImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.streams.processor.internals;
package org.apache.kafka.streams.processor.internals.metrics;


import org.apache.kafka.common.MetricName;
Expand All @@ -23,12 +23,21 @@
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.junit.Test;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.PROCESSOR_NODE_METRICS_GROUP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgMaxLatency;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;

public class StreamsMetricsImplTest {

Expand Down Expand Up @@ -62,6 +71,60 @@ public void testRemoveSensor() {

final Sensor sensor3 = streamsMetrics.addThroughputSensor(scope, entity, operation, Sensor.RecordingLevel.DEBUG);
streamsMetrics.removeSensor(sensor3);

assertEquals(Collections.emptyMap(), streamsMetrics.parentSensors());
}

@Test
public void testMutiLevelSensorRemoval() {

@miguno miguno Nov 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Typo: testMu[l]tiLevelSensorRemoval() {

final Metrics registry = new Metrics();
final StreamsMetricsImpl metrics = new StreamsMetricsImpl(registry, "");
for (final MetricName defaultMetric : registry.metrics().keySet()) {
registry.removeMetric(defaultMetric);
}

final String taskName = "taskName";
final String operation = "operation";
final Map<String, String> taskTags = mkMap(mkEntry("tkey", "value"));

final String processorNodeName = "processorNodeName";
final Map<String, String> nodeTags = mkMap(mkEntry("nkey", "value"));

final Sensor parent1 = metrics.taskLevelSensor(taskName, operation, Sensor.RecordingLevel.DEBUG);
addAvgMaxLatency(parent1, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);
addInvocationRateAndCount(parent1, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);

final int numberOfTaskMetrics = registry.metrics().size();

final Sensor sensor1 = metrics.nodeLevelSensor(taskName, processorNodeName, operation, Sensor.RecordingLevel.DEBUG, parent1);
addAvgMaxLatency(sensor1, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);
addInvocationRateAndCount(sensor1, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);

assertThat(registry.metrics().size(), greaterThan(numberOfTaskMetrics));

metrics.removeAllNodeLevelSensors(taskName, processorNodeName);

assertThat(registry.metrics().size(), equalTo(numberOfTaskMetrics));

final Sensor parent2 = metrics.taskLevelSensor(taskName, operation, Sensor.RecordingLevel.DEBUG);

@guozhangwang guozhangwang Nov 29, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is this used for? Sensor parent1 and parent2 should be referring to exactly the same object right?

@vvcephei vvcephei Nov 29, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They should be, but in reality the same parent sensor is (re)registered each time a child sensor is created, so I wanted to (idempotently) create a second parent sensor to make sure there's no extra references.

addAvgMaxLatency(parent2, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);
addInvocationRateAndCount(parent2, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);

assertThat(registry.metrics().size(), equalTo(numberOfTaskMetrics));

final Sensor sensor2 = metrics.nodeLevelSensor(taskName, processorNodeName, operation, Sensor.RecordingLevel.DEBUG, parent2);
addAvgMaxLatency(sensor2, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);
addInvocationRateAndCount(sensor2, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);

assertThat(registry.metrics().size(), greaterThan(numberOfTaskMetrics));

metrics.removeAllNodeLevelSensors(taskName, processorNodeName);

assertThat(registry.metrics().size(), equalTo(numberOfTaskMetrics));

metrics.removeAllTaskLevelSensors(taskName);

assertThat(registry.metrics().size(), equalTo(0));
}

@Test
Expand Down Expand Up @@ -115,21 +178,21 @@ public void testTotalMetricDoesntDecrease() {
final String operation = "op";

final Sensor sensor = streamsMetrics.addLatencyAndThroughputSensor(
scope,
entity,
operation,
Sensor.RecordingLevel.INFO
scope,
entity,
operation,
Sensor.RecordingLevel.INFO
);

final double latency = 100.0;
final MetricName totalMetricName = metrics.metricName(
"op-total",
"stream-scope-metrics",
"",
"client-id",
"",
"scope-id",
"entity"
"op-total",
"stream-scope-metrics",
"",
"client-id",
"",
"scope-id",
"entity"
);

final KafkaMetric totalMetric = metrics.metric(totalMetricName);
Expand Down