Fix calculation of age of pending tasks

This commit addresses a time unit conversion bug in calculating the age
of a PrioritizedRunnable. The issue was an incorrect conversion from
nanoseconds to milliseconds as instead the conversion was to
microseconds. This leads to the timeInQueue metric for pending tasks to
be off by three orders of magnitude.

Author: jasontedor

core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedRunnable.java

@@ -20,29 +20,39 @@
 
 import org.elasticsearch.common.Priority;
 
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+
 /**
  *
  */
 public abstract class PrioritizedRunnable implements Runnable, Comparable<PrioritizedRunnable> {
 
     private final Priority priority;
     private final long creationDate;
+    private final LongSupplier relativeTimeProvider;
 
     public static PrioritizedRunnable wrap(Runnable runnable, Priority priority) {
         return new Wrapped(runnable, priority);
     }
 
     protected PrioritizedRunnable(Priority priority) {
+        this(priority, System::nanoTime);
+    }
+
+    // package visible for testing
+    PrioritizedRunnable(Priority priority, LongSupplier relativeTimeProvider) {
         this.priority = priority;
-        creationDate = System.nanoTime();
+        this.creationDate = relativeTimeProvider.getAsLong();
+        this.relativeTimeProvider = relativeTimeProvider;
     }
 
     public long getCreationDateInNanos() {
         return creationDate;
     }
 
     public long getAgeInMillis() {
-        return Math.max(0, (System.nanoTime() - creationDate) / 1000);
+        return TimeUnit.MILLISECONDS.convert(relativeTimeProvider.getAsLong() - creationDate, TimeUnit.NANOSECONDS);
     }
 
     @Override

core/src/test/java/org/elasticsearch/common/util/concurrent/PrioritizedRunnableTests.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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://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.elasticsearch.common.util.concurrent;
+
+import org.elasticsearch.common.Priority;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class PrioritizedRunnableTests extends ESTestCase {
+    public void testGetAgeInMillis() throws Exception {
+        AtomicLong time = new AtomicLong();
+
+        PrioritizedRunnable runnable = new PrioritizedRunnable(Priority.NORMAL, time::get) {
+            @Override
+            public void run() {
+
+            }
+        };
+        assertEquals(0, runnable.getAgeInMillis());
+        int milliseconds = randomIntBetween(1, 256);
+        time.addAndGet(TimeUnit.NANOSECONDS.convert(milliseconds, TimeUnit.MILLISECONDS));
+        assertEquals(milliseconds, runnable.getAgeInMillis());
+    }
+}