honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix queuing in AsyncLucenePersistedState (#50958) 

The logic in AsyncLucenePersistedState was flawed, unexpectedly queuing up two update tasks
in parallel.
This commit is contained in:
Yannick Welsch 2020-01-14 14:08:31 +01:00
parent 91d7b446a0
commit f1c5031766
1 changed files with 46 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java
View File

@ -46,7 +46,6 @@ import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.AbstractRunnable; import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.common.util.concurrent.EsExecutors; import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor; import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
import org.elasticsearch.core.internal.io.IOUtils; import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.discovery.DiscoveryModule; import org.elasticsearch.discovery.DiscoveryModule;
@ -356,10 +355,12 @@ public class GatewayMetaState implements Closeable {
} else { } else {
logger.trace("queuing term update (setting term to {})", currentTerm); logger.trace("queuing term update (setting term to {})", currentTerm);
newCurrentTermQueued = true; newCurrentTermQueued = true;
if (newStateQueued == false) {
scheduleUpdate(); scheduleUpdate();
} }
} }
} }
}
@Override @Override
public void setLastAcceptedState(ClusterState clusterState) { public void setLastAcceptedState(ClusterState clusterState) {
@ -370,14 +371,16 @@ public class GatewayMetaState implements Closeable {
} else { } else {
logger.trace("queuing cluster state update (setting cluster state to {})", clusterState.version()); logger.trace("queuing cluster state update (setting cluster state to {})", clusterState.version());
newStateQueued = true; newStateQueued = true;
if (newCurrentTermQueued == false) {
scheduleUpdate(); scheduleUpdate();
} }
} }
} }
}
private void scheduleUpdate() { private void scheduleUpdate() {
assert Thread.holdsLock(mutex); assert Thread.holdsLock(mutex);
try { assert threadPoolExecutor.getQueue().isEmpty() : "threadPoolExecutor queue not empty";
threadPoolExecutor.execute(new AbstractRunnable() { threadPoolExecutor.execute(new AbstractRunnable() {
@Override @Override
@ -385,6 +388,11 @@ public class GatewayMetaState implements Closeable {
logger.error("Exception occurred when storing new meta data", e); logger.error("Exception occurred when storing new meta data", e);
} }
@Override
public void onRejection(Exception e) {
assert threadPoolExecutor.isShutdown() : "only expect rejections when shutting down";
}
@Override @Override
protected void doRun() { protected void doRun() {
final Long term; final Long term;
@ -392,12 +400,14 @@ public class GatewayMetaState implements Closeable {
synchronized (mutex) { synchronized (mutex) {
if (newCurrentTermQueued) { if (newCurrentTermQueued) {
term = getCurrentTerm(); term = getCurrentTerm();
logger.trace("resetting newCurrentTermQueued");
newCurrentTermQueued = false; newCurrentTermQueued = false;
} else { } else {
term = null; term = null;
} }
if (newStateQueued) { if (newStateQueued) {
clusterState = getLastAcceptedState(); clusterState = getLastAcceptedState();
logger.trace("resetting newStateQueued");
newStateQueued = false; newStateQueued = false;
} else { } else {
clusterState = null; clusterState = null;
@ -412,13 +422,6 @@ public class GatewayMetaState implements Closeable {
} }
} }
}); });
} catch (EsRejectedExecutionException e) {
// ignore cases where we are shutting down..., there is really nothing interesting to be done here...
if (threadPoolExecutor.isShutdown() == false) {
assert false : "only expect rejections when shutting down";
throw e;
}
}
} }
static final CoordinationMetaData.VotingConfiguration staleStateConfiguration = static final CoordinationMetaData.VotingConfiguration staleStateConfiguration =