diff --git a/server/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java b/server/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java
index 2da0ff9286f..3e61c203fd0 100644
--- a/server/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java
+++ b/server/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java
@@ -26,6 +26,7 @@ import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.threadpool.ThreadPool;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
@@ -44,6 +45,7 @@ public class ClusterStateObserver {
private final Predicate<ClusterState> MATCH_ALL_CHANGES_PREDICATE = state -> true;
private final ClusterApplierService clusterApplierService;
+ private final ThreadPool threadPool;
private final ThreadContext contextHolder;
volatile TimeValue timeOutValue;
@@ -52,7 +54,7 @@ public class ClusterStateObserver {
final TimeoutClusterStateListener clusterStateListener = new ObserverClusterStateListener();
// observingContext is not null when waiting on cluster state changes
final AtomicReference<ObservingContext> observingContext = new AtomicReference<>(null);
- volatile Long startTimeNS;
+ volatile Long startTimeMS;
volatile boolean timedOut;
@@ -81,10 +83,11 @@ public class ClusterStateObserver {
public ClusterStateObserver(ClusterState initialState, ClusterApplierService clusterApplierService, @Nullable TimeValue timeout,
Logger logger, ThreadContext contextHolder) {
this.clusterApplierService = clusterApplierService;
+ this.threadPool = clusterApplierService.threadPool();
this.lastObservedState = new AtomicReference<>(new StoredState(initialState));
this.timeOutValue = timeout;
if (timeOutValue != null) {
- this.startTimeNS = System.nanoTime();
+ this.startTimeMS = threadPool.relativeTimeInMillis();
}
this.logger = logger;
this.contextHolder = contextHolder;
@@ -134,7 +137,7 @@ public class ClusterStateObserver {
if (timeOutValue == null) {
timeOutValue = this.timeOutValue;
if (timeOutValue != null) {
- long timeSinceStartMS = TimeValue.nsecToMSec(System.nanoTime() - startTimeNS);
+ long timeSinceStartMS = threadPool.relativeTimeInMillis() - startTimeMS;
timeoutTimeLeftMS = timeOutValue.millis() - timeSinceStartMS;
if (timeoutTimeLeftMS <= 0L) {
// things have timeout while we were busy -> notify
@@ -150,7 +153,7 @@ public class ClusterStateObserver {
timeoutTimeLeftMS = null;
}
} else {
- this.startTimeNS = System.nanoTime();
+ this.startTimeMS = threadPool.relativeTimeInMillis();
this.timeOutValue = timeOutValue;
timeoutTimeLeftMS = timeOutValue.millis();
timedOut = false;
@@ -240,7 +243,7 @@ public class ClusterStateObserver {
ObservingContext context = observingContext.getAndSet(null);
if (context != null) {
clusterApplierService.removeTimeoutListener(this);
- long timeSinceStartMS = TimeValue.nsecToMSec(System.nanoTime() - startTimeNS);
+ long timeSinceStartMS = threadPool.relativeTimeInMillis() - startTimeMS;
logger.trace("observer: timeout notification from cluster service. timeout setting [{}], time since start [{}]",
timeOutValue, new TimeValue(timeSinceStartMS));
// update to latest, in case people want to retry
diff --git a/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java b/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java
index e254196caa4..f2b0756d3d8 100644
--- a/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java
@@ -311,6 +311,10 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
runOnApplierThread(source, clusterStateConsumer, listener, Priority.HIGH);
}
+ public ThreadPool threadPool() {
+ return threadPool;
+ }
+
@Override
public void onNewClusterState(final String source, final Supplier<ClusterState> clusterStateSupplier,
final ClusterApplyListener listener) {
@@ -383,12 +387,12 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
logger.debug("processing [{}]: execute", task.source);
final ClusterState previousClusterState = state.get();
- long startTimeNS = currentTimeInNanos();
+ long startTimeMS = currentTimeInMillis();
final ClusterState newClusterState;
try {
newClusterState = task.apply(previousClusterState);
} catch (Exception e) {
- TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, TimeValue.nsecToMSec(currentTimeInNanos() - startTimeNS)));
+ TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, currentTimeInMillis() - startTimeMS));
logger.trace(() -> new ParameterizedMessage(
"failed to execute cluster state applier in [{}], state:\nversion [{}], source [{}]\n{}",
executionTime, previousClusterState.version(), task.source, previousClusterState), e);
@@ -398,7 +402,7 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
}
if (previousClusterState == newClusterState) {
- TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, TimeValue.nsecToMSec(currentTimeInNanos() - startTimeNS)));
+ TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, currentTimeInMillis() - startTimeMS));
logger.debug("processing [{}]: took [{}] no change in cluster state", task.source, executionTime);
warnAboutSlowTaskIfNeeded(executionTime, task.source);
task.listener.onSuccess(task.source);
@@ -411,14 +415,14 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
}
try {
applyChanges(task, previousClusterState, newClusterState);
- TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, TimeValue.nsecToMSec(currentTimeInNanos() - startTimeNS)));
+ TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, currentTimeInMillis() - startTimeMS));
logger.debug("processing [{}]: took [{}] done applying updated cluster state (version: {}, uuid: {})", task.source,
executionTime, newClusterState.version(),
newClusterState.stateUUID());
warnAboutSlowTaskIfNeeded(executionTime, task.source);
task.listener.onSuccess(task.source);
} catch (Exception e) {
- TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, TimeValue.nsecToMSec(currentTimeInNanos() - startTimeNS)));
+ TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, currentTimeInMillis() - startTimeMS));
if (logger.isTraceEnabled()) {
logger.warn(new ParameterizedMessage(
"failed to apply updated cluster state in [{}]:\nversion [{}], uuid [{}], source [{}]\n{}",
@@ -617,8 +621,8 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
}
// this one is overridden in tests so we can control time
- protected long currentTimeInNanos() {
- return System.nanoTime();
+ protected long currentTimeInMillis() {
+ return threadPool.relativeTimeInMillis();
}
}
diff --git a/server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java
index 7f91513e343..740283736a2 100644
--- a/server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java
@@ -88,7 +88,7 @@ public class ClusterApplierServiceTests extends ESTestCase {
super.tearDown();
}
- TimedClusterApplierService createTimedClusterService(boolean makeMaster) throws InterruptedException {
+ TimedClusterApplierService createTimedClusterService(boolean makeMaster) {
DiscoveryNode localNode = new DiscoveryNode("node1", buildNewFakeTransportAddress(), emptyMap(),
emptySet(), Version.CURRENT);
TimedClusterApplierService timedClusterApplierService = new TimedClusterApplierService(Settings.builder().put("cluster.name",
@@ -141,9 +141,9 @@ public class ClusterApplierServiceTests extends ESTestCase {
Logger clusterLogger = LogManager.getLogger(ClusterApplierService.class);
Loggers.addAppender(clusterLogger, mockAppender);
try {
- clusterApplierService.currentTimeOverride = System.nanoTime();
+ clusterApplierService.currentTimeOverride = threadPool.relativeTimeInMillis();
clusterApplierService.runOnApplierThread("test1",
- currentState -> clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(1).nanos(),
+ currentState -> clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(1).millis(),
new ClusterApplyListener() {
@Override
public void onSuccess(String source) { }
@@ -155,7 +155,7 @@ public class ClusterApplierServiceTests extends ESTestCase {
});
clusterApplierService.runOnApplierThread("test2",
currentState -> {
- clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(2).nanos();
+ clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(2).millis();
throw new IllegalArgumentException("Testing handling of exceptions in the cluster state task");
},
new ClusterApplyListener() {
@@ -214,9 +214,9 @@ public class ClusterApplierServiceTests extends ESTestCase {
try {
final CountDownLatch latch = new CountDownLatch(4);
final CountDownLatch processedFirstTask = new CountDownLatch(1);
- clusterApplierService.currentTimeOverride = System.nanoTime();
+ clusterApplierService.currentTimeOverride = threadPool.relativeTimeInMillis();
clusterApplierService.runOnApplierThread("test1",
- currentState -> clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(1).nanos(),
+ currentState -> clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(1).millis(),
new ClusterApplyListener() {
@Override
public void onSuccess(String source) {
@@ -232,7 +232,7 @@ public class ClusterApplierServiceTests extends ESTestCase {
processedFirstTask.await();
clusterApplierService.runOnApplierThread("test2",
currentState -> {
- clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(32).nanos();
+ clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(32).millis();
throw new IllegalArgumentException("Testing handling of exceptions in the cluster state task");
},
new ClusterApplyListener() {
@@ -247,7 +247,7 @@ public class ClusterApplierServiceTests extends ESTestCase {
}
});
clusterApplierService.runOnApplierThread("test3",
- currentState -> clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(34).nanos(),
+ currentState -> clusterApplierService.currentTimeOverride += TimeValue.timeValueSeconds(34).millis(),
new ClusterApplyListener() {
@Override
public void onSuccess(String source) {
@@ -510,11 +510,11 @@ public class ClusterApplierServiceTests extends ESTestCase {
}
@Override
- protected long currentTimeInNanos() {
+ protected long currentTimeInMillis() {
if (currentTimeOverride != null) {
return currentTimeOverride;
}
- return super.currentTimeInNanos();
+ return super.currentTimeInMillis();
}
}
}