diff --git a/server/src/internalClusterTest/java/org/opensearch/discovery/ClusterManagerDisruptionIT.java b/server/src/internalClusterTest/java/org/opensearch/discovery/ClusterManagerDisruptionIT.java index 6c055c4015ff0..2f19d1e476c94 100644 --- a/server/src/internalClusterTest/java/org/opensearch/discovery/ClusterManagerDisruptionIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/discovery/ClusterManagerDisruptionIT.java @@ -43,7 +43,7 @@ import org.opensearch.common.unit.TimeValue; import org.opensearch.common.xcontent.XContentType; import org.opensearch.test.OpenSearchIntegTestCase; -import org.opensearch.test.disruption.BlockClusterManagerServiceOnClusterManager; +import org.opensearch.test.disruption.BlockMasterServiceOnMaster; import org.opensearch.test.disruption.IntermittentLongGCDisruption; import org.opensearch.test.disruption.NetworkDisruption; import org.opensearch.test.disruption.NetworkDisruption.TwoPartitions; @@ -308,7 +308,7 @@ public void testMappingTimeout() throws Exception { .setTransientSettings(Settings.builder().put("indices.mapping.dynamic_timeout", "1ms")) ); - ServiceDisruptionScheme disruption = new BlockClusterManagerServiceOnClusterManager(random()); + ServiceDisruptionScheme disruption = new BlockMasterServiceOnMaster(random()); setDisruptionScheme(disruption); disruption.startDisrupting(); diff --git a/server/src/internalClusterTest/java/org/opensearch/snapshots/DedicatedClusterSnapshotRestoreIT.java b/server/src/internalClusterTest/java/org/opensearch/snapshots/DedicatedClusterSnapshotRestoreIT.java index a4afdc0dba85c..2eca8555e1388 100644 --- a/server/src/internalClusterTest/java/org/opensearch/snapshots/DedicatedClusterSnapshotRestoreIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/snapshots/DedicatedClusterSnapshotRestoreIT.java @@ -94,7 +94,7 @@ import org.opensearch.test.OpenSearchIntegTestCase.Scope; import org.opensearch.test.InternalTestCluster; import org.opensearch.test.TestCustomMetadata; -import org.opensearch.test.disruption.BusyClusterManagerServiceDisruption; +import org.opensearch.test.disruption.BusyMasterServiceDisruption; import org.opensearch.test.disruption.ServiceDisruptionScheme; import org.opensearch.test.rest.FakeRestRequest; import org.opensearch.test.transport.MockTransportService; @@ -1157,7 +1157,7 @@ public void testDataNodeRestartWithBusyClusterManagerDuringSnapshot() throws Exc final String dataNode = blockNodeWithIndex("test-repo", "test-idx"); logger.info("--> snapshot"); - ServiceDisruptionScheme disruption = new BusyClusterManagerServiceDisruption(random(), Priority.HIGH); + ServiceDisruptionScheme disruption = new BusyMasterServiceDisruption(random(), Priority.HIGH); setDisruptionScheme(disruption); client(internalCluster().getMasterName()).admin() .cluster() diff --git a/server/src/main/java/org/opensearch/OpenSearchException.java b/server/src/main/java/org/opensearch/OpenSearchException.java index 4c12d031a028f..4ebcd9622ce38 100644 --- a/server/src/main/java/org/opensearch/OpenSearchException.java +++ b/server/src/main/java/org/opensearch/OpenSearchException.java @@ -33,7 +33,6 @@ package org.opensearch; import org.opensearch.action.support.replication.ReplicationOperation; -import org.opensearch.cluster.NotClusterManagerException; import org.opensearch.cluster.action.shard.ShardStateAction; import org.opensearch.common.CheckedFunction; import org.opensearch.common.Nullable; @@ -47,7 +46,6 @@ import org.opensearch.common.xcontent.XContentBuilder; import org.opensearch.common.xcontent.XContentParseException; import org.opensearch.common.xcontent.XContentParser; -import org.opensearch.discovery.ClusterManagerNotDiscoveredException; import org.opensearch.index.Index; import org.opensearch.index.shard.ShardId; import org.opensearch.rest.RestStatus; @@ -791,8 +789,8 @@ private enum OpenSearchExceptionHandle { UNKNOWN_VERSION_ADDED ), CLUSTER_MANAGER_NOT_DISCOVERED_EXCEPTION( - ClusterManagerNotDiscoveredException.class, - ClusterManagerNotDiscoveredException::new, + org.opensearch.discovery.ClusterManagerNotDiscoveredException.class, + org.opensearch.discovery.ClusterManagerNotDiscoveredException::new, 3, UNKNOWN_VERSION_ADDED ), @@ -1501,7 +1499,12 @@ private enum OpenSearchExceptionHandle { 143, UNKNOWN_VERSION_ADDED ), - NOT_CLUSTER_MANAGER_EXCEPTION(NotClusterManagerException.class, NotClusterManagerException::new, 144, UNKNOWN_VERSION_ADDED), + NOT_CLUSTER_MANAGER_EXCEPTION( + org.opensearch.cluster.NotClusterManagerException.class, + org.opensearch.cluster.NotClusterManagerException::new, + 144, + UNKNOWN_VERSION_ADDED + ), STATUS_EXCEPTION( org.opensearch.OpenSearchStatusException.class, org.opensearch.OpenSearchStatusException::new, diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java index 8a2ad77ac1693..3e3d373ff3b31 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java @@ -226,8 +226,8 @@ public void onNoLongerMaster(String source) { "stopped being cluster-manager while waiting for events with priority [{}]. retrying.", request.waitForEvents() ); - // TransportClusterManagerNodeAction implements the retry logic, which is triggered by passing a - // NotClusterManagerException + // TransportClusterManagerNodeAction implements the retry logic, + // which is triggered by passing a NotClusterManagerException listener.onFailure(new NotClusterManagerException("no longer cluster-manager. source: [" + source + "]")); } diff --git a/server/src/main/java/org/opensearch/cluster/ClusterStateTaskListener.java b/server/src/main/java/org/opensearch/cluster/ClusterStateTaskListener.java index 6a12e0cd5ed46..74cc118892579 100644 --- a/server/src/main/java/org/opensearch/cluster/ClusterStateTaskListener.java +++ b/server/src/main/java/org/opensearch/cluster/ClusterStateTaskListener.java @@ -31,7 +31,7 @@ package org.opensearch.cluster; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import java.util.List; @@ -49,7 +49,7 @@ public interface ClusterStateTaskListener { /** * called when the task was rejected because the local node is no longer cluster-manager. - * Used only for tasks submitted to {@link ClusterManagerService}. + * Used only for tasks submitted to {@link MasterService}. */ default void onNoLongerMaster(String source) { onFailure(source, new NotClusterManagerException("no longer cluster-manager. source: [" + source + "]")); diff --git a/server/src/main/java/org/opensearch/cluster/LocalNodeMasterListener.java b/server/src/main/java/org/opensearch/cluster/LocalNodeMasterListener.java new file mode 100644 index 0000000000000..eebfb60d8472d --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/LocalNodeMasterListener.java @@ -0,0 +1,44 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.cluster; + +/** + * Enables listening to cluster-manager changes events of the local node (when the local node becomes the cluster-manager, and when the local + * node cease being a cluster-manager). + * + * @opensearch.internal + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link LocalNodeClusterManagerListener} + */ +@Deprecated +public interface LocalNodeMasterListener extends LocalNodeClusterManagerListener { + +} diff --git a/server/src/main/java/org/opensearch/cluster/MasterNodeChangePredicate.java b/server/src/main/java/org/opensearch/cluster/MasterNodeChangePredicate.java new file mode 100644 index 0000000000000..d06aa219e3ca6 --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/MasterNodeChangePredicate.java @@ -0,0 +1,53 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ + +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.cluster; + +import java.util.function.Predicate; + +/** + * Utility class to build a predicate that accepts cluster state changes + * + * @opensearch.internal + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link ClusterManagerNodeChangePredicate} + */ +@Deprecated +public final class MasterNodeChangePredicate { + + private MasterNodeChangePredicate() { + + } + + public static Predicate build(ClusterState currentState) { + return ClusterManagerNodeChangePredicate.build(currentState); + } +} diff --git a/server/src/main/java/org/opensearch/cluster/NotMasterException.java b/server/src/main/java/org/opensearch/cluster/NotMasterException.java new file mode 100644 index 0000000000000..1dbf13212d2dd --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/NotMasterException.java @@ -0,0 +1,57 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.cluster; + +import org.opensearch.common.io.stream.StreamInput; + +import java.io.IOException; + +/** + * Thrown when a node join request or a cluster-manager ping reaches a node which is not + * currently acting as a cluster-manager or when a cluster state update task is to be executed + * on a node that is no longer cluster-manager. + * + * @opensearch.internal + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link NotClusterManagerException} + */ +@Deprecated +public class NotMasterException extends NotClusterManagerException { + + public NotMasterException(String msg) { + super(msg); + } + + public NotMasterException(StreamInput in) throws IOException { + super(in); + } + +} diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java index ee71307e82ec7..56d585d81e9dc 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java +++ b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java @@ -58,7 +58,7 @@ import org.opensearch.cluster.routing.allocation.AllocationService; import org.opensearch.cluster.service.ClusterApplier; import org.opensearch.cluster.service.ClusterApplier.ClusterApplyListener; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.Booleans; import org.opensearch.common.Nullable; import org.opensearch.common.Priority; @@ -141,7 +141,7 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery private final boolean singleNodeDiscovery; private final ElectionStrategy electionStrategy; private final TransportService transportService; - private final ClusterManagerService clusterManagerService; + private final MasterService masterService; private final AllocationService allocationService; private final JoinHelper joinHelper; private final NodeRemovalClusterStateTaskExecutor nodeRemovalExecutor; @@ -191,7 +191,7 @@ public Coordinator( TransportService transportService, NamedWriteableRegistry namedWriteableRegistry, AllocationService allocationService, - ClusterManagerService clusterManagerService, + MasterService masterService, Supplier persistedStateSupplier, SeedHostsProvider seedHostsProvider, ClusterApplier clusterApplier, @@ -203,7 +203,7 @@ public Coordinator( ) { this.settings = settings; this.transportService = transportService; - this.clusterManagerService = clusterManagerService; + this.masterService = masterService; this.allocationService = allocationService; this.onJoinValidators = JoinTaskExecutor.addBuiltInJoinValidators(onJoinValidators); this.singleNodeDiscovery = DiscoveryModule.isSingleNodeDiscovery(settings); @@ -211,7 +211,7 @@ public Coordinator( this.joinHelper = new JoinHelper( settings, allocationService, - clusterManagerService, + masterService, transportService, this::getCurrentTerm, this::getStateForClusterManagerService, @@ -260,7 +260,7 @@ public Coordinator( ); this.nodeRemovalExecutor = new NodeRemovalClusterStateTaskExecutor(allocationService, logger); this.clusterApplier = clusterApplier; - clusterManagerService.setClusterStateSupplier(this::getStateForClusterManagerService); + masterService.setClusterStateSupplier(this::getStateForClusterManagerService); this.reconfigurator = new Reconfigurator(settings, clusterSettings); this.clusterBootstrapService = new ClusterBootstrapService( settings, @@ -310,7 +310,7 @@ private void onLeaderFailure(Exception e) { private void removeNode(DiscoveryNode discoveryNode, String reason) { synchronized (mutex) { if (mode == Mode.LEADER) { - clusterManagerService.submitStateUpdateTask( + masterService.submitStateUpdateTask( "node-left", new NodeRemovalClusterStateTaskExecutor.Task(discoveryNode, reason), ClusterStateTaskConfig.build(Priority.IMMEDIATE), @@ -756,7 +756,7 @@ void becomeFollower(String method, DiscoveryNode leaderNode) { } private void cleanClusterManagerService() { - clusterManagerService.submitStateUpdateTask("clean-up after stepping down as cluster-manager", new LocalClusterUpdateTask() { + masterService.submitStateUpdateTask("clean-up after stepping down as cluster-manager", new LocalClusterUpdateTask() { @Override public void onFailure(String source, Exception e) { // ignore @@ -1126,7 +1126,7 @@ private void scheduleReconfigurationIfNeeded() { final ClusterState state = getLastAcceptedState(); if (improveConfiguration(state) != state && reconfigurationTaskScheduled.compareAndSet(false, true)) { logger.trace("scheduling reconfiguration"); - clusterManagerService.submitStateUpdateTask("reconfigure", new ClusterStateUpdateTask(Priority.URGENT) { + masterService.submitStateUpdateTask("reconfigure", new ClusterStateUpdateTask(Priority.URGENT) { @Override public ClusterState execute(ClusterState currentState) { reconfigurationTaskScheduled.set(false); diff --git a/server/src/main/java/org/opensearch/cluster/coordination/JoinHelper.java b/server/src/main/java/org/opensearch/cluster/coordination/JoinHelper.java index aeff856bef51a..3accc4f1d5baf 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/JoinHelper.java +++ b/server/src/main/java/org/opensearch/cluster/coordination/JoinHelper.java @@ -46,7 +46,7 @@ import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.routing.RerouteService; import org.opensearch.cluster.routing.allocation.AllocationService; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.Priority; import org.opensearch.common.collect.Tuple; import org.opensearch.common.io.stream.StreamInput; @@ -106,7 +106,7 @@ public class JoinHelper { Setting.Property.Deprecated ); - private final ClusterManagerService clusterManagerService; + private final MasterService masterService; private final TransportService transportService; private volatile JoinTaskExecutor joinTaskExecutor; @@ -122,7 +122,7 @@ public class JoinHelper { JoinHelper( Settings settings, AllocationService allocationService, - ClusterManagerService clusterManagerService, + MasterService masterService, TransportService transportService, LongSupplier currentTermSupplier, Supplier currentStateSupplier, @@ -132,7 +132,7 @@ public class JoinHelper { RerouteService rerouteService, NodeHealthService nodeHealthService ) { - this.clusterManagerService = clusterManagerService; + this.masterService = masterService; this.transportService = transportService; this.nodeHealthService = nodeHealthService; this.joinTimeout = JOIN_TIMEOUT_SETTING.get(settings); @@ -143,9 +143,8 @@ public class JoinHelper { @Override public ClusterTasksResult execute(ClusterState currentState, List joiningTasks) throws Exception { - // The current state that ClusterManagerService uses might have been updated by a (different) cluster-manager in a higher - // term - // already + // The current state that ClusterManagerService uses might have been updated by a (different) cluster-manager + // in a higher term already // Stop processing the current cluster state update, as there's no point in continuing to compute it as // it will later be rejected by Coordinator.publish(...) anyhow if (currentState.term() > term) { @@ -455,7 +454,7 @@ class LeaderJoinAccumulator implements JoinAccumulator { public void handleJoinRequest(DiscoveryNode sender, JoinCallback joinCallback) { final JoinTaskExecutor.Task task = new JoinTaskExecutor.Task(sender, "join existing leader"); assert joinTaskExecutor != null; - clusterManagerService.submitStateUpdateTask( + masterService.submitStateUpdateTask( "node-join", task, ClusterStateTaskConfig.build(Priority.URGENT), @@ -540,7 +539,7 @@ public void close(Mode newMode) { pendingAsTasks.put(JoinTaskExecutor.newBecomeMasterTask(), (source, e) -> {}); pendingAsTasks.put(JoinTaskExecutor.newFinishElectionTask(), (source, e) -> {}); joinTaskExecutor = joinTaskExecutorGenerator.get(); - clusterManagerService.submitStateUpdateTasks( + masterService.submitStateUpdateTasks( stateUpdateSource, pendingAsTasks, ClusterStateTaskConfig.build(Priority.URGENT), diff --git a/server/src/main/java/org/opensearch/cluster/coordination/NoMasterBlockService.java b/server/src/main/java/org/opensearch/cluster/coordination/NoMasterBlockService.java new file mode 100644 index 0000000000000..1dff7b2a8f0d6 --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/coordination/NoMasterBlockService.java @@ -0,0 +1,50 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.cluster.coordination; + +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; + +/** + * Service to block the master node + * + * @opensearch.internal + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link NoClusterManagerBlockService} + */ +@Deprecated +public class NoMasterBlockService extends NoClusterManagerBlockService { + + public NoMasterBlockService(Settings settings, ClusterSettings clusterSettings) { + super(settings, clusterSettings); + } + +} diff --git a/server/src/main/java/org/opensearch/cluster/coordination/UnsafeBootstrapMasterCommand.java b/server/src/main/java/org/opensearch/cluster/coordination/UnsafeBootstrapMasterCommand.java new file mode 100644 index 0000000000000..6014dc0b44ab8 --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/coordination/UnsafeBootstrapMasterCommand.java @@ -0,0 +1,47 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.cluster.coordination; + +/** + * Tool to run an unsafe bootstrap + * + * @opensearch.internal + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link UnsafeBootstrapClusterManagerCommand} + */ +@Deprecated +public class UnsafeBootstrapMasterCommand extends UnsafeBootstrapClusterManagerCommand { + + UnsafeBootstrapMasterCommand() { + super(); + } + +} diff --git a/server/src/main/java/org/opensearch/cluster/service/ClusterService.java b/server/src/main/java/org/opensearch/cluster/service/ClusterService.java index 7fd8ed62def07..baf453d18b3b3 100644 --- a/server/src/main/java/org/opensearch/cluster/service/ClusterService.java +++ b/server/src/main/java/org/opensearch/cluster/service/ClusterService.java @@ -62,7 +62,7 @@ * @opensearch.internal */ public class ClusterService extends AbstractLifecycleComponent { - private final ClusterManagerService clusterManagerService; + private final MasterService masterService; private final ClusterApplierService clusterApplierService; @@ -92,7 +92,7 @@ public ClusterService(Settings settings, ClusterSettings clusterSettings, Thread this( settings, clusterSettings, - new ClusterManagerService(settings, clusterSettings, threadPool), + new MasterService(settings, clusterSettings, threadPool), new ClusterApplierService(Node.NODE_NAME_SETTING.get(settings), settings, clusterSettings, threadPool) ); } @@ -100,12 +100,12 @@ public ClusterService(Settings settings, ClusterSettings clusterSettings, Thread public ClusterService( Settings settings, ClusterSettings clusterSettings, - ClusterManagerService clusterManagerService, + MasterService masterService, ClusterApplierService clusterApplierService ) { this.settings = settings; this.nodeName = Node.NODE_NAME_SETTING.get(settings); - this.clusterManagerService = clusterManagerService; + this.masterService = masterService; this.operationRouting = new OperationRouting(settings, clusterSettings); this.clusterSettings = clusterSettings; this.clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings); @@ -131,18 +131,18 @@ public RerouteService getRerouteService() { @Override protected synchronized void doStart() { clusterApplierService.start(); - clusterManagerService.start(); + masterService.start(); } @Override protected synchronized void doStop() { - clusterManagerService.stop(); + masterService.stop(); clusterApplierService.stop(); } @Override protected synchronized void doClose() { - clusterManagerService.close(); + masterService.close(); clusterApplierService.close(); } @@ -218,8 +218,8 @@ public void addLocalNodeMasterListener(LocalNodeClusterManagerListener listener) clusterApplierService.addLocalNodeMasterListener(listener); } - public ClusterManagerService getMasterService() { - return clusterManagerService; + public MasterService getMasterService() { + return masterService; } /** @@ -242,7 +242,7 @@ public ClusterApplierService getClusterApplierService() { public static boolean assertClusterOrMasterStateThread() { assert Thread.currentThread().getName().contains(ClusterApplierService.CLUSTER_UPDATE_THREAD_NAME) - || Thread.currentThread().getName().contains(ClusterManagerService.MASTER_UPDATE_THREAD_NAME) + || Thread.currentThread().getName().contains(MasterService.MASTER_UPDATE_THREAD_NAME) : "not called from the master/cluster state update thread"; return true; } @@ -333,6 +333,6 @@ public void submitStateUpdateTasks( final ClusterStateTaskConfig config, final ClusterStateTaskExecutor executor ) { - clusterManagerService.submitStateUpdateTasks(source, tasks, config, executor); + masterService.submitStateUpdateTasks(source, tasks, config, executor); } } diff --git a/server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java b/server/src/main/java/org/opensearch/cluster/service/MasterService.java similarity index 99% rename from server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java rename to server/src/main/java/org/opensearch/cluster/service/MasterService.java index df691c1443a56..2d52887858372 100644 --- a/server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java +++ b/server/src/main/java/org/opensearch/cluster/service/MasterService.java @@ -88,8 +88,8 @@ * * @opensearch.internal */ -public class ClusterManagerService extends AbstractLifecycleComponent { - private static final Logger logger = LogManager.getLogger(ClusterManagerService.class); +public class MasterService extends AbstractLifecycleComponent { + private static final Logger logger = LogManager.getLogger(MasterService.class); public static final Setting MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING = Setting.positiveTimeSetting( "cluster.service.slow_master_task_logging_threshold", @@ -122,7 +122,7 @@ public class ClusterManagerService extends AbstractLifecycleComponent { private volatile PrioritizedOpenSearchThreadPoolExecutor threadPoolExecutor; private volatile Batcher taskBatcher; - public ClusterManagerService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) { + public MasterService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) { this.nodeName = Objects.requireNonNull(Node.NODE_NAME_SETTING.get(settings)); this.slowTaskLoggingThreshold = CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings); diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java index 4451d11a7b976..f3d2ab0859513 100644 --- a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java @@ -81,7 +81,7 @@ import org.opensearch.cluster.routing.allocation.decider.ThrottlingAllocationDecider; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.logging.Loggers; import org.opensearch.common.network.NetworkModule; import org.opensearch.common.network.NetworkService; @@ -334,8 +334,8 @@ public void apply(Settings value, Settings current, Settings previous) { IndexModule.NODE_STORE_ALLOW_MMAP, ClusterApplierService.CLUSTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, ClusterService.USER_DEFINED_METADATA, - ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, // deprecated - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, + MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, // deprecated + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, SearchService.DEFAULT_SEARCH_TIMEOUT_SETTING, SearchService.DEFAULT_ALLOW_PARTIAL_SEARCH_RESULTS, TransportSearchAction.SHARD_COUNT_LIMIT_SETTING, diff --git a/server/src/main/java/org/opensearch/common/settings/ConsistentSettingsService.java b/server/src/main/java/org/opensearch/common/settings/ConsistentSettingsService.java index 062d62956d377..3be1c4b080b5f 100644 --- a/server/src/main/java/org/opensearch/common/settings/ConsistentSettingsService.java +++ b/server/src/main/java/org/opensearch/common/settings/ConsistentSettingsService.java @@ -36,7 +36,7 @@ import org.apache.logging.log4j.Logger; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.ClusterStateUpdateTask; -import org.opensearch.cluster.LocalNodeClusterManagerListener; +import org.opensearch.cluster.LocalNodeMasterListener; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.Priority; @@ -88,10 +88,10 @@ public ConsistentSettingsService(Settings settings, ClusterService clusterServic } /** - * Returns a {@link LocalNodeClusterManagerListener} that will publish hashes of all the settings passed in the constructor. These hashes are + * Returns a {@link LocalNodeMasterListener} that will publish hashes of all the settings passed in the constructor. These hashes are * published by the cluster-manager node only. Note that this is not designed for {@link SecureSettings} implementations that are mutable. */ - public LocalNodeClusterManagerListener newHashPublisher() { + public LocalNodeMasterListener newHashPublisher() { // eagerly compute hashes to be published final Map computedHashesOfConsistentSettings = computeHashesOfConsistentSecureSettings(); return new HashesPublisher(computedHashesOfConsistentSettings, clusterService); @@ -246,7 +246,7 @@ private byte[] computeSaltedPBKDF2Hash(byte[] bytes, byte[] salt) { } } - static final class HashesPublisher implements LocalNodeClusterManagerListener { + static final class HashesPublisher implements LocalNodeMasterListener { // eagerly compute hashes to be published final Map computedHashesOfConsistentSettings; diff --git a/server/src/main/java/org/opensearch/common/util/concurrent/BaseFuture.java b/server/src/main/java/org/opensearch/common/util/concurrent/BaseFuture.java index e4f8e1a221a0d..fef37299b349d 100644 --- a/server/src/main/java/org/opensearch/common/util/concurrent/BaseFuture.java +++ b/server/src/main/java/org/opensearch/common/util/concurrent/BaseFuture.java @@ -33,7 +33,7 @@ package org.opensearch.common.util.concurrent; import org.opensearch.cluster.service.ClusterApplierService; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.Nullable; import org.opensearch.threadpool.ThreadPool; import org.opensearch.transport.Transports; @@ -109,7 +109,7 @@ protected boolean blockingAllowed() { return Transports.assertNotTransportThread(BLOCKING_OP_REASON) && ThreadPool.assertNotScheduleThread(BLOCKING_OP_REASON) && ClusterApplierService.assertNotClusterStateUpdateThread(BLOCKING_OP_REASON) - && ClusterManagerService.assertNotMasterUpdateThread(BLOCKING_OP_REASON); + && MasterService.assertNotMasterUpdateThread(BLOCKING_OP_REASON); } @Override diff --git a/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java b/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java index 44f44fa055b2b..6b746e5963bdc 100644 --- a/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java +++ b/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java @@ -41,7 +41,7 @@ import org.opensearch.cluster.routing.RerouteService; import org.opensearch.cluster.routing.allocation.AllocationService; import org.opensearch.cluster.service.ClusterApplier; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.Randomness; import org.opensearch.common.io.stream.NamedWriteableRegistry; import org.opensearch.common.network.NetworkService; @@ -121,7 +121,7 @@ public DiscoveryModule( TransportService transportService, NamedWriteableRegistry namedWriteableRegistry, NetworkService networkService, - ClusterManagerService clusterManagerService, + MasterService masterService, ClusterApplier clusterApplier, ClusterSettings clusterSettings, List plugins, @@ -197,7 +197,7 @@ public DiscoveryModule( transportService, namedWriteableRegistry, allocationService, - clusterManagerService, + masterService, gatewayMetaState::getPersistedState, seedHostsProvider, clusterApplier, diff --git a/server/src/main/java/org/opensearch/discovery/MasterNotDiscoveredException.java b/server/src/main/java/org/opensearch/discovery/MasterNotDiscoveredException.java new file mode 100644 index 0000000000000..7ed21f924b27c --- /dev/null +++ b/server/src/main/java/org/opensearch/discovery/MasterNotDiscoveredException.java @@ -0,0 +1,63 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ + +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.discovery; + +import org.opensearch.common.io.stream.StreamInput; + +import java.io.IOException; + +/** + * Exception when the cluster-manager is not discovered + * + * @opensearch.internal + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link ClusterManagerNotDiscoveredException} + */ +@Deprecated +public class MasterNotDiscoveredException extends ClusterManagerNotDiscoveredException { + + public MasterNotDiscoveredException() { + super(); + } + + public MasterNotDiscoveredException(Throwable cause) { + super(cause); + } + + public MasterNotDiscoveredException(String message) { + super(message); + } + + public MasterNotDiscoveredException(StreamInput in) throws IOException { + super(in); + } +} diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestMasterAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestMasterAction.java new file mode 100644 index 0000000000000..20f7b01ef2b42 --- /dev/null +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestMasterAction.java @@ -0,0 +1,44 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +/* + * 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. + */ + +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.rest.action.cat; + +/** + * _cat API action to list cluster_manager information + * + * @opensearch.api + * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link RestClusterManagerAction} + */ +@Deprecated +public class RestMasterAction extends RestClusterManagerAction { + +} diff --git a/server/src/test/java/org/opensearch/ExceptionSerializationTests.java b/server/src/test/java/org/opensearch/ExceptionSerializationTests.java index 2af68bbedb456..c0d30a92e2d4d 100644 --- a/server/src/test/java/org/opensearch/ExceptionSerializationTests.java +++ b/server/src/test/java/org/opensearch/ExceptionSerializationTests.java @@ -44,7 +44,7 @@ import org.opensearch.action.search.ShardSearchFailure; import org.opensearch.action.support.replication.ReplicationOperation; import org.opensearch.client.AbstractClientHeadersTestCase; -import org.opensearch.cluster.NotClusterManagerException; +import org.opensearch.cluster.NotMasterException; import org.opensearch.cluster.action.shard.ShardStateAction; import org.opensearch.cluster.block.ClusterBlockException; import org.opensearch.cluster.coordination.CoordinationStateRejectedException; @@ -70,7 +70,7 @@ import org.opensearch.common.util.CancellableThreadsTests; import org.opensearch.common.util.set.Sets; import org.opensearch.common.xcontent.XContentLocation; -import org.opensearch.discovery.ClusterManagerNotDiscoveredException; +import org.opensearch.discovery.MasterNotDiscoveredException; import org.opensearch.env.ShardLockObtainFailedException; import org.opensearch.index.Index; import org.opensearch.index.engine.RecoveryEngineException; @@ -237,6 +237,9 @@ public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOEx Files.walkFileTree(testStartPath, visitor); assertTrue(notRegistered.remove(TestException.class)); assertTrue(notRegistered.remove(UnknownHeaderException.class)); + // Remove the deprecated exception classes from the unregistered list. + assertTrue(notRegistered.remove(NotMasterException.class)); + assertTrue(notRegistered.remove(MasterNotDiscoveredException.class)); assertTrue("Classes subclassing OpenSearchException must be registered \n" + notRegistered.toString(), notRegistered.isEmpty()); assertTrue(registered.removeAll(OpenSearchException.getRegisteredKeys())); // check assertEquals(registered.toString(), 0, registered.size()); @@ -697,7 +700,7 @@ public void testIds() { ids.put(0, org.opensearch.index.snapshots.IndexShardSnapshotFailedException.class); ids.put(1, org.opensearch.search.dfs.DfsPhaseExecutionException.class); ids.put(2, org.opensearch.common.util.CancellableThreads.ExecutionCancelledException.class); - ids.put(3, ClusterManagerNotDiscoveredException.class); + ids.put(3, org.opensearch.discovery.ClusterManagerNotDiscoveredException.class); ids.put(4, org.opensearch.OpenSearchSecurityException.class); ids.put(5, org.opensearch.index.snapshots.IndexShardRestoreException.class); ids.put(6, org.opensearch.indices.IndexClosedException.class); @@ -835,7 +838,7 @@ public void testIds() { ids.put(141, org.opensearch.index.query.QueryShardException.class); ids.put(142, ShardStateAction.NoLongerPrimaryShardException.class); ids.put(143, org.opensearch.script.ScriptException.class); - ids.put(144, NotClusterManagerException.class); + ids.put(144, org.opensearch.cluster.NotClusterManagerException.class); ids.put(145, org.opensearch.OpenSearchStatusException.class); ids.put(146, org.opensearch.tasks.TaskCancelledException.class); ids.put(147, org.opensearch.env.ShardLockObtainFailedException.class); diff --git a/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java b/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java index 976ce5cfee63e..b1ebc02842d25 100644 --- a/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java +++ b/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java @@ -67,6 +67,7 @@ import org.opensearch.rest.action.admin.cluster.RestGetStoredScriptAction; import org.opensearch.rest.action.admin.cluster.RestPutStoredScriptAction; import org.opensearch.rest.action.cat.RestAllocationAction; +import org.opensearch.rest.action.cat.RestMasterAction; import org.opensearch.rest.action.cat.RestRepositoriesAction; import org.opensearch.rest.action.cat.RestThreadPoolAction; import org.opensearch.rest.action.cat.RestClusterManagerAction; @@ -167,6 +168,13 @@ public void testCatClusterManager() { assertWarnings(MASTER_TIMEOUT_DEPRECATED_MESSAGE); } + public void testCatMaster() { + RestMasterAction action = new RestMasterAction(); + Exception e = assertThrows(OpenSearchParseException.class, () -> action.doCatRequest(getRestRequestWithBothParams(), client)); + assertThat(e.getMessage(), containsString(DUPLICATE_PARAMETER_ERROR_MESSAGE)); + assertWarnings(MASTER_TIMEOUT_DEPRECATED_MESSAGE); + } + public void testCatNodeattrs() { RestNodeAttrsAction action = new RestNodeAttrsAction(); Exception e = assertThrows(OpenSearchParseException.class, () -> action.doCatRequest(getRestRequestWithBothParams(), client)); diff --git a/server/src/test/java/org/opensearch/cluster/InternalClusterInfoServiceSchedulingTests.java b/server/src/test/java/org/opensearch/cluster/InternalClusterInfoServiceSchedulingTests.java index 6005ca9150488..7f09f1eb8e2be 100644 --- a/server/src/test/java/org/opensearch/cluster/InternalClusterInfoServiceSchedulingTests.java +++ b/server/src/test/java/org/opensearch/cluster/InternalClusterInfoServiceSchedulingTests.java @@ -48,8 +48,8 @@ import org.opensearch.cluster.service.ClusterApplier; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; -import org.opensearch.cluster.service.FakeThreadPoolClusterManagerService; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.FakeThreadPoolMasterService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.common.util.concurrent.PrioritizedOpenSearchThreadPoolExecutor; @@ -87,14 +87,14 @@ protected PrioritizedOpenSearchThreadPoolExecutor createThreadPoolExecutor() { } }; - final ClusterManagerService clusterManagerService = new FakeThreadPoolClusterManagerService( + final MasterService masterService = new FakeThreadPoolMasterService( "test", "clusterManagerService", threadPool, r -> { fail("cluster-manager service should not run any tasks"); } ); - final ClusterService clusterService = new ClusterService(settings, clusterSettings, clusterManagerService, clusterApplierService); + final ClusterService clusterService = new ClusterService(settings, clusterSettings, masterService, clusterApplierService); final FakeClusterInfoServiceClient client = new FakeClusterInfoServiceClient(threadPool); final InternalClusterInfoService clusterInfoService = new InternalClusterInfoService(settings, clusterService, threadPool, client); @@ -103,8 +103,8 @@ protected PrioritizedOpenSearchThreadPoolExecutor createThreadPoolExecutor() { clusterService.setNodeConnectionsService(ClusterServiceUtils.createNoOpNodeConnectionsService()); clusterApplierService.setInitialState(ClusterState.builder(new ClusterName("cluster")).nodes(noClusterManager).build()); - clusterManagerService.setClusterStatePublisher((clusterChangedEvent, publishListener, ackListener) -> fail("should not publish")); - clusterManagerService.setClusterStateSupplier(clusterApplierService::state); + masterService.setClusterStatePublisher((clusterChangedEvent, publishListener, ackListener) -> fail("should not publish")); + masterService.setClusterStateSupplier(clusterApplierService::state); clusterService.start(); final AtomicBoolean becameClusterManager1 = new AtomicBoolean(); diff --git a/server/src/test/java/org/opensearch/cluster/coordination/NodeJoinTests.java b/server/src/test/java/org/opensearch/cluster/coordination/NodeJoinTests.java index 3606d24222679..806468ed5e761 100644 --- a/server/src/test/java/org/opensearch/cluster/coordination/NodeJoinTests.java +++ b/server/src/test/java/org/opensearch/cluster/coordination/NodeJoinTests.java @@ -43,9 +43,9 @@ import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.node.DiscoveryNodeRole; import org.opensearch.cluster.node.DiscoveryNodes; -import org.opensearch.cluster.service.FakeThreadPoolClusterManagerService; -import org.opensearch.cluster.service.ClusterManagerService; -import org.opensearch.cluster.service.ClusterManagerServiceTests; +import org.opensearch.cluster.service.FakeThreadPoolMasterService; +import org.opensearch.cluster.service.MasterService; +import org.opensearch.cluster.service.MasterServiceTests; import org.opensearch.common.Randomness; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; @@ -98,7 +98,7 @@ public class NodeJoinTests extends OpenSearchTestCase { private static ThreadPool threadPool; - private ClusterManagerService clusterManagerService; + private MasterService masterService; private Coordinator coordinator; private DeterministicTaskQueue deterministicTaskQueue; private Transport transport; @@ -117,7 +117,7 @@ public static void afterClass() { @After public void tearDown() throws Exception { super.tearDown(); - clusterManagerService.close(); + masterService.close(); } private static ClusterState initialState(DiscoveryNode localNode, long term, long version, VotingConfiguration config) { @@ -144,7 +144,7 @@ private void setupFakeClusterManagerServiceAndCoordinator(long term, ClusterStat random() ); final ThreadPool fakeThreadPool = deterministicTaskQueue.getThreadPool(); - FakeThreadPoolClusterManagerService fakeClusterManagerService = new FakeThreadPoolClusterManagerService( + FakeThreadPoolMasterService fakeClusterManagerService = new FakeThreadPoolMasterService( "test_node", "test", fakeThreadPool, @@ -166,40 +166,40 @@ private void setupFakeClusterManagerServiceAndCoordinator(long term, ClusterStat } private void setupRealClusterManagerServiceAndCoordinator(long term, ClusterState initialState) { - ClusterManagerService clusterManagerService = new ClusterManagerService( + MasterService masterService = new MasterService( Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "test_node").build(), new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), threadPool ); AtomicReference clusterStateRef = new AtomicReference<>(initialState); - clusterManagerService.setClusterStatePublisher((event, publishListener, ackListener) -> { + masterService.setClusterStatePublisher((event, publishListener, ackListener) -> { clusterStateRef.set(event.state()); publishListener.onResponse(null); }); setupClusterManagerServiceAndCoordinator( term, initialState, - clusterManagerService, + masterService, threadPool, new Random(Randomness.get().nextLong()), () -> new StatusInfo(HEALTHY, "healthy-info") ); - clusterManagerService.setClusterStateSupplier(clusterStateRef::get); - clusterManagerService.start(); + masterService.setClusterStateSupplier(clusterStateRef::get); + masterService.start(); } private void setupClusterManagerServiceAndCoordinator( long term, ClusterState initialState, - ClusterManagerService clusterManagerService, + MasterService masterService, ThreadPool threadPool, Random random, NodeHealthService nodeHealthService ) { - if (this.clusterManagerService != null || coordinator != null) { + if (this.masterService != null || coordinator != null) { throw new IllegalStateException("method setupClusterManagerServiceAndCoordinator can only be called once"); } - this.clusterManagerService = clusterManagerService; + this.masterService = masterService; CapturingTransport capturingTransport = new CapturingTransport() { @Override protected void onSendRequest(long requestId, String action, TransportRequest request, DiscoveryNode destination) { @@ -231,7 +231,7 @@ protected void onSendRequest(long requestId, String action, TransportRequest req transportService, writableRegistry(), OpenSearchAllocationTestCase.createAllocationService(Settings.EMPTY), - clusterManagerService, + masterService, () -> new InMemoryPersistedState(term, initialState), r -> emptyList(), new NoOpClusterApplier(), @@ -514,7 +514,7 @@ public void testJoinUpdateVotingConfigExclusion() throws Exception { ); assertTrue( - ClusterManagerServiceTests.discoveryState(clusterManagerService) + MasterServiceTests.discoveryState(masterService) .getVotingConfigExclusions() .stream() .anyMatch( @@ -746,7 +746,7 @@ public void testConcurrentJoining() { throw new RuntimeException(e); } - assertTrue(ClusterManagerServiceTests.discoveryState(clusterManagerService).nodes().isLocalNodeElectedMaster()); + assertTrue(MasterServiceTests.discoveryState(masterService).nodes().isLocalNodeElectedMaster()); for (DiscoveryNode successfulNode : successfulNodes) { assertTrue(successfulNode + " joined cluster", clusterStateHasNode(successfulNode)); assertFalse(successfulNode + " voted for cluster-manager", coordinator.missingJoinVoteFrom(successfulNode)); @@ -776,10 +776,10 @@ public void testJoinElectedLeaderWithDeprecatedMasterRole() { } private boolean isLocalNodeElectedMaster() { - return ClusterManagerServiceTests.discoveryState(clusterManagerService).nodes().isLocalNodeElectedMaster(); + return MasterServiceTests.discoveryState(masterService).nodes().isLocalNodeElectedMaster(); } private boolean clusterStateHasNode(DiscoveryNode node) { - return node.equals(ClusterManagerServiceTests.discoveryState(clusterManagerService).nodes().get(node.getId())); + return node.equals(MasterServiceTests.discoveryState(masterService).nodes().get(node.getId())); } } diff --git a/server/src/test/java/org/opensearch/cluster/service/ClusterApplierServiceTests.java b/server/src/test/java/org/opensearch/cluster/service/ClusterApplierServiceTests.java index 166fb7a935009..a603dac9f42ca 100644 --- a/server/src/test/java/org/opensearch/cluster/service/ClusterApplierServiceTests.java +++ b/server/src/test/java/org/opensearch/cluster/service/ClusterApplierServiceTests.java @@ -39,6 +39,7 @@ import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.ClusterStateObserver; import org.opensearch.cluster.LocalNodeClusterManagerListener; +import org.opensearch.cluster.LocalNodeMasterListener; import org.opensearch.cluster.block.ClusterBlocks; import org.opensearch.cluster.coordination.NoClusterManagerBlockService; import org.opensearch.cluster.metadata.Metadata; @@ -331,6 +332,43 @@ public void offMaster() { timedClusterApplierService.close(); } + /* Validate the backwards compatibility of LocalNodeMasterListener remains + * after making it a subclass of LocalNodeClusterManagerListener. + * Overriding the methods with non-inclusive words are intentional. + * To support inclusive language, LocalNodeMasterListener is deprecated in 2.2. + */ + public void testDeprecatedLocalNodeMasterListenerCallbacks() { + TimedClusterApplierService timedClusterApplierService = createTimedClusterService(false); + + AtomicBoolean isClusterManager = new AtomicBoolean(); + timedClusterApplierService.addLocalNodeMasterListener(new LocalNodeMasterListener() { + @Override + public void onMaster() { + isClusterManager.set(true); + } + + @Override + public void offMaster() { + isClusterManager.set(false); + } + }); + + ClusterState state = timedClusterApplierService.state(); + DiscoveryNodes nodes = state.nodes(); + DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder(nodes).masterNodeId(nodes.getLocalNodeId()); + state = ClusterState.builder(state).nodes(nodesBuilder).build(); + setState(timedClusterApplierService, state); + assertThat(isClusterManager.get(), is(true)); + + nodes = state.nodes(); + nodesBuilder = DiscoveryNodes.builder(nodes).masterNodeId(null); + state = ClusterState.builder(state).nodes(nodesBuilder).build(); + setState(timedClusterApplierService, state); + assertThat(isClusterManager.get(), is(false)); + + timedClusterApplierService.close(); + } + public void testClusterStateApplierCantSampleClusterState() throws InterruptedException { AtomicReference error = new AtomicReference<>(); AtomicBoolean applierCalled = new AtomicBoolean(); diff --git a/server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceRenamedSettingTests.java b/server/src/test/java/org/opensearch/cluster/service/MasterServiceRenamedSettingTests.java similarity index 68% rename from server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceRenamedSettingTests.java rename to server/src/test/java/org/opensearch/cluster/service/MasterServiceRenamedSettingTests.java index e71d872c87527..acf089dc43b56 100644 --- a/server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceRenamedSettingTests.java +++ b/server/src/test/java/org/opensearch/cluster/service/MasterServiceRenamedSettingTests.java @@ -22,7 +22,7 @@ * after it is deprecated, so that the backwards compatibility is maintained. * The test can be removed along with removing support of the deprecated setting. */ -public class ClusterManagerServiceRenamedSettingTests extends OpenSearchTestCase { +public class MasterServiceRenamedSettingTests extends OpenSearchTestCase { /** * Validate the both settings are known and supported. @@ -33,8 +33,8 @@ public void testClusterManagerServiceSettingsExist() { "Both 'cluster.service.slow_cluster_manager_task_logging_threshold' and its predecessor should be supported built-in settings", settings.containsAll( Arrays.asList( - ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING + MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING ) ) ); @@ -45,8 +45,8 @@ public void testClusterManagerServiceSettingsExist() { */ public void testSettingFallback() { assertEquals( - ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY), - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY) + MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY), + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY) ); } @@ -57,11 +57,11 @@ public void testSettingGetValue() { Settings settings = Settings.builder().put("cluster.service.slow_cluster_manager_task_logging_threshold", "9s").build(); assertEquals( TimeValue.timeValueSeconds(9), - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) ); assertEquals( - ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.getDefault(Settings.EMPTY), - ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) + MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.getDefault(Settings.EMPTY), + MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) ); } @@ -73,10 +73,10 @@ public void testSettingGetValueWithFallback() { Settings settings = Settings.builder().put("cluster.service.slow_master_task_logging_threshold", "8s").build(); assertEquals( TimeValue.timeValueSeconds(8), - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) ); - assertSettingDeprecationsAndWarnings(new Setting[] { ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING }); + assertSettingDeprecationsAndWarnings(new Setting[] { MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING }); } /** @@ -89,11 +89,11 @@ public void testSettingGetValueWhenBothAreConfigured() { .build(); assertEquals( TimeValue.timeValueSeconds(9), - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings) ); - assertEquals(TimeValue.timeValueSeconds(8), ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings)); - assertSettingDeprecationsAndWarnings(new Setting[] { ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING }); + assertEquals(TimeValue.timeValueSeconds(8), MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings)); + assertSettingDeprecationsAndWarnings(new Setting[] { MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING }); } } diff --git a/server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceTests.java b/server/src/test/java/org/opensearch/cluster/service/MasterServiceTests.java similarity index 87% rename from server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceTests.java rename to server/src/test/java/org/opensearch/cluster/service/MasterServiceTests.java index 7c7a725032b52..c829bef576be5 100644 --- a/server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceTests.java +++ b/server/src/test/java/org/opensearch/cluster/service/MasterServiceTests.java @@ -93,14 +93,14 @@ import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.hasKey; -public class ClusterManagerServiceTests extends OpenSearchTestCase { +public class MasterServiceTests extends OpenSearchTestCase { private static ThreadPool threadPool; private static long relativeTimeInMillis; @BeforeClass public static void createThreadPool() { - threadPool = new TestThreadPool(ClusterManagerServiceTests.class.getName()) { + threadPool = new TestThreadPool(MasterServiceTests.class.getName()) { @Override public long relativeTimeInMillis() { return relativeTimeInMillis; @@ -121,17 +121,17 @@ public void randomizeCurrentTime() { relativeTimeInMillis = randomLongBetween(0L, 1L << 62); } - private ClusterManagerService createClusterManagerService(boolean makeClusterManager) { + private MasterService createClusterManagerService(boolean makeClusterManager) { final DiscoveryNode localNode = new DiscoveryNode("node1", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT); - final ClusterManagerService clusterManagerService = new ClusterManagerService( + final MasterService masterService = new MasterService( Settings.builder() - .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), ClusterManagerServiceTests.class.getSimpleName()) + .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), MasterServiceTests.class.getSimpleName()) .put(Node.NODE_NAME_SETTING.getKey(), "test_node") .build(), new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), threadPool ); - final ClusterState initialClusterState = ClusterState.builder(new ClusterName(ClusterManagerServiceTests.class.getSimpleName())) + final ClusterState initialClusterState = ClusterState.builder(new ClusterName(MasterServiceTests.class.getSimpleName())) .nodes( DiscoveryNodes.builder() .add(localNode) @@ -141,17 +141,17 @@ private ClusterManagerService createClusterManagerService(boolean makeClusterMan .blocks(ClusterBlocks.EMPTY_CLUSTER_BLOCK) .build(); final AtomicReference clusterStateRef = new AtomicReference<>(initialClusterState); - clusterManagerService.setClusterStatePublisher((event, publishListener, ackListener) -> { + masterService.setClusterStatePublisher((event, publishListener, ackListener) -> { clusterStateRef.set(event.state()); publishListener.onResponse(null); }); - clusterManagerService.setClusterStateSupplier(clusterStateRef::get); - clusterManagerService.start(); - return clusterManagerService; + masterService.setClusterStateSupplier(clusterStateRef::get); + masterService.start(); + return masterService; } public void testClusterManagerAwareExecution() throws Exception { - final ClusterManagerService nonClusterManager = createClusterManagerService(false); + final MasterService nonClusterManager = createClusterManagerService(false); final boolean[] taskFailed = { false }; final CountDownLatch latch1 = new CountDownLatch(1); @@ -194,7 +194,7 @@ public void onFailure(String source, Exception e) { } public void testThreadContext() throws InterruptedException { - final ClusterManagerService clusterManager = createClusterManagerService(true); + final MasterService masterService = createClusterManagerService(true); final CountDownLatch latch = new CountDownLatch(1); try (ThreadContext.StoredContext ignored = threadPool.getThreadContext().stashContext()) { @@ -208,7 +208,7 @@ public void testThreadContext() throws InterruptedException { final TimeValue ackTimeout = randomBoolean() ? TimeValue.ZERO : TimeValue.timeValueMillis(randomInt(10000)); final TimeValue clusterManagerTimeout = randomBoolean() ? TimeValue.ZERO : TimeValue.timeValueMillis(randomInt(10000)); - clusterManager.submitStateUpdateTask("test", new AckedClusterStateUpdateTask(null, null) { + masterService.submitStateUpdateTask("test", new AckedClusterStateUpdateTask(null, null) { @Override public ClusterState execute(ClusterState currentState) { assertTrue(threadPool.getThreadContext().isSystemContext()); @@ -280,7 +280,7 @@ public void onAckTimeout() { latch.await(); - clusterManager.close(); + masterService.close(); } /* @@ -292,8 +292,8 @@ public void testClusterStateTaskListenerThrowingExceptionIsOkay() throws Interru final CountDownLatch latch = new CountDownLatch(1); AtomicBoolean published = new AtomicBoolean(); - try (ClusterManagerService clusterManagerService = createClusterManagerService(true)) { - clusterManagerService.submitStateUpdateTask( + try (MasterService masterService = createClusterManagerService(true)) { + masterService.submitStateUpdateTask( "testClusterStateTaskListenerThrowingExceptionIsOkay", new Object(), ClusterStateTaskConfig.build(Priority.NORMAL), @@ -328,11 +328,11 @@ public void onFailure(String source, Exception e) {} @TestLogging(value = "org.opensearch.cluster.service:TRACE", reason = "to ensure that we log cluster state events on TRACE level") public void testClusterStateUpdateLogging() throws Exception { - try (MockLogAppender mockAppender = MockLogAppender.createForLoggers(LogManager.getLogger(ClusterManagerService.class))) { + try (MockLogAppender mockAppender = MockLogAppender.createForLoggers(LogManager.getLogger(MasterService.class))) { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test1 start", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "executing cluster state update for [test1]" ) @@ -340,7 +340,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test1 computation", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "took [1s] to compute cluster state update for [test1]" ) @@ -348,7 +348,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test1 notification", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "took [0s] to notify listeners on unchanged cluster state for [test1]" ) @@ -357,7 +357,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test2 start", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "executing cluster state update for [test2]" ) @@ -365,7 +365,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test2 failure", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.TRACE, "failed to execute cluster state update (on version: [*], uuid: [*]) for [test2]*" ) @@ -373,7 +373,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test2 computation", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "took [2s] to compute cluster state update for [test2]" ) @@ -381,7 +381,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test2 notification", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "took [0s] to notify listeners on unchanged cluster state for [test2]" ) @@ -390,7 +390,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test3 start", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "executing cluster state update for [test3]" ) @@ -398,7 +398,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test3 computation", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "took [3s] to compute cluster state update for [test3]" ) @@ -406,7 +406,7 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test3 notification", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "took [4s] to notify listeners on successful publication of cluster state (version: *, uuid: *) for [test3]" ) @@ -415,14 +415,14 @@ public void testClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test4", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.DEBUG, "executing cluster state update for [test4]" ) ); - try (ClusterManagerService clusterManagerService = createClusterManagerService(true)) { - clusterManagerService.submitStateUpdateTask("test1", new ClusterStateUpdateTask() { + try (MasterService masterService = createClusterManagerService(true)) { + masterService.submitStateUpdateTask("test1", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { relativeTimeInMillis += TimeValue.timeValueSeconds(1).millis(); @@ -437,7 +437,7 @@ public void onFailure(String source, Exception e) { fail(); } }); - clusterManagerService.submitStateUpdateTask("test2", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test2", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { relativeTimeInMillis += TimeValue.timeValueSeconds(2).millis(); @@ -452,7 +452,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS @Override public void onFailure(String source, Exception e) {} }); - clusterManagerService.submitStateUpdateTask("test3", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test3", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { relativeTimeInMillis += TimeValue.timeValueSeconds(3).millis(); @@ -469,7 +469,7 @@ public void onFailure(String source, Exception e) { fail(); } }); - clusterManagerService.submitStateUpdateTask("test4", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test4", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { return currentState; @@ -617,7 +617,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS } }; - try (ClusterManagerService clusterManagerService = createClusterManagerService(true)) { + try (MasterService masterService = createClusterManagerService(true)) { final ConcurrentMap submittedTasksPerThread = new ConcurrentHashMap<>(); CyclicBarrier barrier = new CyclicBarrier(1 + numberOfThreads); for (int i = 0; i < numberOfThreads; i++) { @@ -632,7 +632,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS submittedTasksPerThread.computeIfAbsent(threadName, key -> new AtomicInteger()).addAndGet(tasks.size()); final TaskExecutor executor = assignment.v1(); if (tasks.size() == 1) { - clusterManagerService.submitStateUpdateTask( + masterService.submitStateUpdateTask( threadName, tasks.stream().findFirst().get(), ClusterStateTaskConfig.build(randomFrom(Priority.values())), @@ -642,7 +642,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS } else { Map taskListeners = new HashMap<>(); tasks.forEach(t -> taskListeners.put(t, listener)); - clusterManagerService.submitStateUpdateTasks( + masterService.submitStateUpdateTasks( threadName, taskListeners, ClusterStateTaskConfig.build(randomFrom(Priority.values())), @@ -696,8 +696,8 @@ public void testBlockingCallInClusterStateTaskListenerFails() throws Interrupted final CountDownLatch latch = new CountDownLatch(1); final AtomicReference assertionRef = new AtomicReference<>(); - try (ClusterManagerService clusterManagerService = createClusterManagerService(true)) { - clusterManagerService.submitStateUpdateTask( + try (MasterService masterService = createClusterManagerService(true)) { + masterService.submitStateUpdateTask( "testBlockingCallInClusterStateTaskListenerFails", new Object(), ClusterStateTaskConfig.build(Priority.NORMAL), @@ -737,11 +737,11 @@ public void onFailure(String source, Exception e) {} @TestLogging(value = "org.opensearch.cluster.service:WARN", reason = "to ensure that we log cluster state events on WARN level") public void testLongClusterStateUpdateLogging() throws Exception { - try (MockLogAppender mockAppender = MockLogAppender.createForLoggers(LogManager.getLogger(ClusterManagerService.class))) { + try (MockLogAppender mockAppender = MockLogAppender.createForLoggers(LogManager.getLogger(MasterService.class))) { mockAppender.addExpectation( new MockLogAppender.UnseenEventExpectation( "test1 shouldn't log because it was fast enough", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.WARN, "*took*test1*" ) @@ -749,7 +749,7 @@ public void testLongClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test2", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.WARN, "*took [*], which is over [10s], to compute cluster state update for [test2]" ) @@ -757,7 +757,7 @@ public void testLongClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test3", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.WARN, "*took [*], which is over [10s], to compute cluster state update for [test3]" ) @@ -765,7 +765,7 @@ public void testLongClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test4", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.WARN, "*took [*], which is over [10s], to compute cluster state update for [test4]" ) @@ -773,7 +773,7 @@ public void testLongClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.UnseenEventExpectation( "test5 should not log despite publishing slowly", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.WARN, "*took*test5*" ) @@ -781,16 +781,16 @@ public void testLongClusterStateUpdateLogging() throws Exception { mockAppender.addExpectation( new MockLogAppender.SeenEventExpectation( "test6 should log due to slow and failing publication", - ClusterManagerService.class.getCanonicalName(), + MasterService.class.getCanonicalName(), Level.WARN, "took [*] and then failed to publish updated cluster state (version: *, uuid: *) for [test6]:*" ) ); try ( - ClusterManagerService clusterManagerService = new ClusterManagerService( + MasterService masterService = new MasterService( Settings.builder() - .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), ClusterManagerServiceTests.class.getSimpleName()) + .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), MasterServiceTests.class.getSimpleName()) .put(Node.NODE_NAME_SETTING.getKey(), "test_node") .build(), new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), @@ -805,21 +805,19 @@ public void testLongClusterStateUpdateLogging() throws Exception { emptySet(), Version.CURRENT ); - final ClusterState initialClusterState = ClusterState.builder( - new ClusterName(ClusterManagerServiceTests.class.getSimpleName()) - ) + final ClusterState initialClusterState = ClusterState.builder(new ClusterName(MasterServiceTests.class.getSimpleName())) .nodes(DiscoveryNodes.builder().add(localNode).localNodeId(localNode.getId()).masterNodeId(localNode.getId())) .blocks(ClusterBlocks.EMPTY_CLUSTER_BLOCK) .build(); final AtomicReference clusterStateRef = new AtomicReference<>(initialClusterState); - clusterManagerService.setClusterStatePublisher((event, publishListener, ackListener) -> { + masterService.setClusterStatePublisher((event, publishListener, ackListener) -> { if (event.source().contains("test5")) { - relativeTimeInMillis += ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( + relativeTimeInMillis += MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( Settings.EMPTY ).millis() + randomLongBetween(1, 1000000); } if (event.source().contains("test6")) { - relativeTimeInMillis += ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( + relativeTimeInMillis += MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( Settings.EMPTY ).millis() + randomLongBetween(1, 1000000); throw new OpenSearchException("simulated error during slow publication which should trigger logging"); @@ -827,17 +825,17 @@ public void testLongClusterStateUpdateLogging() throws Exception { clusterStateRef.set(event.state()); publishListener.onResponse(null); }); - clusterManagerService.setClusterStateSupplier(clusterStateRef::get); - clusterManagerService.start(); + masterService.setClusterStateSupplier(clusterStateRef::get); + masterService.start(); final CountDownLatch latch = new CountDownLatch(6); final CountDownLatch processedFirstTask = new CountDownLatch(1); - clusterManagerService.submitStateUpdateTask("test1", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test1", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { relativeTimeInMillis += randomLongBetween( 0L, - ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY).millis() + MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY).millis() ); return currentState; } @@ -855,10 +853,10 @@ public void onFailure(String source, Exception e) { }); processedFirstTask.await(); - clusterManagerService.submitStateUpdateTask("test2", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test2", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { - relativeTimeInMillis += ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( + relativeTimeInMillis += MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( Settings.EMPTY ).millis() + randomLongBetween(1, 1000000); throw new IllegalArgumentException("Testing handling of exceptions in the cluster state task"); @@ -874,10 +872,10 @@ public void onFailure(String source, Exception e) { latch.countDown(); } }); - clusterManagerService.submitStateUpdateTask("test3", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test3", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { - relativeTimeInMillis += ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( + relativeTimeInMillis += MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( Settings.EMPTY ).millis() + randomLongBetween(1, 1000000); return ClusterState.builder(currentState).incrementVersion().build(); @@ -893,10 +891,10 @@ public void onFailure(String source, Exception e) { fail(); } }); - clusterManagerService.submitStateUpdateTask("test4", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test4", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { - relativeTimeInMillis += ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( + relativeTimeInMillis += MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get( Settings.EMPTY ).millis() + randomLongBetween(1, 1000000); return currentState; @@ -912,7 +910,7 @@ public void onFailure(String source, Exception e) { fail(); } }); - clusterManagerService.submitStateUpdateTask("test5", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test5", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState).incrementVersion().build(); @@ -928,7 +926,7 @@ public void onFailure(String source, Exception e) { fail(); } }); - clusterManagerService.submitStateUpdateTask("test6", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test6", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState).incrementVersion().build(); @@ -946,7 +944,7 @@ public void onFailure(String source, Exception e) { }); // Additional update task to make sure all previous logging made it to the loggerName // We don't check logging for this on since there is no guarantee that it will occur before our check - clusterManagerService.submitStateUpdateTask("test7", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test7", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { return currentState; @@ -973,9 +971,9 @@ public void testAcking() throws InterruptedException { final DiscoveryNode node2 = new DiscoveryNode("node2", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT); final DiscoveryNode node3 = new DiscoveryNode("node3", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT); try ( - ClusterManagerService clusterManagerService = new ClusterManagerService( + MasterService masterService = new MasterService( Settings.builder() - .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), ClusterManagerServiceTests.class.getSimpleName()) + .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), MasterServiceTests.class.getSimpleName()) .put(Node.NODE_NAME_SETTING.getKey(), "test_node") .build(), new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), @@ -983,14 +981,14 @@ public void testAcking() throws InterruptedException { ) ) { - final ClusterState initialClusterState = ClusterState.builder(new ClusterName(ClusterManagerServiceTests.class.getSimpleName())) + final ClusterState initialClusterState = ClusterState.builder(new ClusterName(MasterServiceTests.class.getSimpleName())) .nodes(DiscoveryNodes.builder().add(node1).add(node2).add(node3).localNodeId(node1.getId()).masterNodeId(node1.getId())) .blocks(ClusterBlocks.EMPTY_CLUSTER_BLOCK) .build(); final AtomicReference publisherRef = new AtomicReference<>(); - clusterManagerService.setClusterStatePublisher((e, pl, al) -> publisherRef.get().publish(e, pl, al)); - clusterManagerService.setClusterStateSupplier(() -> initialClusterState); - clusterManagerService.start(); + masterService.setClusterStatePublisher((e, pl, al) -> publisherRef.get().publish(e, pl, al)); + masterService.setClusterStateSupplier(() -> initialClusterState); + masterService.start(); // check that we don't time out before even committing the cluster state { @@ -1002,7 +1000,7 @@ public void testAcking() throws InterruptedException { ) ); - clusterManagerService.submitStateUpdateTask("test2", new AckedClusterStateUpdateTask(null, null) { + masterService.submitStateUpdateTask("test2", new AckedClusterStateUpdateTask(null, null) { @Override public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState).build(); @@ -1057,7 +1055,7 @@ public void onAckTimeout() { ackListener.onNodeAck(node3, null); }); - clusterManagerService.submitStateUpdateTask("test2", new AckedClusterStateUpdateTask(null, null) { + masterService.submitStateUpdateTask("test2", new AckedClusterStateUpdateTask(null, null) { @Override public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState).build(); @@ -1103,8 +1101,8 @@ public void onAckTimeout() { /** * Returns the cluster state that the cluster-manager service uses (and that is provided by the discovery layer) */ - public static ClusterState discoveryState(ClusterManagerService clusterManagerService) { - return clusterManagerService.state(); + public static ClusterState discoveryState(MasterService masterService) { + return masterService.state(); } } diff --git a/server/src/test/java/org/opensearch/discovery/DiscoveryModuleTests.java b/server/src/test/java/org/opensearch/discovery/DiscoveryModuleTests.java index 5588e9c1ceba8..d1e3f406b4933 100644 --- a/server/src/test/java/org/opensearch/discovery/DiscoveryModuleTests.java +++ b/server/src/test/java/org/opensearch/discovery/DiscoveryModuleTests.java @@ -37,7 +37,7 @@ import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.routing.RerouteService; import org.opensearch.cluster.service.ClusterApplier; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.io.stream.NamedWriteableRegistry; import org.opensearch.common.network.NetworkService; import org.opensearch.common.settings.ClusterSettings; @@ -70,7 +70,7 @@ public class DiscoveryModuleTests extends OpenSearchTestCase { private TransportService transportService; private NamedWriteableRegistry namedWriteableRegistry; - private ClusterManagerService clusterManagerService; + private MasterService masterService; private ClusterApplier clusterApplier; private ThreadPool threadPool; private ClusterSettings clusterSettings; @@ -93,7 +93,7 @@ public void setupDummyServices() { threadPool = mock(ThreadPool.class); when(threadPool.getThreadContext()).thenReturn(new ThreadContext(Settings.EMPTY)); transportService = MockTransportService.createNewService(Settings.EMPTY, Version.CURRENT, threadPool, null); - clusterManagerService = mock(ClusterManagerService.class); + masterService = mock(MasterService.class); namedWriteableRegistry = new NamedWriteableRegistry(Collections.emptyList()); clusterApplier = mock(ClusterApplier.class); clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); @@ -112,7 +112,7 @@ private DiscoveryModule newModule(Settings settings, List plugi transportService, namedWriteableRegistry, null, - clusterManagerService, + masterService, clusterApplier, clusterSettings, plugins, diff --git a/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java b/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java index 6faaa6973953c..7e0f977e6e835 100644 --- a/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java +++ b/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java @@ -144,8 +144,8 @@ import org.opensearch.cluster.routing.allocation.command.AllocateEmptyPrimaryAllocationCommand; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; -import org.opensearch.cluster.service.FakeThreadPoolClusterManagerService; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.FakeThreadPoolMasterService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.CheckedConsumer; import org.opensearch.common.Nullable; import org.opensearch.common.io.stream.NamedWriteableRegistry; @@ -1643,7 +1643,7 @@ private final class TestClusterNode { private final DiscoveryNode node; - private final ClusterManagerService clusterManagerService; + private final MasterService masterService; private final AllocationService allocationService; @@ -1663,18 +1663,13 @@ private final class TestClusterNode { this.node = node; final Environment environment = createEnvironment(node.getName()); threadPool = deterministicTaskQueue.getThreadPool(runnable -> CoordinatorTests.onNodeLog(node, runnable)); - clusterManagerService = new FakeThreadPoolClusterManagerService( - node.getName(), - "test", - threadPool, - deterministicTaskQueue::scheduleNow - ); + masterService = new FakeThreadPoolMasterService(node.getName(), "test", threadPool, deterministicTaskQueue::scheduleNow); final Settings settings = environment.settings(); final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); clusterService = new ClusterService( settings, clusterSettings, - clusterManagerService, + masterService, new ClusterApplierService(node.getName(), settings, clusterSettings, threadPool) { @Override protected PrioritizedOpenSearchThreadPoolExecutor createThreadPoolExecutor() { @@ -2205,7 +2200,7 @@ public void start(ClusterState initialState) { transportService, namedWriteableRegistry, allocationService, - clusterManagerService, + masterService, () -> persistedState, hostsResolver -> nodes.values() .stream() @@ -2219,7 +2214,7 @@ public void start(ClusterState initialState) { ElectionStrategy.DEFAULT_INSTANCE, () -> new StatusInfo(HEALTHY, "healthy-info") ); - clusterManagerService.setClusterStatePublisher(coordinator); + masterService.setClusterStatePublisher(coordinator); coordinator.start(); clusterService.getClusterApplierService().setNodeConnectionsService(nodeConnectionsService); nodeConnectionsService.start(); diff --git a/test/framework/src/main/java/org/opensearch/cluster/coordination/AbstractCoordinatorTestCase.java b/test/framework/src/main/java/org/opensearch/cluster/coordination/AbstractCoordinatorTestCase.java index 0e0b466408f77..19551b0adecb2 100644 --- a/test/framework/src/main/java/org/opensearch/cluster/coordination/AbstractCoordinatorTestCase.java +++ b/test/framework/src/main/java/org/opensearch/cluster/coordination/AbstractCoordinatorTestCase.java @@ -54,7 +54,7 @@ import org.opensearch.cluster.routing.allocation.AllocationService; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; -import org.opensearch.cluster.service.FakeThreadPoolClusterManagerService; +import org.opensearch.cluster.service.FakeThreadPoolMasterService; import org.opensearch.common.Nullable; import org.opensearch.common.Randomness; import org.opensearch.common.UUIDs; @@ -557,7 +557,7 @@ void stabilise(long stabilisationDurationMillis) { final ClusterNode leader = getAnyLeader(); final long leaderTerm = leader.coordinator.getCurrentTerm(); - final int pendingTaskCount = leader.clusterManagerService.getFakeMasterServicePendingTaskCount(); + final int pendingTaskCount = leader.masterService.getFakeMasterServicePendingTaskCount(); runFor((pendingTaskCount + 1) * DEFAULT_CLUSTER_STATE_UPDATE_DELAY, "draining task queue"); final Matcher isEqualToLeaderVersion = equalTo(leader.coordinator.getLastAcceptedState().getVersion()); @@ -1025,7 +1025,7 @@ class ClusterNode { private final DiscoveryNode localNode; final MockPersistedState persistedState; final Settings nodeSettings; - private AckedFakeThreadPoolClusterManagerService clusterManagerService; + private AckedFakeThreadPoolMasterService masterService; private DisruptableClusterApplierService clusterApplierService; private ClusterService clusterService; TransportService transportService; @@ -1105,7 +1105,7 @@ protected Optional getDisruptableMockTransport(Transpo null, emptySet() ); - clusterManagerService = new AckedFakeThreadPoolClusterManagerService( + masterService = new AckedFakeThreadPoolMasterService( localNode.getId(), "test", threadPool, @@ -1119,7 +1119,7 @@ protected Optional getDisruptableMockTransport(Transpo deterministicTaskQueue, threadPool ); - clusterService = new ClusterService(settings, clusterSettings, clusterManagerService, clusterApplierService); + clusterService = new ClusterService(settings, clusterSettings, masterService, clusterApplierService); clusterService.setNodeConnectionsService( new NodeConnectionsService(clusterService.getSettings(), threadPool, transportService) ); @@ -1134,7 +1134,7 @@ protected Optional getDisruptableMockTransport(Transpo transportService, writableRegistry(), allocationService, - clusterManagerService, + masterService, this::getPersistedState, Cluster.this::provideSeedHosts, clusterApplierService, @@ -1144,7 +1144,7 @@ protected Optional getDisruptableMockTransport(Transpo getElectionStrategy(), nodeHealthService ); - clusterManagerService.setClusterStatePublisher(coordinator); + masterService.setClusterStatePublisher(coordinator); final GatewayService gatewayService = new GatewayService( settings, allocationService, @@ -1331,11 +1331,11 @@ AckCollector submitUpdateTask( onNode(() -> { logger.trace("[{}] submitUpdateTask: enqueueing [{}]", localNode.getId(), source); final long submittedTerm = coordinator.getCurrentTerm(); - clusterManagerService.submitStateUpdateTask(source, new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask(source, new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { assertThat(currentState.term(), greaterThanOrEqualTo(submittedTerm)); - clusterManagerService.nextAckCollector = ackCollector; + masterService.nextAckCollector = ackCollector; return clusterStateUpdate.apply(currentState); } @@ -1510,11 +1510,11 @@ int getSuccessfulAckIndex(ClusterNode clusterNode) { } } - static class AckedFakeThreadPoolClusterManagerService extends FakeThreadPoolClusterManagerService { + static class AckedFakeThreadPoolMasterService extends FakeThreadPoolMasterService { AckCollector nextAckCollector = new AckCollector(); - AckedFakeThreadPoolClusterManagerService( + AckedFakeThreadPoolMasterService( String nodeName, String serviceName, ThreadPool threadPool, diff --git a/test/framework/src/main/java/org/opensearch/cluster/service/FakeThreadPoolClusterManagerService.java b/test/framework/src/main/java/org/opensearch/cluster/service/FakeThreadPoolMasterService.java similarity index 96% rename from test/framework/src/main/java/org/opensearch/cluster/service/FakeThreadPoolClusterManagerService.java rename to test/framework/src/main/java/org/opensearch/cluster/service/FakeThreadPoolMasterService.java index 46893fe08bd76..c532a0cc36472 100644 --- a/test/framework/src/main/java/org/opensearch/cluster/service/FakeThreadPoolClusterManagerService.java +++ b/test/framework/src/main/java/org/opensearch/cluster/service/FakeThreadPoolMasterService.java @@ -55,8 +55,8 @@ import static org.apache.lucene.tests.util.LuceneTestCase.random; import static org.opensearch.test.OpenSearchTestCase.randomInt; -public class FakeThreadPoolClusterManagerService extends ClusterManagerService { - private static final Logger logger = LogManager.getLogger(FakeThreadPoolClusterManagerService.class); +public class FakeThreadPoolMasterService extends MasterService { + private static final Logger logger = LogManager.getLogger(FakeThreadPoolMasterService.class); private final String name; private final List pendingTasks = new ArrayList<>(); @@ -65,7 +65,7 @@ public class FakeThreadPoolClusterManagerService extends ClusterManagerService { private boolean taskInProgress = false; private boolean waitForPublish = false; - public FakeThreadPoolClusterManagerService( + public FakeThreadPoolMasterService( String nodeName, String serviceName, ThreadPool threadPool, @@ -137,7 +137,7 @@ public void run() { if (waitForPublish == false) { taskInProgress = false; } - FakeThreadPoolClusterManagerService.this.scheduleNextTaskIfNecessary(); + FakeThreadPoolMasterService.this.scheduleNextTaskIfNecessary(); } }); } diff --git a/test/framework/src/main/java/org/opensearch/test/ClusterServiceUtils.java b/test/framework/src/main/java/org/opensearch/test/ClusterServiceUtils.java index 93c437d3cf91b..99f9b86fb6479 100644 --- a/test/framework/src/main/java/org/opensearch/test/ClusterServiceUtils.java +++ b/test/framework/src/main/java/org/opensearch/test/ClusterServiceUtils.java @@ -47,7 +47,7 @@ import org.opensearch.cluster.service.ClusterApplier.ClusterApplyListener; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; -import org.opensearch.cluster.service.ClusterManagerService; +import org.opensearch.cluster.service.MasterService; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.node.Node; @@ -61,23 +61,23 @@ public class ClusterServiceUtils { - public static ClusterManagerService createMasterService(ThreadPool threadPool, ClusterState initialClusterState) { - ClusterManagerService clusterManagerService = new ClusterManagerService( + public static MasterService createMasterService(ThreadPool threadPool, ClusterState initialClusterState) { + MasterService masterService = new MasterService( Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), "test_cluster_manager_node").build(), new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), threadPool ); AtomicReference clusterStateRef = new AtomicReference<>(initialClusterState); - clusterManagerService.setClusterStatePublisher((event, publishListener, ackListener) -> { + masterService.setClusterStatePublisher((event, publishListener, ackListener) -> { clusterStateRef.set(event.state()); publishListener.onResponse(null); }); - clusterManagerService.setClusterStateSupplier(clusterStateRef::get); - clusterManagerService.start(); - return clusterManagerService; + masterService.setClusterStateSupplier(clusterStateRef::get); + masterService.start(); + return masterService; } - public static ClusterManagerService createMasterService(ThreadPool threadPool, DiscoveryNode localNode) { + public static MasterService createMasterService(ThreadPool threadPool, DiscoveryNode localNode) { ClusterState initialClusterState = ClusterState.builder(new ClusterName(ClusterServiceUtils.class.getSimpleName())) .nodes(DiscoveryNodes.builder().add(localNode).localNodeId(localNode.getId()).masterNodeId(localNode.getId())) .blocks(ClusterBlocks.EMPTY_CLUSTER_BLOCK) @@ -114,7 +114,7 @@ public void onFailure(String source, Exception e) { } } - public static void setState(ClusterManagerService executor, ClusterState clusterState) { + public static void setState(MasterService executor, ClusterState clusterState) { CountDownLatch latch = new CountDownLatch(1); executor.submitStateUpdateTask("test setting state", new ClusterStateUpdateTask() { @Override diff --git a/test/framework/src/main/java/org/opensearch/test/disruption/BlockClusterManagerServiceOnClusterManager.java b/test/framework/src/main/java/org/opensearch/test/disruption/BlockMasterServiceOnMaster.java similarity index 96% rename from test/framework/src/main/java/org/opensearch/test/disruption/BlockClusterManagerServiceOnClusterManager.java rename to test/framework/src/main/java/org/opensearch/test/disruption/BlockMasterServiceOnMaster.java index 8edb508625341..85f8e5c250066 100644 --- a/test/framework/src/main/java/org/opensearch/test/disruption/BlockClusterManagerServiceOnClusterManager.java +++ b/test/framework/src/main/java/org/opensearch/test/disruption/BlockMasterServiceOnMaster.java @@ -43,11 +43,11 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; -public class BlockClusterManagerServiceOnClusterManager extends SingleNodeDisruption { +public class BlockMasterServiceOnMaster extends SingleNodeDisruption { AtomicReference disruptionLatch = new AtomicReference<>(); - public BlockClusterManagerServiceOnClusterManager(Random random) { + public BlockMasterServiceOnMaster(Random random) { super(random); } diff --git a/test/framework/src/main/java/org/opensearch/test/disruption/BusyClusterManagerServiceDisruption.java b/test/framework/src/main/java/org/opensearch/test/disruption/BusyMasterServiceDisruption.java similarity index 95% rename from test/framework/src/main/java/org/opensearch/test/disruption/BusyClusterManagerServiceDisruption.java rename to test/framework/src/main/java/org/opensearch/test/disruption/BusyMasterServiceDisruption.java index 55eb90a4dc89d..4830f9b0359fb 100644 --- a/test/framework/src/main/java/org/opensearch/test/disruption/BusyClusterManagerServiceDisruption.java +++ b/test/framework/src/main/java/org/opensearch/test/disruption/BusyMasterServiceDisruption.java @@ -41,11 +41,11 @@ import java.util.Random; import java.util.concurrent.atomic.AtomicBoolean; -public class BusyClusterManagerServiceDisruption extends SingleNodeDisruption { +public class BusyMasterServiceDisruption extends SingleNodeDisruption { private final AtomicBoolean active = new AtomicBoolean(); private final Priority priority; - public BusyClusterManagerServiceDisruption(Random random, Priority priority) { + public BusyMasterServiceDisruption(Random random, Priority priority) { super(random); this.priority = priority; } diff --git a/test/framework/src/test/java/org/opensearch/cluster/service/FakeThreadPoolClusterManagerServiceTests.java b/test/framework/src/test/java/org/opensearch/cluster/service/FakeThreadPoolMasterServiceTests.java similarity index 92% rename from test/framework/src/test/java/org/opensearch/cluster/service/FakeThreadPoolClusterManagerServiceTests.java rename to test/framework/src/test/java/org/opensearch/cluster/service/FakeThreadPoolMasterServiceTests.java index f09e62fa574e6..b4cdb0b33391a 100644 --- a/test/framework/src/test/java/org/opensearch/cluster/service/FakeThreadPoolClusterManagerServiceTests.java +++ b/test/framework/src/test/java/org/opensearch/cluster/service/FakeThreadPoolMasterServiceTests.java @@ -61,7 +61,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; -public class FakeThreadPoolClusterManagerServiceTests extends OpenSearchTestCase { +public class FakeThreadPoolMasterServiceTests extends OpenSearchTestCase { public void testFakeClusterManagerService() { List runnableTasks = new ArrayList<>(); @@ -84,21 +84,21 @@ public void testFakeClusterManagerService() { doAnswer(invocationOnMock -> runnableTasks.add((Runnable) invocationOnMock.getArguments()[0])).when(executorService).execute(any()); when(mockThreadPool.generic()).thenReturn(executorService); - FakeThreadPoolClusterManagerService clusterManagerService = new FakeThreadPoolClusterManagerService( + FakeThreadPoolMasterService masterService = new FakeThreadPoolMasterService( "test_node", "test", mockThreadPool, runnableTasks::add ); - clusterManagerService.setClusterStateSupplier(lastClusterStateRef::get); - clusterManagerService.setClusterStatePublisher((event, publishListener, ackListener) -> { + masterService.setClusterStateSupplier(lastClusterStateRef::get); + masterService.setClusterStatePublisher((event, publishListener, ackListener) -> { lastClusterStateRef.set(event.state()); publishingCallback.set(publishListener); }); - clusterManagerService.start(); + masterService.start(); AtomicBoolean firstTaskCompleted = new AtomicBoolean(); - clusterManagerService.submitStateUpdateTask("test1", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test1", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState) @@ -138,7 +138,7 @@ public void onFailure(String source, Exception e) { assertThat(runnableTasks.size(), equalTo(0)); AtomicBoolean secondTaskCompleted = new AtomicBoolean(); - clusterManagerService.submitStateUpdateTask("test2", new ClusterStateUpdateTask() { + masterService.submitStateUpdateTask("test2", new ClusterStateUpdateTask() { @Override public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState)