Skip to content

Commit

Permalink
Revert "Apply cluster states in system context (#53785)" (#53842)
Browse files Browse the repository at this point in the history 
This reverts commit c1dc523.
  • Loading branch information
@DaveCTurner
DaveCTurner authored Mar 20, 2020
1 parent 7a1502d commit 76cd638
Show file tree
Hide file tree
Showing 11 changed files with 40 additions and 103 deletions.
2 changes: 0 additions & 2 deletions server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,6 @@ public PublicationContext newPublicationContext(ClusterChangedEvent clusterChang
public void sendPublishRequest(DiscoveryNode destination, PublishRequest publishRequest,
ActionListener<PublishWithJoinResponse> originalListener) {
assert publishRequest.getAcceptedState() == clusterChangedEvent.state() : "state got switched on us";
assert transportService.getThreadPool().getThreadContext().isSystemContext();
final ActionListener<PublishWithJoinResponse> responseActionListener;
if (destination.equals(nodes.getLocalNode())) {
// if publishing to self, use original request instead (see currentPublishRequestToSelf for explanation)
Expand Down Expand Up @@ -198,7 +197,6 @@ public void onFailure(Exception e) {
@Override
public void sendApplyCommit(DiscoveryNode destination, ApplyCommitRequest applyCommitRequest,
ActionListener<TransportResponse.Empty> responseActionListener) {
assert transportService.getThreadPool().getThreadContext().isSystemContext();
transportService.sendRequest(destination, COMMIT_STATE_ACTION_NAME, applyCommitRequest, stateRequestOptions,
new TransportResponseHandler<TransportResponse.Empty>() {

Expand Down
5 changes: 1 addition & 4 deletions server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,6 @@
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.common.util.iterable.Iterables;
import org.elasticsearch.threadpool.Scheduler;
import org.elasticsearch.threadpool.ThreadPool;
Expand Down Expand Up @@ -348,9 +347,7 @@ private void submitStateUpdateTask(final String source, final ClusterStateTaskCo
if (!lifecycle.started()) {
return;
}
final ThreadContext threadContext = threadPool.getThreadContext();
try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
threadContext.markAsSystemContext();
try {
UpdateTask updateTask = new UpdateTask(config.priority(), source, new SafeClusterApplyListener(listener, logger), executor);
if (config.timeout() != null) {
threadPoolExecutor.execute(updateTask, config.timeout(),
Expand Down
5 changes: 2 additions & 3 deletions server/src/test/java/org/elasticsearch/cluster/coordination/NodeJoinTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,10 +122,9 @@ private static ClusterState initialState(DiscoveryNode localNode, long term, lon
private void setupFakeMasterServiceAndCoordinator(long term, ClusterState initialState) {
deterministicTaskQueue
= new DeterministicTaskQueue(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "test").build(), random());
final ThreadPool fakeThreadPool = deterministicTaskQueue.getThreadPool();
FakeThreadPoolMasterService fakeMasterService = new FakeThreadPoolMasterService("test_node","test",
fakeThreadPool, deterministicTaskQueue::scheduleNow);
setupMasterServiceAndCoordinator(term, initialState, fakeMasterService, fakeThreadPool, Randomness.get());
deterministicTaskQueue::scheduleNow);
setupMasterServiceAndCoordinator(term, initialState, fakeMasterService, deterministicTaskQueue.getThreadPool(), Randomness.get());
fakeMasterService.setClusterStatePublisher((event, publishListener, ackListener) -> {
coordinator.handlePublishRequest(new PublishRequest(event.state()));
publishListener.onResponse(null);
Expand Down
8 changes: 4 additions & 4 deletions server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1163,15 +1163,15 @@ private final class TestClusterNode {
TestClusterNode(DiscoveryNode node) throws IOException {
this.node = node;
final Environment environment = createEnvironment(node.getName());
threadPool = deterministicTaskQueue.getThreadPool(runnable -> CoordinatorTests.onNodeLog(node, runnable));
masterService = new FakeThreadPoolMasterService(node.getName(), "test", threadPool, deterministicTaskQueue::scheduleNow);
masterService = new FakeThreadPoolMasterService(node.getName(), "test", deterministicTaskQueue::scheduleNow);
final Settings settings = environment.settings();
final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
threadPool = deterministicTaskQueue.getThreadPool();
clusterService = new ClusterService(settings, clusterSettings, masterService,
new ClusterApplierService(node.getName(), settings, clusterSettings, threadPool) {
@Override
protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
return new MockSinglePrioritizingExecutor(node.getName(), deterministicTaskQueue, threadPool);
return new MockSinglePrioritizingExecutor(node.getName(), deterministicTaskQueue);
}

@Override
Expand Down Expand Up @@ -1211,7 +1211,7 @@ protected NamedWriteableRegistry writeableRegistry() {
}
};
transportService = mockTransport.createTransportService(
settings, threadPool,
settings, deterministicTaskQueue.getThreadPool(runnable -> CoordinatorTests.onNodeLog(node, runnable)),
new TransportInterceptor() {
@Override
public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(String action, String executor,
Expand Down
30 changes: 16 additions & 14 deletions ...ork/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -911,7 +911,6 @@ class ClusterNode {
}

private void setUp() {
final ThreadPool threadPool = deterministicTaskQueue.getThreadPool(this::onNode);
mockTransport = new DisruptableMockTransport(localNode, logger) {
@Override
protected void execute(Runnable runnable) {
Expand All @@ -929,20 +928,24 @@ protected Optional<DisruptableMockTransport> getDisruptableMockTransport(Transpo
.filter(transport -> transport.getLocalNode().getAddress().equals(address)).findAny();
}
};

final Settings settings = nodeSettings.hasValue(DiscoveryModule.DISCOVERY_TYPE_SETTING.getKey()) ?
nodeSettings : Settings.builder().put(nodeSettings)
.putList(ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING.getKey(),
ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING.get(Settings.EMPTY)).build(); // suppress auto-bootstrap
transportService = mockTransport.createTransportService(settings, threadPool,
getTransportInterceptor(localNode, threadPool), a -> localNode, null, emptySet());
masterService = new AckedFakeThreadPoolMasterService(localNode.getId(), "test", threadPool,
transportService = mockTransport.createTransportService(
settings, deterministicTaskQueue.getThreadPool(this::onNode),
getTransportInterceptor(localNode, deterministicTaskQueue.getThreadPool(this::onNode)),
a -> localNode, null, emptySet());
masterService = new AckedFakeThreadPoolMasterService(localNode.getId(), "test",
runnable -> deterministicTaskQueue.scheduleNow(onNode(runnable)));
final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
clusterApplierService = new DisruptableClusterApplierService(localNode.getId(), settings, clusterSettings,
deterministicTaskQueue, threadPool);
deterministicTaskQueue, this::onNode);
clusterService = new ClusterService(settings, clusterSettings, masterService, clusterApplierService);
clusterService.setNodeConnectionsService(
new NodeConnectionsService(clusterService.getSettings(), threadPool, transportService));
new NodeConnectionsService(clusterService.getSettings(), deterministicTaskQueue.getThreadPool(this::onNode),
transportService));
final Collection<BiConsumer<DiscoveryNode, ClusterState>> onJoinValidators =
Collections.singletonList((dn, cs) -> extraJoinValidators.forEach(validator -> validator.accept(dn, cs)));
final AllocationService allocationService = ESAllocationTestCase.createAllocationService(Settings.EMPTY);
Expand All @@ -951,8 +954,8 @@ protected Optional<DisruptableMockTransport> getDisruptableMockTransport(Transpo
Cluster.this::provideSeedHosts, clusterApplierService, onJoinValidators, Randomness.get(), (s, p, r) -> {},
getElectionStrategy());
masterService.setClusterStatePublisher(coordinator);
final GatewayService gatewayService
= new GatewayService(settings, allocationService, clusterService, threadPool, coordinator, null);
final GatewayService gatewayService = new GatewayService(settings, allocationService, clusterService,
deterministicTaskQueue.getThreadPool(this::onNode), coordinator, null);

logger.trace("starting up [{}]", localNode);
transportService.start();
Expand Down Expand Up @@ -1289,9 +1292,8 @@ static class AckedFakeThreadPoolMasterService extends FakeThreadPoolMasterServic

AckCollector nextAckCollector = new AckCollector();

AckedFakeThreadPoolMasterService(String nodeName, String serviceName, ThreadPool threadPool,
Consumer<Runnable> onTaskAvailableToRun) {
super(nodeName, serviceName, threadPool, onTaskAvailableToRun);
AckedFakeThreadPoolMasterService(String nodeName, String serviceName, Consumer<Runnable> onTaskAvailableToRun) {
super(nodeName, serviceName, onTaskAvailableToRun);
}

@Override
Expand Down Expand Up @@ -1321,8 +1323,8 @@ static class DisruptableClusterApplierService extends ClusterApplierService {
private boolean applicationMayFail;

DisruptableClusterApplierService(String nodeName, Settings settings, ClusterSettings clusterSettings,
DeterministicTaskQueue deterministicTaskQueue, ThreadPool threadPool) {
super(nodeName, settings, clusterSettings, threadPool);
DeterministicTaskQueue deterministicTaskQueue, Function<Runnable, Runnable> runnableWrapper) {
super(nodeName, settings, clusterSettings, deterministicTaskQueue.getThreadPool(runnableWrapper));
this.nodeName = nodeName;
this.deterministicTaskQueue = deterministicTaskQueue;
addStateApplier(event -> {
Expand All @@ -1342,7 +1344,7 @@ static class DisruptableClusterApplierService extends ClusterApplierService {

@Override
protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
return new MockSinglePrioritizingExecutor(nodeName, deterministicTaskQueue, threadPool);
return new MockSinglePrioritizingExecutor(nodeName, deterministicTaskQueue);
}

@Override
Expand Down
5 changes: 2 additions & 3 deletions .../src/main/java/org/elasticsearch/cluster/coordination/MockSinglePrioritizingExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@

import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
import org.elasticsearch.threadpool.ThreadPool;

import java.util.concurrent.TimeUnit;

Expand All @@ -30,7 +29,7 @@
*/
public class MockSinglePrioritizingExecutor extends PrioritizedEsThreadPoolExecutor {

public MockSinglePrioritizingExecutor(String name, DeterministicTaskQueue deterministicTaskQueue, ThreadPool threadPool) {
public MockSinglePrioritizingExecutor(String name, DeterministicTaskQueue deterministicTaskQueue) {
super(name, 0, 1, 0L, TimeUnit.MILLISECONDS,
r -> new Thread() {
@Override
Expand All @@ -52,7 +51,7 @@ public String toString() {
});
}
},
threadPool.getThreadContext(), threadPool.scheduler());
deterministicTaskQueue.getThreadPool().getThreadContext(), deterministicTaskQueue.getThreadPool().scheduler());
}

@Override
Expand Down
50 changes: 0 additions & 50 deletions ...mework/src/main/java/org/elasticsearch/cluster/service/ClusterApplierAssertionPlugin.java
View file

This file was deleted.

22 changes: 13 additions & 9 deletions ...ramework/src/main/java/org/elasticsearch/cluster/service/FakeThreadPoolMasterService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@

import static org.apache.lucene.util.LuceneTestCase.random;
import static org.elasticsearch.test.ESTestCase.randomInt;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public class FakeThreadPoolMasterService extends MasterService {
private static final Logger logger = LogManager.getLogger(FakeThreadPoolMasterService.class);
Expand All @@ -52,14 +54,21 @@ public class FakeThreadPoolMasterService extends MasterService {
private boolean taskInProgress = false;
private boolean waitForPublish = false;

public FakeThreadPoolMasterService(String nodeName, String serviceName, ThreadPool threadPool,
Consumer<Runnable> onTaskAvailableToRun) {
public FakeThreadPoolMasterService(String nodeName, String serviceName, Consumer<Runnable> onTaskAvailableToRun) {
super(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), nodeName).build(),
new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), threadPool);
new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
createMockThreadPool());
this.name = serviceName;
this.onTaskAvailableToRun = onTaskAvailableToRun;
}

private static ThreadPool createMockThreadPool() {
final ThreadContext context = new ThreadContext(Settings.EMPTY);
final ThreadPool mockThreadPool = mock(ThreadPool.class);
when(mockThreadPool.getThreadContext()).thenReturn(context);
return mockThreadPool;
}

@Override
protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
return new PrioritizedEsThreadPoolExecutor(name, 1, 1, 1, TimeUnit.SECONDS, EsExecutors.daemonThreadFactory(name),
Expand Down Expand Up @@ -101,11 +110,7 @@ public void run() {
final Runnable task = pendingTasks.remove(taskIndex);
taskInProgress = true;
scheduledNextTask = false;
final ThreadContext threadContext = threadPool.getThreadContext();
try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
threadContext.markAsSystemContext();
task.run();
}
task.run();
if (waitForPublish == false) {
taskInProgress = false;
}
Expand Down Expand Up @@ -163,5 +168,4 @@ public void onFailure(Exception e) {
protected AckListener wrapAckListener(AckListener ackListener) {
return ackListener;
}

}
4 changes: 0 additions & 4 deletions test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,6 @@
import org.elasticsearch.cluster.routing.UnassignedInfo;
import org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings;
import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
import org.elasticsearch.cluster.service.ClusterApplierAssertionPlugin;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Priority;
Expand Down Expand Up @@ -1785,9 +1784,6 @@ protected Collection<Class<? extends Plugin>> getMockPlugins() {
if (randomBoolean()) {
mocks.add(MockFieldFilterPlugin.class);
}
if (randomBoolean()) {
mocks.add(ClusterApplierAssertionPlugin.class);
}
}

if (addMockTransportService()) {
Expand Down
2 changes: 1 addition & 1 deletion ...test/java/org/elasticsearch/cluster/coordination/MockSinglePrioritizingExecutorTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ public class MockSinglePrioritizingExecutorTests extends ESTestCase {

public void testPrioritizedEsThreadPoolExecutor() {
final DeterministicTaskQueue taskQueue = DeterministicTaskQueueTests.newTaskQueue();
final PrioritizedEsThreadPoolExecutor executor = new MockSinglePrioritizingExecutor("test", taskQueue, taskQueue.getThreadPool());
final PrioritizedEsThreadPoolExecutor executor = new MockSinglePrioritizingExecutor("test", taskQueue);
final AtomicBoolean called1 = new AtomicBoolean();
final AtomicBoolean called2 = new AtomicBoolean();
executor.execute(new PrioritizedRunnable(Priority.NORMAL) {
Expand Down
Loading

0 comments on commit 76cd638

Please sign in to comment.