Skip to content

KAFKA-7660: fix streams and Metrics memory leaks #5980

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
mjsax merged 2 commits into from
Dec 4, 2018
Merged

KAFKA-7660: fix streams and Metrics memory leaks #5980

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7e71073
KAFKA-7660: fix parentSensors memory leak (#5953)
vvcephei Nov 29, 2018
134f9c5
KAFKA-7660: Fix child sensor memory leak (#5974)
vvcephei Nov 30, 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
5 changes: 5 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static java.util.Collections.emptyList;

/**
* A registry of sensors and metrics.
* <p>
Expand Down Expand Up @@ -446,6 +448,9 @@ public void removeSensor(String name) {
removeMetric(metric.metricName());
log.debug("Removed sensor with name {}", name);
childSensors = childrenSensors.remove(sensor);
for (final Sensor parent : sensor.parents()) {
childrenSensors.getOrDefault(parent, emptyList()).remove(sensor);
}
}
}
}
Expand Down
10 changes: 8 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import org.apache.kafka.common.utils.Utils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedList;
Expand All @@ -32,6 +31,9 @@
import java.util.Set;
import java.util.concurrent.TimeUnit;

import static java.util.Arrays.asList;
import static java.util.Collections.unmodifiableList;

/**
* A sensor applies a continuous sequence of numerical values to a set of associated metrics. For example a sensor on
* message size would record a sequence of message sizes using the {@link #record(double)} api and would maintain a set
Expand Down Expand Up @@ -133,6 +135,10 @@ public String name() {
return this.name;
}

List<Sensor> parents() {
return unmodifiableList(asList(parents));
}

/**
* Record an occurrence, this is just short-hand for {@link #record(double) record(1.0)}
*/
Expand Down Expand Up @@ -291,7 +297,7 @@ public boolean hasExpired() {
}

synchronized List<KafkaMetric> metrics() {
return Collections.unmodifiableList(new LinkedList<>(this.metrics.values()));
return unmodifiableList(new LinkedList<>(this.metrics.values()));
}

/**
Expand Down
16 changes: 16 additions & 0 deletions clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@
*/
package org.apache.kafka.common.metrics;

import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
Expand Down Expand Up @@ -197,6 +199,20 @@ public void testBadSensorHierarchy() {
metrics.sensor("gc", c1, c2); // should fail
}

@Test
public void testRemoveChildSensor() {
final Metrics metrics = new Metrics();

final Sensor parent = metrics.sensor("parent");
final Sensor child = metrics.sensor("child", parent);

assertEquals(singletonList(child), metrics.childrenSensors().get(parent));

metrics.removeSensor("child");

assertEquals(emptyList(), metrics.childrenSensors().get(parent));
}

@Test
public void testRemoveSensor() {
int size = metrics.metrics().size();
Expand Down
24 changes: 13 additions & 11 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 @@ -110,11 +110,9 @@ public final Sensor taskLevelSensor(final String taskName,
public final void removeAllTaskLevelSensors(final String taskName) {
final String key = threadName + "." + 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 @@ -143,11 +141,9 @@ public final Sensor cacheLevelSensor(final String taskName,
public final void removeAllCacheLevelSensors(final String taskName, final String cacheName) {
final String key = threadName + "." + 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 @@ -361,10 +357,16 @@ 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);
if (parent != null) {
metrics.removeSensor(parent.name());
}
}

/**
* Visible for testing
*/
Map<Sensor, Sensor> parentSensors() {
return Collections.unmodifiableMap(parentSensors);
}
}
61 changes: 58 additions & 3 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,14 +14,20 @@
* 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;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.common.metrics.stats.Count;
import org.junit.Test;

import java.util.Collections;
import java.util.Map;

import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.junit.Assert.assertEquals;

public class StreamsMetricsImplTest {
Expand Down Expand Up @@ -57,6 +63,55 @@ public void testRemoveSensor() {

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

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

@Test
public void testMutiLevelSensorRemoval() {
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> threadTags = mkMap(mkEntry("threadkey", "value"));

final Map<String, String> taskTags = mkMap(mkEntry("taskkey", "value"));

final Sensor parent1 = metrics.threadLevelSensor(operation, Sensor.RecordingLevel.DEBUG);
parent1.add(new MetricName("name", "group", "description", threadTags), new Count());

assertEquals(1, registry.metrics().size());

final Sensor sensor1 = metrics.taskLevelSensor(taskName, operation, Sensor.RecordingLevel.DEBUG, parent1);
sensor1.add(new MetricName("name", "group", "description", taskTags), new Count());

assertEquals(2, registry.metrics().size());

metrics.removeAllTaskLevelSensors(taskName);

assertEquals(1, registry.metrics().size());

final Sensor parent2 = metrics.threadLevelSensor(operation, Sensor.RecordingLevel.DEBUG);
parent2.add(new MetricName("name", "group", "description", threadTags), new Count());

assertEquals(1, registry.metrics().size());

final Sensor sensor2 = metrics.taskLevelSensor(taskName, operation, Sensor.RecordingLevel.DEBUG, parent2);
sensor2.add(new MetricName("name", "group", "description", taskTags), new Count());

assertEquals(2, registry.metrics().size());

metrics.removeAllTaskLevelSensors(taskName);

assertEquals(1, registry.metrics().size());

metrics.removeAllThreadLevelSensors();

assertEquals(0, registry.metrics().size());
}

@Test
Expand Down Expand Up @@ -90,7 +145,7 @@ public void testThroughputMetrics() {
final String entity = "entity";
final String operation = "put";

final Sensor sensor1 = streamsMetrics.addThroughputSensor(taskName, scope, entity, operation, Sensor.RecordingLevel.DEBUG);
final Sensor sensor1 = streamsMetrics.addThroughputSensor(taskName, scope, entity, operation, Sensor.RecordingLevel.DEBUG);

final int meterMetricsCount = 2; // Each Meter is a combination of a Rate and a Total
// 2 meter metrics plus a common metric that keeps track of total registered metrics in Metrics() constructor
Expand Down