Fix calculation of age of pending tasks

This commit backports commit aa28133
from master to 2.2.

Relates #15995

Author: jasontedor

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

@@ -19,31 +19,51 @@
 package org.elasticsearch.common.util.concurrent;
 
 import org.elasticsearch.common.Priority;
-import org.elasticsearch.common.unit.TimeValue;
+
+import java.util.concurrent.TimeUnit;
 
 /**
  *
  */
 public abstract class PrioritizedRunnable implements Runnable, Comparable<PrioritizedRunnable> {
 
+    // package visible for testing
+    interface LongSupplier {
+        long getAsLong();
+    }
+
     private final Priority priority;
     private final long creationDate;
+    private final LongSupplier relativeTimeProvider;
+
+    private static final LongSupplier SYSTEM_NANO_TIME = new LongSupplier() {
+        @Override
+        public long getAsLong() {
+            return System.nanoTime();
+        }
+    };
 
     public static PrioritizedRunnable wrap(Runnable runnable, Priority priority) {
         return new Wrapped(runnable, priority);
     }
 
     protected PrioritizedRunnable(Priority priority) {
+        this(priority, SYSTEM_NANO_TIME);
+    }
+
+    // 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
@@ -69,4 +89,4 @@ public void run() {
             runnable.run();
         }
     }
-}
+}

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

@@ -0,0 +1,50 @@
+/*
+ * 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;
+
+import static org.hamcrest.Matchers.greaterThan;
+
+public class PrioritizedRunnableTests extends ESTestCase {
+    public void testGetAgeInMillis() throws Exception {
+        final AtomicLong time = new AtomicLong();
+
+        PrioritizedRunnable runnable = new PrioritizedRunnable(Priority.NORMAL, new PrioritizedRunnable.LongSupplier() {
+            @Override
+            public long getAsLong() {
+                return 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());
+    }
+}