|
@@ -911,6 +911,7 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
|
|
|
}
|
|
|
|
|
|
private void setUp() {
|
|
|
+ final ThreadPool threadPool = deterministicTaskQueue.getThreadPool(this::onNode);
|
|
|
mockTransport = new DisruptableMockTransport(localNode, logger) {
|
|
|
@Override
|
|
|
protected void execute(Runnable runnable) {
|
|
@@ -928,24 +929,20 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
|
|
|
.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, deterministicTaskQueue.getThreadPool(this::onNode),
|
|
|
- getTransportInterceptor(localNode, deterministicTaskQueue.getThreadPool(this::onNode)),
|
|
|
- a -> localNode, null, emptySet());
|
|
|
- masterService = new AckedFakeThreadPoolMasterService(localNode.getId(), "test",
|
|
|
+ transportService = mockTransport.createTransportService(settings, threadPool,
|
|
|
+ getTransportInterceptor(localNode, threadPool), a -> localNode, null, emptySet());
|
|
|
+ masterService = new AckedFakeThreadPoolMasterService(localNode.getId(), "test", threadPool,
|
|
|
runnable -> deterministicTaskQueue.scheduleNow(onNode(runnable)));
|
|
|
final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
|
|
|
clusterApplierService = new DisruptableClusterApplierService(localNode.getId(), settings, clusterSettings,
|
|
|
- deterministicTaskQueue, this::onNode);
|
|
|
+ deterministicTaskQueue, threadPool);
|
|
|
clusterService = new ClusterService(settings, clusterSettings, masterService, clusterApplierService);
|
|
|
clusterService.setNodeConnectionsService(
|
|
|
- new NodeConnectionsService(clusterService.getSettings(), deterministicTaskQueue.getThreadPool(this::onNode),
|
|
|
- transportService));
|
|
|
+ new NodeConnectionsService(clusterService.getSettings(), threadPool, 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);
|
|
@@ -954,8 +951,8 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
|
|
|
Cluster.this::provideSeedHosts, clusterApplierService, onJoinValidators, Randomness.get(), (s, p, r) -> {},
|
|
|
getElectionStrategy());
|
|
|
masterService.setClusterStatePublisher(coordinator);
|
|
|
- final GatewayService gatewayService = new GatewayService(settings, allocationService, clusterService,
|
|
|
- deterministicTaskQueue.getThreadPool(this::onNode), coordinator, null);
|
|
|
+ final GatewayService gatewayService
|
|
|
+ = new GatewayService(settings, allocationService, clusterService, threadPool, coordinator, null);
|
|
|
|
|
|
logger.trace("starting up [{}]", localNode);
|
|
|
transportService.start();
|
|
@@ -1292,8 +1289,9 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
|
|
|
|
|
|
AckCollector nextAckCollector = new AckCollector();
|
|
|
|
|
|
- AckedFakeThreadPoolMasterService(String nodeName, String serviceName, Consumer<Runnable> onTaskAvailableToRun) {
|
|
|
- super(nodeName, serviceName, onTaskAvailableToRun);
|
|
|
+ AckedFakeThreadPoolMasterService(String nodeName, String serviceName, ThreadPool threadPool,
|
|
|
+ Consumer<Runnable> onTaskAvailableToRun) {
|
|
|
+ super(nodeName, serviceName, threadPool, onTaskAvailableToRun);
|
|
|
}
|
|
|
|
|
|
@Override
|
|
@@ -1323,8 +1321,8 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
|
|
|
private boolean applicationMayFail;
|
|
|
|
|
|
DisruptableClusterApplierService(String nodeName, Settings settings, ClusterSettings clusterSettings,
|
|
|
- DeterministicTaskQueue deterministicTaskQueue, Function<Runnable, Runnable> runnableWrapper) {
|
|
|
- super(nodeName, settings, clusterSettings, deterministicTaskQueue.getThreadPool(runnableWrapper));
|
|
|
+ DeterministicTaskQueue deterministicTaskQueue, ThreadPool threadPool) {
|
|
|
+ super(nodeName, settings, clusterSettings, threadPool);
|
|
|
this.nodeName = nodeName;
|
|
|
this.deterministicTaskQueue = deterministicTaskQueue;
|
|
|
addStateApplier(event -> {
|
|
@@ -1344,7 +1342,7 @@ public class AbstractCoordinatorTestCase extends ESTestCase {
|
|
|
|
|
|
@Override
|
|
|
protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
|
|
|
- return new MockSinglePrioritizingExecutor(nodeName, deterministicTaskQueue);
|
|
|
+ return new MockSinglePrioritizingExecutor(nodeName, deterministicTaskQueue, threadPool);
|
|
|
}
|
|
|
|
|
|
@Override
|