tests for min/max partitions per task metrics and minor code quality improvements by ryannedolan · Pull Request #887 · linkedin/brooklin

ryannedolan · 2022-02-26T06:53:21Z

Fixed bug where min/max partitions per task was being reported incorrectly. Previously, the min/max were the same across all datastreams.

New setGauge method in DynamicMetricsManager enables explicitly setting a gauge value. This reduces some lifecycle management complexity in LoadBalancedPartitionAssigner.

Incidental code quality improvements.

Tests added to verify affected metrics (min/maxPartitionsPerTask).

Fixed bug related to registration of Gauges and Timers (they weren't actually registered before).

Dropped (accidentally?) exported class that was holding min/max stats per task.

vmaheshw · 2022-03-01T23:31:03Z

ryanne-test-case.txt

Use the diff to try out. The test case creates 2 datastreams. For ds1: the expectation is minTasks as 2 and maxTasks as 3, for ds2: the expectation is minTasks as 1 and maxTasks as 1.

datastream-common/src/main/java/com/linkedin/datastream/metrics/DynamicMetricsManager.java

vmaheshw · 2022-03-07T23:52:34Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

-              return newTask;
-            }
+    Map<String, Set<DatastreamTask>> newAssignments = currentAssignment.entrySet().stream()
+      .collect(Collectors.toMap(x -> x.getKey(), x -> x.getValue().stream()


nit: you can replace lambda with method reference

x -> x.getKey() -> Map.Entry::getKey

vmaheshw · 2022-03-07T23:53:10Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

+        .collect(Collectors.toSet())));
+
+    IntSummaryStatistics stats = newAssignments.values().stream()
+      .flatMap(x -> x.stream()) // flatten


nit: you can replace x -> x.stream() with Collection::stream

vmaheshw · 2022-03-07T23:55:58Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

+    IntSummaryStatistics stats = newAssignments.values().stream()
+      .flatMap(x -> x.stream()) // flatten
+      .filter(x -> x.getTaskPrefix().equals(datastreamGroupName))
+      .collect(Collectors.summarizingInt(x -> x.getPartitionsV2().size()));


This code block has simplified the code with the overhead of an extra walk on the entire assignment list. I think it should be fine.

Well notice that the previous version has several side effects within a stream pipeline. The surrounding code is similarly problematic, but I have left unchanged for now, as I don't understand the code very well (probably cuz of all the side effects).

...ain/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssignmentStrategy.java

datastream-common/src/main/java/com/linkedin/datastream/metrics/DynamicMetricsManager.java

vmaheshw · 2022-03-22T21:26:32Z

datastream-common/src/main/java/com/linkedin/datastream/metrics/DynamicMetricsManager.java

    } else if (clazz.equals(Timer.class)) {
-      return (T) new Timer();
+      return (T) _metricRegistry.timer(name);
    } else {


I don't see the usage of Timer.class to be able to comment why Timer class was done in a different way. Do you have any idea?

no, but both Timer and Gauge were done this way (new T()) instead of actually registering with the underlying MetricRegistry. It seems that any attempt to register a Gauge or Timer wasn't really doing anything prior to this change.

What's the reason gauge registrations are illegal here?

vmaheshw · 2022-03-22T21:27:55Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

 import org.slf4j.LoggerFactory;

+import com.codahale.metrics.Gauge;
+


nit: empty line.

vmaheshw · 2022-03-22T21:30:28Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

-    private final int _maxPartitionsAcrossTasks;

-    public static final PartitionAssignmentStats DEFAULT = new PartitionAssignmentStats(0, 0);
+  private static class DatastreamMetrics {


Suppose in the future there is a requirement to expose methods for minPartitionsAcrossTasks and maxPartitionsAcrossTasks, will we use the gauge to return it?

no. I'd think we'd track that in the outer class and provide getters.

I do see that there is an incidental public here, so one could argue I'm accidentally dropping a public API. Concern?

I think that was one of the idea of the original code to track those values so that they can be exposed using getters. With the new code, that data structure is dropped.

Thoughts? We seem to be stuck on this issue. I was trying to minimize the lifecycle mgmt we are doing here, since this class doesn't really have any hooks into when datastreams are created or deleted etc. I think it's a bad idea to piggyback on the metrics registration/deregistration hooks that we do have.

If we need to track these values and expose via getters, I think we'd need to add more lifecycle hooks. I'm skeptical we'd do that in the interface we're impl'ing here.

Seems like we're not exposing these outside this class at the moment. I'd suggest to worry about how that is happening if there's really a need to use these metric values outside LoadBasedPartitionAssigner

vmaheshw · 2022-03-22T23:26:08Z

.../src/test/java/com/linkedin/datastream/server/assignment/TestLoadBasedPartitionAssigner.java

    Assert.assertEquals(statObj.getThroughputRateInKBps(), 5);
+
+    assertMetricEquals("LoadBasedPartitionAssigner.ds1.minPartitionsAcrossTasks", 1);
+    assertMetricEquals("LoadBasedPartitionAssigner.ds1.maxPartitionsAcrossTasks", 1);


formatting issue.

vmaheshw · 2022-03-22T23:26:30Z

.../src/test/java/com/linkedin/datastream/server/assignment/TestLoadBasedPartitionAssigner.java

+
+    assertMetricEquals("LoadBasedPartitionAssigner.ds1.minPartitionsAcrossTasks", 2);
+    assertMetricEquals("LoadBasedPartitionAssigner.ds1.maxPartitionsAcrossTasks", 2);
+
+    MetricsTestUtils.verifyMetrics(assigner, DynamicMetricsManager.getInstance());


nit: formatting issue.

vmaheshw · 2022-03-22T23:28:21Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

+    _metricsForDatastream.keySet().forEach(this::unregisterMetricsForDatastream);
  }

  void unregisterMetricsForDatastream(String datastream) {


Can you double check if there is a test to cover this code path of registration and deregistration? My guess is no, otherwise, the test would have caught issue with the older patch. Can you add a test to cover that scenario?

ryannedolan · 2022-04-05T19:16:44Z

I'll squash these commits once outstanding comments are resolved.

jzakaryan

lgtm. I think we can go ahead with the way metrics registrations are handled in LoadBasedPartitionAssigner. I also left a comment about Gauge registrations, but that shouldn't stop you from merging the code.

jzakaryan · 2022-04-11T17:15:37Z

...rver/src/main/java/com/linkedin/datastream/server/assignment/LoadBasedPartitionAssigner.java

-    private final int _maxPartitionsAcrossTasks;

-    public static final PartitionAssignmentStats DEFAULT = new PartitionAssignmentStats(0, 0);
+  private static class DatastreamMetrics {


Seems like we're not exposing these outside this class at the moment. I'd suggest to worry about how that is happening if there's really a need to use these metric values outside LoadBasedPartitionAssigner

jzakaryan · 2022-04-11T17:17:20Z

datastream-common/src/main/java/com/linkedin/datastream/metrics/DynamicMetricsManager.java

    } else if (clazz.equals(Timer.class)) {
-      return (T) new Timer();
+      return (T) _metricRegistry.timer(name);
    } else {


What's the reason gauge registrations are illegal here?

ryannedolan added 3 commits February 25, 2022 12:14

Enable expclicitly setting gauges without a supplier

a42818f

Minor cleanup of min/max partitions per task metrics and assoc. tests

b0ddc57

Fix docstrings

b6d4737

Fix misreported min/max partitions per task metric

a6091dd

vmaheshw reviewed Mar 8, 2022

View reviewed changes

ryannedolan added 2 commits March 21, 2022 17:07

fix Gauge registration bug in DynamicMetricsManager

16588c4

de/register metrics in LoadBasedPartitionAssigner

78f0bc7

vmaheshw reviewed Mar 22, 2022

View reviewed changes

Drop createOrUpdateGauge for now

83463dd

jzakaryan approved these changes Apr 11, 2022

View reviewed changes

vmaheshw approved these changes Apr 11, 2022

View reviewed changes

vmaheshw merged commit a67108b into linkedin:master Apr 12, 2022

		import org.slf4j.LoggerFactory;

		import com.codahale.metrics.Gauge;

Conversation

ryannedolan commented Feb 26, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

vmaheshw commented Mar 1, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryannedolan commented Apr 5, 2022

Uh oh!

jzakaryan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ryannedolan commented Feb 26, 2022 •

edited

Loading

vmaheshw commented Mar 1, 2022 •

edited

Loading