Simply and enhance the RebalanceLatencyGauge so it can be used in mul…

…ti-threads. (#636)

The previous design of RebalanceLatencyGauge won't support asynchronous metric data emitting. This PR adds support by using a ThreadLocal object.
The metric logic is not changed.

helix-core/src/main/java/org/apache/helix/monitoring/metrics/implementation/RebalanceLatencyGauge.java

@@ -19,9 +19,10 @@
  * under the License.
  */
 
+import java.util.concurrent.TimeUnit;
+
 import com.codahale.metrics.Histogram;
 import com.codahale.metrics.SlidingTimeWindowArrayReservoir;
-import java.util.concurrent.TimeUnit;
 import org.apache.helix.monitoring.metrics.model.LatencyMetric;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,6 +31,8 @@ public class RebalanceLatencyGauge extends LatencyMetric {
   private static final Logger LOG = LoggerFactory.getLogger(RebalanceLatencyGauge.class);
   private static final long VALUE_NOT_SET = -1;
   private long _lastEmittedMetricValue = VALUE_NOT_SET;
+  // Use threadlocal here so the start time can be updated and recorded in multi-threads.
+  private final ThreadLocal<Long> _startTime;
 
   /**
    * Instantiates a new Histogram dynamic metric.
@@ -39,35 +42,32 @@ public RebalanceLatencyGauge(String metricName, long slidingTimeWindow) {
     super(metricName, new Histogram(
         new SlidingTimeWindowArrayReservoir(slidingTimeWindow, TimeUnit.MILLISECONDS)));
     _metricName = metricName;
+    _startTime = ThreadLocal.withInitial(() -> VALUE_NOT_SET);
   }
 
   /**
-   * WARNING: this method is not thread-safe.
    * Calling this method multiple times would simply overwrite the previous state. This is because
    * the rebalancer could fail at any point, and we want it to recover gracefully by resetting the
    * internal state of this metric.
    */
   @Override
   public void startMeasuringLatency() {
     reset();
-    _startTime = System.currentTimeMillis();
+    _startTime.set(System.currentTimeMillis());
   }
 
-  /**
-   * WARNING: this method is not thread-safe.
-   */
   @Override
   public void endMeasuringLatency() {
-    if (_startTime == VALUE_NOT_SET || _endTime != VALUE_NOT_SET) {
+    if (_startTime.get() == VALUE_NOT_SET) {
       LOG.error(
           "Needs to call startMeasuringLatency first! Ignoring and resetting the metric. Metric name: {}",
           _metricName);
-      reset();
       return;
     }
-    _endTime = System.currentTimeMillis();
-    _lastEmittedMetricValue = _endTime - _startTime;
-    updateValue(_lastEmittedMetricValue);
+    synchronized (this) {
+      _lastEmittedMetricValue = System.currentTimeMillis() - _startTime.get();
+      updateValue(_lastEmittedMetricValue);
+    }
     reset();
   }
 
@@ -84,7 +84,6 @@ public Long getLastEmittedMetricValue() {
    * Resets the internal state of this metric.
    */
   private void reset() {
-    _startTime = VALUE_NOT_SET;
-    _endTime = VALUE_NOT_SET;
+    _startTime.set(VALUE_NOT_SET);
   }
 }

helix-core/src/main/java/org/apache/helix/monitoring/metrics/model/LatencyMetric.java

@@ -28,8 +28,6 @@
  * how long a particular stage in the logic took in milliseconds.
  */
 public abstract class LatencyMetric extends HistogramDynamicMetric implements Metric<Long> {
-  protected long _startTime;
-  protected long _endTime;
   protected String _metricName;
 
   /**