[FLINK-12987][metrics] fix DescriptiveStatisticsHistogram#getCount() …

…not returning the number of elements seen

DescriptiveStatisticsHistogram#getCount() only returned the number of currently
stored elements, not the total number seen.

Also add a unit test for verifying any Histogram implementation (based on
the previous test from DropwizardFlinkHistogramWrapperTest) and run this with
DescriptiveStatisticsHistogram.

flink-metrics/flink-metrics-core/pom.xml

@@ -34,6 +34,15 @@ under the License.
 
  <packaging>jar</packaging>
 
+ <dependencies>
+ <!-- ================== test dependencies ================== -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils-junit</artifactId>
+ </dependency>
+ </dependencies>
+
  <build>
  <plugins>
  <!-- activate API compatibility checks -->

flink-metrics/flink-metrics-core/src/test/java/org/apache/flink/metrics/AbstractHistogramTest.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.metrics;
+
+import org.apache.flink.util.TestLogger;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Abstract base class for testing {@link Histogram} and {@link HistogramStatistics} implementations.
+ */
+public class AbstractHistogramTest extends TestLogger {
+ protected void testHistogram(int size, Histogram histogram) {
+ HistogramStatistics statistics;
+
+ for (int i = 0; i < size; i++) {
+ histogram.update(i);
+
+ statistics = histogram.getStatistics();
+ assertEquals(i + 1, histogram.getCount());
+ assertEquals(histogram.getCount(), statistics.size());
+ assertEquals(i, statistics.getMax());
+ assertEquals(0, statistics.getMin());
+ }
+
+ statistics = histogram.getStatistics();
+ assertEquals(size, statistics.size());
+ assertEquals((size - 1) / 2.0, statistics.getQuantile(0.5), 0.001);
+
+ for (int i = size; i < 2 * size; i++) {
+ histogram.update(i);
+
+ statistics = histogram.getStatistics();
+ assertEquals(i + 1, histogram.getCount());
+ assertEquals(size, statistics.size());
+ assertEquals(i, statistics.getMax());
+ assertEquals(i + 1 - size, statistics.getMin());
+ }
+
+ statistics = histogram.getStatistics();
+ assertEquals(size, statistics.size());
+ assertEquals(size + (size - 1) / 2.0, statistics.getQuantile(0.5), 0.001);
+ }
+}

flink-metrics/flink-metrics-dropwizard/src/test/java/org/apache/flink/dropwizard/metrics/DropwizardFlinkHistogramWrapperTest.java

@@ -20,13 +20,13 @@
 
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.dropwizard.ScheduledDropwizardReporter;
+import org.apache.flink.metrics.AbstractHistogramTest;
 import org.apache.flink.metrics.MetricConfig;
 import org.apache.flink.metrics.reporter.MetricReporter;
 import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
 import org.apache.flink.runtime.metrics.MetricRegistryImpl;
 import org.apache.flink.runtime.metrics.ReporterSetup;
 import org.apache.flink.runtime.metrics.groups.TaskManagerMetricGroup;
-import org.apache.flink.util.TestLogger;
 
 import com.codahale.metrics.Counter;
 import com.codahale.metrics.Gauge;
@@ -56,7 +56,7 @@
 /**
  * Tests for the DropwizardFlinkHistogramWrapper.
  */
-public class DropwizardFlinkHistogramWrapperTest extends TestLogger {
+public class DropwizardFlinkHistogramWrapperTest extends AbstractHistogramTest {
 
  /**
  * Tests the histogram functionality of the DropwizardHistogramWrapper.
@@ -66,28 +66,7 @@ public void testDropwizardHistogramWrapper() {
  int size = 10;
  DropwizardHistogramWrapper histogramWrapper = new DropwizardHistogramWrapper(
  new com.codahale.metrics.Histogram(new SlidingWindowReservoir(size)));
-
- for (int i = 0; i < size; i++) {
- histogramWrapper.update(i);
-
- assertEquals(i + 1, histogramWrapper.getCount());
- assertEquals(i, histogramWrapper.getStatistics().getMax());
- assertEquals(0, histogramWrapper.getStatistics().getMin());
- }
-
- assertEquals(size, histogramWrapper.getStatistics().size());
- assertEquals((size - 1) / 2.0, histogramWrapper.getStatistics().getQuantile(0.5), 0.001);
-
- for (int i = size; i < 2 * size; i++) {
- histogramWrapper.update(i);
-
- assertEquals(i + 1, histogramWrapper.getCount());
- assertEquals(i, histogramWrapper.getStatistics().getMax());
- assertEquals(i + 1 - size, histogramWrapper.getStatistics().getMin());
- }
-
- assertEquals(size, histogramWrapper.getStatistics().size());
- assertEquals(size + (size - 1) / 2.0, histogramWrapper.getStatistics().getQuantile(0.5), 0.001);
+ testHistogram(size, histogramWrapper);
  }
 
  /**

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/DescriptiveStatisticsHistogram.java

@@ -29,18 +29,21 @@ public class DescriptiveStatisticsHistogram implements org.apache.flink.metrics.
 
  private final DescriptiveStatistics descriptiveStatistics;
 
+ private long elementsSeen = 0L;
+
  public DescriptiveStatisticsHistogram(int windowSize) {
  this.descriptiveStatistics = new DescriptiveStatistics(windowSize);
  }
 
  @Override
  public void update(long value) {
+ elementsSeen += 1L;
  this.descriptiveStatistics.addValue(value);
  }
 
  @Override
  public long getCount() {
- return this.descriptiveStatistics.getN();
+ return this.elementsSeen;
  }
 
  @Override

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/DescriptiveStatisticsHistogramTest.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.metrics;
+
+import org.apache.flink.metrics.AbstractHistogramTest;
+
+import org.junit.Test;
+
+/**
+ * Tests for {@link DescriptiveStatisticsHistogram} and
+ * {@link DescriptiveStatisticsHistogramStatistics}.
+ */
+public class DescriptiveStatisticsHistogramTest extends AbstractHistogramTest {
+
+ /**
+ * Tests the histogram functionality of the DropwizardHistogramWrapper.
+ */
+ @Test
+ public void testDescriptiveHistogram() {
+ int size = 10;
+ testHistogram(size, new DescriptiveStatisticsHistogram(size));
+ }
+
+}