Skip to content

KAFKA-8782: Close metrics in QuotaManagerTests #7191

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
ijuma merged 3 commits into from
Aug 11, 2019
Merged

KAFKA-8782: Close metrics in QuotaManagerTests #7191

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9ec3d46
KAFKA-8782: ReplicationQuotaManagerTest and ClientQuotaManagerTest sh…
ijuma Aug 10, 2019
2ef6401
Use try with resources when closing metrics
ijuma Aug 10, 2019
3ebc52b
Remove redundant code
ijuma Aug 10, 2019
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
6 changes: 3 additions & 3 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 @@ -90,15 +90,15 @@ public Metrics() {
* Expiration of Sensors is disabled.
*/
public Metrics(Time time) {
this(new MetricConfig(), new ArrayList<MetricsReporter>(0), time);
this(new MetricConfig(), new ArrayList<>(0), time);
}

/**
* Create a metrics repository with no metric reporters and the given default configuration.
* Expiration of Sensors is disabled.
*/
public Metrics(MetricConfig defaultConfig, Time time) {
this(defaultConfig, new ArrayList<MetricsReporter>(0), time);
this(defaultConfig, new ArrayList<>(0), time);
}


Expand All @@ -108,7 +108,7 @@ public Metrics(MetricConfig defaultConfig, Time time) {
* @param defaultConfig The default config to use for all metrics that don't override their config
*/
public Metrics(MetricConfig defaultConfig) {
this(defaultConfig, new ArrayList<MetricsReporter>(0), Time.SYSTEM);
this(defaultConfig, new ArrayList<>(0), Time.SYSTEM);
}

/**
Expand Down
7 changes: 3 additions & 4 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 @@ -63,7 +63,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("deprecation")
public class MetricsTest {
private static final Logger log = LoggerFactory.getLogger(MetricsTest.class);

Expand All @@ -75,7 +74,7 @@ public class MetricsTest {

@Before
public void setup() {
this.metrics = new Metrics(config, Arrays.asList((MetricsReporter) new JmxReporter()), time, true);
this.metrics = new Metrics(config, Arrays.asList(new JmxReporter()), time, true);
}

@After
Expand Down Expand Up @@ -631,7 +630,7 @@ public void testMetricInstances() {
Map<String, String> childTagsWithValues = new HashMap<>();
childTagsWithValues.put("child-tag", "child-tag-value");

try (Metrics inherited = new Metrics(new MetricConfig().tags(parentTagsWithValues), Arrays.asList((MetricsReporter) new JmxReporter()), time, true)) {
try (Metrics inherited = new Metrics(new MetricConfig().tags(parentTagsWithValues), Arrays.asList(new JmxReporter()), time, true)) {
MetricName inheritedMetric = inherited.metricInstance(SampleMetrics.METRIC_WITH_INHERITED_TAGS, childTagsWithValues);

Map<String, String> filledOutTags = inheritedMetric.tags();
Expand Down Expand Up @@ -731,7 +730,7 @@ synchronized void processMetrics() {

final LockingReporter reporter = new LockingReporter();
this.metrics.close();
this.metrics = new Metrics(config, Arrays.asList((MetricsReporter) reporter), new MockTime(10), true);
this.metrics = new Metrics(config, Arrays.asList(reporter), new MockTime(10), true);
final Deque<Sensor> sensors = new ConcurrentLinkedDeque<>();
SensorCreator sensorCreator = new SensorCreator(metrics);

Expand Down
36 changes: 17 additions & 19 deletions clients/src/test/java/org/apache/kafka/common/metrics/SensorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,25 +86,23 @@ public void testShouldRecord() {
public void testExpiredSensor() {
MetricConfig config = new MetricConfig();
Time mockTime = new MockTime();
Metrics metrics = new Metrics(config, Arrays.asList((MetricsReporter) new JmxReporter()), mockTime, true);

long inactiveSensorExpirationTimeSeconds = 60L;
Sensor sensor = new Sensor(metrics, "sensor", null, config, mockTime,
inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel.INFO);

assertTrue(sensor.add(metrics.metricName("test1", "grp1"), new Avg()));

Map<String, String> emptyTags = Collections.emptyMap();
MetricName rateMetricName = new MetricName("rate", "test", "", emptyTags);
MetricName totalMetricName = new MetricName("total", "test", "", emptyTags);
Meter meter = new Meter(rateMetricName, totalMetricName);
assertTrue(sensor.add(meter));

mockTime.sleep(TimeUnit.SECONDS.toMillis(inactiveSensorExpirationTimeSeconds + 1));
assertFalse(sensor.add(metrics.metricName("test3", "grp1"), new Avg()));
assertFalse(sensor.add(meter));

metrics.close();
try (Metrics metrics = new Metrics(config, Arrays.asList(new JmxReporter()), mockTime, true)) {
long inactiveSensorExpirationTimeSeconds = 60L;
Sensor sensor = new Sensor(metrics, "sensor", null, config, mockTime,
inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel.INFO);

assertTrue(sensor.add(metrics.metricName("test1", "grp1"), new Avg()));

Map<String, String> emptyTags = Collections.emptyMap();
MetricName rateMetricName = new MetricName("rate", "test", "", emptyTags);
MetricName totalMetricName = new MetricName("total", "test", "", emptyTags);
Meter meter = new Meter(rateMetricName, totalMetricName);
assertTrue(sensor.add(meter));

mockTime.sleep(TimeUnit.SECONDS.toMillis(inactiveSensorExpirationTimeSeconds + 1));
assertFalse(sensor.add(metrics.metricName("test3", "grp1"), new Avg()));
assertFalse(sensor.add(meter));
}
}

@Test
Expand Down
3 changes: 1 addition & 2 deletions clients/src/test/java/org/apache/kafka/common/metrics/stats/FrequenciesTest.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.metrics.JmxReporter;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
Expand All @@ -46,7 +45,7 @@ public class FrequenciesTest {
public void setup() {
config = new MetricConfig().eventWindow(50).samples(2);
time = new MockTime();
metrics = new Metrics(config, Arrays.asList((MetricsReporter) new JmxReporter()), time, true);

@omkreddy omkreddy Aug 11, 2019

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.

unused import checkstyle error.

@ijuma ijuma Aug 11, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Thanks, I had forgotten to push the updated branch after I fixed this.

metrics = new Metrics(config, Arrays.asList(new JmxReporter()), time, true);
}

@After
Expand Down
28 changes: 10 additions & 18 deletions core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,14 +33,20 @@ import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.utils.{MockTime, Sanitizer}
import org.easymock.EasyMock
import org.junit.Assert.{assertEquals, assertTrue}
import org.junit.{Before, Test}
import org.junit.{After, Test}

class ClientQuotaManagerTest {
private val time = new MockTime

private val metrics = new Metrics(new MetricConfig(), Collections.emptyList(), time)
private val config = ClientQuotaManagerConfig(quotaBytesPerSecondDefault = 500)

var numCallbacks: Int = 0

@After
def tearDown(): Unit = {
metrics.close()
}

def callback (response: RequestChannel.Response) {
// Count how many times this callback is called for notifyThrottlingDone().
response match {
Expand All @@ -49,11 +55,6 @@ class ClientQuotaManagerTest {
}
}

@Before
def beforeMethod() {
numCallbacks = 0
}

private def buildRequest[T <: AbstractRequest](builder: AbstractRequest.Builder[T],
listenerName: ListenerName = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)): (T, RequestChannel.Request) = {

Expand Down Expand Up @@ -81,7 +82,7 @@ class ClientQuotaManagerTest {
}

private def testQuotaParsing(config: ClientQuotaManagerConfig, client1: UserClient, client2: UserClient, randomClient: UserClient, defaultConfigClient: UserClient) {
val clientMetrics = new ClientQuotaManager(config, newMetrics, Produce, time, "")
val clientMetrics = new ClientQuotaManager(config, metrics, Produce, time, "")

try {
// Case 1: Update the quota. Assert that the new quota value is returned
Expand Down Expand Up @@ -193,7 +194,7 @@ class ClientQuotaManagerTest {
@Test
def testQuotaConfigPrecedence() {
val quotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(quotaBytesPerSecondDefault=Long.MaxValue),
newMetrics, Produce, time, "")
metrics, Produce, time, "")

def checkQuota(user: String, clientId: String, expectedBound: Int, value: Int, expectThrottle: Boolean) {
assertEquals(expectedBound, quotaManager.quota(user, clientId).bound, 0.0)
Expand Down Expand Up @@ -265,7 +266,6 @@ class ClientQuotaManagerTest {

@Test
def testQuotaViolation() {
val metrics = newMetrics
val clientMetrics = new ClientQuotaManager(config, metrics, Produce, time, "")
val queueSizeMetric = metrics.metrics().get(metrics.metricName("queue-size", "Produce", ""))
try {
Expand Down Expand Up @@ -313,7 +313,6 @@ class ClientQuotaManagerTest {

@Test
def testRequestPercentageQuotaViolation() {
val metrics = newMetrics
val quotaManager = new ClientRequestQuotaManager(config, metrics, time, "", None)
quotaManager.updateQuota(Some("ANONYMOUS"), Some("test-client"), Some("test-client"), Some(Quota.upperBound(1)))
val queueSizeMetric = metrics.metrics().get(metrics.metricName("queue-size", "Request", ""))
Expand Down Expand Up @@ -376,7 +375,6 @@ class ClientQuotaManagerTest {

@Test
def testExpireThrottleTimeSensor() {
val metrics = newMetrics
val clientMetrics = new ClientQuotaManager(config, metrics, Produce, time, "")
try {
maybeRecord(clientMetrics, "ANONYMOUS", "client1", 100)
Expand All @@ -396,7 +394,6 @@ class ClientQuotaManagerTest {

@Test
def testExpireQuotaSensors() {
val metrics = newMetrics
val clientMetrics = new ClientQuotaManager(config, metrics, Produce, time, "")
try {
maybeRecord(clientMetrics, "ANONYMOUS", "client1", 100)
Expand All @@ -420,7 +417,6 @@ class ClientQuotaManagerTest {

@Test
def testClientIdNotSanitized() {
val metrics = newMetrics
val clientMetrics = new ClientQuotaManager(config, metrics, Produce, time, "")
val clientId = "client@#$%"
try {
Expand All @@ -437,10 +433,6 @@ class ClientQuotaManagerTest {
}
}

def newMetrics: Metrics = {
new Metrics(new MetricConfig(), Collections.emptyList(), time)
}

private case class UserClient(val user: String, val clientId: String, val configUser: Option[String] = None, val configClientId: Option[String] = None) {
// The class under test expects only sanitized client configs. We pass both the default value (which should not be
// sanitized to ensure it remains unique) and non-default values, so we need to take care in generating the sanitized
Expand Down
17 changes: 9 additions & 8 deletions core/src/test/scala/unit/kafka/server/ReplicationQuotaManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,16 +23,22 @@ import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.metrics.{MetricConfig, Metrics, Quota}
import org.apache.kafka.common.utils.MockTime
import org.junit.Assert.{assertEquals, assertFalse, assertTrue}
import org.junit.Test
import org.junit.{After, Test}

import scala.collection.JavaConverters._

class ReplicationQuotaManagerTest {
private val time = new MockTime
private val metrics = new Metrics(new MetricConfig(), Collections.emptyList(), time)

@After
def tearDown: Unit = {
metrics.close()
}

@Test
def shouldThrottleOnlyDefinedReplicas(): Unit = {
val quota = new ReplicationQuotaManager(ReplicationQuotaManagerConfig(), newMetrics, QuotaType.Fetch, time)
val quota = new ReplicationQuotaManager(ReplicationQuotaManagerConfig(), metrics, QuotaType.Fetch, time)
quota.markThrottled("topic1", Seq(1, 2, 3))

assertTrue(quota.isThrottled(tp1(1)))
Expand All @@ -43,7 +49,6 @@ class ReplicationQuotaManagerTest {

@Test
def shouldExceedQuotaThenReturnBackBelowBoundAsTimePasses(): Unit = {
val metrics = newMetrics()
val quota = new ReplicationQuotaManager(ReplicationQuotaManagerConfig(numQuotaSamples = 10, quotaWindowSizeSeconds = 1), metrics, LeaderReplication, time)

//Given
Expand Down Expand Up @@ -105,7 +110,7 @@ class ReplicationQuotaManagerTest {

@Test
def shouldSupportWildcardThrottledReplicas(): Unit = {
val quota = new ReplicationQuotaManager(ReplicationQuotaManagerConfig(), newMetrics, LeaderReplication, time)
val quota = new ReplicationQuotaManager(ReplicationQuotaManagerConfig(), metrics, LeaderReplication, time)

//When
quota.markThrottled("MyTopic")
Expand All @@ -116,8 +121,4 @@ class ReplicationQuotaManagerTest {
}

private def tp1(id: Int): TopicPartition = new TopicPartition("topic1", id)

private def newMetrics(): Metrics = {
new Metrics(new MetricConfig(), Collections.emptyList(), time)
}
}