Deprecate class 'MasterService' and create alternative class 'Cluster…

…ManagerService' (#4022) (#4050)

To support inclusive language, the `master` terminology is going to be replaced by `cluster manager` in the code base.

- Deprecate class `MasterService` and create alternative class `ClusterManagerService`.
- Add a unit test to validate the method `ClusterService.getMasterService()` can still return an object in type `MasterService`.
- Rename all the existing references of `MasterService` to `ClusterManagerService`, and rename the local variable names.
- Deprecate public methods `ClusterServiceUtils.createMasterService(...)` and create alternative methods `createClusterManagerService(...)`

Note:
The class `ClusterManagerService` is a subclass of `MasterService`, the inheritance relationship is opposite from most of the other classes with `Master` in the name (that covered by issue #1684).
The reason is:
There is a public method that has return value in type `MasterService`,
https:/opensearch-project/OpenSearch/blob/388c80ad94529b1d9aad0a735c4740dce2932a32/server/src/main/java/org/opensearch/cluster/service/ClusterService.java#L221
And in the code for Performance Analyzer plugin, there is a local variable in type `MasterService`:
https:/opensearch-project/performance-analyzer/blob/5ee4809ac1cda6517ed871aeb12c6635203e7f1d/src/main/java/org/opensearch/performanceanalyzer/collectors/MasterServiceEventMetrics.java#L219
If making the old class `MasterService` a subclass of the new class `ClusterManagerService`, the above usage will be broken.
Reversing the inheritance relationship, I'm able to keep the backwards compatibility of the method `getMasterService()` while deprecating the class `MasterService` and encourage using a new class `ClusterManagerService`.

Signed-off-by: Tianli Feng <[email protected]>
(cherry picked from commit 740f75d)

server/src/internalClusterTest/java/org/opensearch/cluster/coordination/ZenDiscoveryIT.java

@@ -170,7 +170,9 @@ public void testDiscoveryStats() throws Exception {
  ensureGreen(); // ensures that all events are processed (in particular state recovery fully completed)
  assertBusy(
  () -> assertThat(
- internalCluster().clusterService(internalCluster().getClusterManagerName()).getMasterService().numberOfPendingTasks(),
+ internalCluster().clusterService(internalCluster().getClusterManagerName())
+ .getClusterManagerService()
+ .numberOfPendingTasks(),
  equalTo(0)
  )
  ); // see https:/elastic/elasticsearch/issues/24388

server/src/internalClusterTest/java/org/opensearch/cluster/service/ClusterServiceIT.java

@@ -391,7 +391,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
 
  // The tasks can be re-ordered, so we need to check out-of-order
  Set<String> controlSources = new HashSet<>(Arrays.asList("1", "2", "3", "4", "5", "6", "7", "8", "9", "10"));
- List<PendingClusterTask> pendingClusterTasks = clusterService.getMasterService().pendingTasks();
+ List<PendingClusterTask> pendingClusterTasks = clusterService.getClusterManagerService().pendingTasks();
  assertThat(pendingClusterTasks.size(), greaterThanOrEqualTo(10));
  assertThat(pendingClusterTasks.get(0).getSource().string(), equalTo("1"));
  assertThat(pendingClusterTasks.get(0).isExecuting(), equalTo(true));
@@ -413,7 +413,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
  invoked2.await();
 
  // whenever we test for no tasks, we need to wait since this is a live node
- assertBusy(() -> assertTrue("Pending tasks not empty", clusterService.getMasterService().pendingTasks().isEmpty()));
+ assertBusy(() -> assertTrue("Pending tasks not empty", clusterService.getClusterManagerService().pendingTasks().isEmpty()));
  waitNoPendingTasksOnAll();
 
  final CountDownLatch block2 = new CountDownLatch(1);
@@ -453,7 +453,7 @@ public void onFailure(String source, Exception e) {
  }
  Thread.sleep(100);
 
- pendingClusterTasks = clusterService.getMasterService().pendingTasks();
+ pendingClusterTasks = clusterService.getClusterManagerService().pendingTasks();
  assertThat(pendingClusterTasks.size(), greaterThanOrEqualTo(5));
  controlSources = new HashSet<>(Arrays.asList("1", "2", "3", "4", "5"));
  for (PendingClusterTask task : pendingClusterTasks) {

server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -319,9 +319,9 @@ private boolean validateRequest(final ClusterHealthRequest request, ClusterState
  ClusterHealthResponse response = clusterHealth(
  request,
  clusterState,
- clusterService.getMasterService().numberOfPendingTasks(),
+ clusterService.getClusterManagerService().numberOfPendingTasks(),
  allocationService.getNumberOfInFlightFetches(),
- clusterService.getMasterService().getMaxTaskWaitTime()
+ clusterService.getClusterManagerService().getMaxTaskWaitTime()
  );
  return prepareResponse(request, response, clusterState, indexNameExpressionResolver) == waitCount;
  }
@@ -341,9 +341,9 @@ private ClusterHealthResponse getResponse(
  ClusterHealthResponse response = clusterHealth(
  request,
  clusterState,
- clusterService.getMasterService().numberOfPendingTasks(),
+ clusterService.getClusterManagerService().numberOfPendingTasks(),
  allocationService.getNumberOfInFlightFetches(),
- clusterService.getMasterService().getMaxTaskWaitTime()
+ clusterService.getClusterManagerService().getMaxTaskWaitTime()
  );
  int readyCounter = prepareResponse(request, response, clusterState, indexNameExpressionResolver);
  boolean valid = (readyCounter == waitFor);

server/src/main/java/org/opensearch/action/admin/cluster/tasks/TransportPendingClusterTasksAction.java

@@ -106,7 +106,7 @@ protected void clusterManagerOperation(
  ActionListener<PendingClusterTasksResponse> listener
  ) {
  logger.trace("fetching pending tasks from cluster service");
- final List<PendingClusterTask> pendingTasks = clusterService.getMasterService().pendingTasks();
+ final List<PendingClusterTask> pendingTasks = clusterService.getClusterManagerService().pendingTasks();
  logger.trace("done fetching pending tasks from cluster service");
  listener.onResponse(new PendingClusterTasksResponse(pendingTasks));
  }

server/src/main/java/org/opensearch/cluster/ClusterStateTaskListener.java

@@ -31,7 +31,7 @@
 
 package org.opensearch.cluster;
 
-import org.opensearch.cluster.service.MasterService;
+import org.opensearch.cluster.service.ClusterManagerService;
 
 import java.util.List;
 
@@ -49,15 +49,15 @@ 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 MasterService}.
+ * Used only for tasks submitted to {@link ClusterManagerService}.
  */
  default void onNoLongerClusterManager(String source) {
  onFailure(source, new NotClusterManagerException("no longer cluster-manager. source: [" + source + "]"));
  }
 
  /**
  * called when the task was rejected because the local node is no longer cluster-manager.
- * Used only for tasks submitted to {@link MasterService}.
+ * Used only for tasks submitted to {@link ClusterManagerService}.
  *
  * @deprecated As of 2.1, because supporting inclusive language, replaced by {@link #onNoLongerClusterManager(String)}
  */

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.MasterService;
+import org.opensearch.cluster.service.ClusterManagerService;
 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 MasterService masterService;
+ private final ClusterManagerService clusterManagerService;
  private final AllocationService allocationService;
  private final JoinHelper joinHelper;
  private final NodeRemovalClusterStateTaskExecutor nodeRemovalExecutor;
@@ -191,7 +191,7 @@ public Coordinator(
  TransportService transportService,
  NamedWriteableRegistry namedWriteableRegistry,
  AllocationService allocationService,
- MasterService masterService,
+ ClusterManagerService clusterManagerService,
  Supplier<CoordinationState.PersistedState> persistedStateSupplier,
  SeedHostsProvider seedHostsProvider,
  ClusterApplier clusterApplier,
@@ -203,15 +203,15 @@ public Coordinator(
  ) {
  this.settings = settings;
  this.transportService = transportService;
- this.masterService = masterService;
+ this.clusterManagerService = clusterManagerService;
  this.allocationService = allocationService;
  this.onJoinValidators = JoinTaskExecutor.addBuiltInJoinValidators(onJoinValidators);
  this.singleNodeDiscovery = DiscoveryModule.isSingleNodeDiscovery(settings);
  this.electionStrategy = electionStrategy;
  this.joinHelper = new JoinHelper(
  settings,
  allocationService,
- masterService,
+ clusterManagerService,
  transportService,
  this::getCurrentTerm,
  this::getStateForClusterManagerService,
@@ -260,7 +260,7 @@ public Coordinator(
  );
  this.nodeRemovalExecutor = new NodeRemovalClusterStateTaskExecutor(allocationService, logger);
  this.clusterApplier = clusterApplier;
- masterService.setClusterStateSupplier(this::getStateForClusterManagerService);
+ clusterManagerService.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) {
- masterService.submitStateUpdateTask(
+ clusterManagerService.submitStateUpdateTask(
  "node-left",
  new NodeRemovalClusterStateTaskExecutor.Task(discoveryNode, reason),
  ClusterStateTaskConfig.build(Priority.IMMEDIATE),
@@ -757,7 +757,7 @@ void becomeFollower(String method, DiscoveryNode leaderNode) {
  }
 
  private void cleanClusterManagerService() {
- masterService.submitStateUpdateTask("clean-up after stepping down as cluster-manager", new LocalClusterUpdateTask() {
+ clusterManagerService.submitStateUpdateTask("clean-up after stepping down as cluster-manager", new LocalClusterUpdateTask() {
  @Override
  public void onFailure(String source, Exception e) {
  // ignore
@@ -1129,7 +1129,7 @@ private void scheduleReconfigurationIfNeeded() {
  final ClusterState state = getLastAcceptedState();
  if (improveConfiguration(state) != state && reconfigurationTaskScheduled.compareAndSet(false, true)) {
  logger.trace("scheduling reconfiguration");
- masterService.submitStateUpdateTask("reconfigure", new ClusterStateUpdateTask(Priority.URGENT) {
+ clusterManagerService.submitStateUpdateTask("reconfigure", new ClusterStateUpdateTask(Priority.URGENT) {
  @Override
  public ClusterState execute(ClusterState currentState) {
  reconfigurationTaskScheduled.set(false);

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.MasterService;
+import org.opensearch.cluster.service.ClusterManagerService;
 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 MasterService masterService;
+ private final ClusterManagerService clusterManagerService;
  private final TransportService transportService;
  private volatile JoinTaskExecutor joinTaskExecutor;
 
@@ -122,7 +122,7 @@ public class JoinHelper {
  JoinHelper(
  Settings settings,
  AllocationService allocationService,
- MasterService masterService,
+ ClusterManagerService clusterManagerService,
  TransportService transportService,
  LongSupplier currentTermSupplier,
  Supplier<ClusterState> currentStateSupplier,
@@ -132,7 +132,7 @@ public class JoinHelper {
  RerouteService rerouteService,
  NodeHealthService nodeHealthService
  ) {
- this.masterService = masterService;
+ this.clusterManagerService = clusterManagerService;
  this.transportService = transportService;
  this.nodeHealthService = nodeHealthService;
  this.joinTimeout = JOIN_TIMEOUT_SETTING.get(settings);
@@ -458,7 +458,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;
- masterService.submitStateUpdateTask(
+ clusterManagerService.submitStateUpdateTask(
  "node-join",
  task,
  ClusterStateTaskConfig.build(Priority.URGENT),
@@ -543,7 +543,7 @@ public void close(Mode newMode) {
  pendingAsTasks.put(JoinTaskExecutor.newBecomeClusterManagerTask(), (source, e) -> {});
  pendingAsTasks.put(JoinTaskExecutor.newFinishElectionTask(), (source, e) -> {});
  joinTaskExecutor = joinTaskExecutorGenerator.get();
- masterService.submitStateUpdateTasks(
+ clusterManagerService.submitStateUpdateTasks(
  stateUpdateSource,
  pendingAsTasks,
  ClusterStateTaskConfig.build(Priority.URGENT),

server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java

@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+package org.opensearch.cluster.service;
+
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.threadpool.ThreadPool;
+
+/**
+ * Main Cluster Manager Node Service
+ *
+ * @opensearch.internal
+ */
+public class ClusterManagerService extends MasterService {
+ public ClusterManagerService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) {
+ super(settings, clusterSettings, threadPool);
+ }
+}

server/src/main/java/org/opensearch/cluster/service/ClusterService.java

@@ -63,7 +63,7 @@
  * @opensearch.internal
  */
 public class ClusterService extends AbstractLifecycleComponent {
- private final MasterService masterService;
+ private final ClusterManagerService clusterManagerService;
 
  private final ClusterApplierService clusterApplierService;
 
@@ -93,20 +93,20 @@ public ClusterService(Settings settings, ClusterSettings clusterSettings, Thread
  this(
  settings,
  clusterSettings,
- new MasterService(settings, clusterSettings, threadPool),
+ new ClusterManagerService(settings, clusterSettings, threadPool),
  new ClusterApplierService(Node.NODE_NAME_SETTING.get(settings), settings, clusterSettings, threadPool)
  );
  }
 
  public ClusterService(
  Settings settings,
  ClusterSettings clusterSettings,
- MasterService masterService,
+ ClusterManagerService clusterManagerService,
  ClusterApplierService clusterApplierService
  ) {
  this.settings = settings;
  this.nodeName = Node.NODE_NAME_SETTING.get(settings);
- this.masterService = masterService;
+ this.clusterManagerService = clusterManagerService;
  this.operationRouting = new OperationRouting(settings, clusterSettings);
  this.clusterSettings = clusterSettings;
  this.clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings);
@@ -132,18 +132,18 @@ public RerouteService getRerouteService() {
  @Override
  protected synchronized void doStart() {
  clusterApplierService.start();
- masterService.start();
+ clusterManagerService.start();
  }
 
  @Override
  protected synchronized void doStop() {
- masterService.stop();
+ clusterManagerService.stop();
  clusterApplierService.stop();
  }
 
  @Override
  protected synchronized void doClose() {
- masterService.close();
+ clusterManagerService.close();
  clusterApplierService.close();
  }
 
@@ -228,8 +228,14 @@ public void addLocalNodeMasterListener(LocalNodeMasterListener listener) {
  addLocalNodeClusterManagerListener(listener);
  }
 
+ public ClusterManagerService getClusterManagerService() {
+ return clusterManagerService;
+ }
+
+ /** @deprecated As of 2.2, because supporting inclusive language, replaced by {@link #getClusterManagerService()} */
+ @Deprecated
  public MasterService getMasterService() {
- return masterService;
+ return clusterManagerService;
  }
 
  /**
@@ -252,7 +258,7 @@ public ClusterApplierService getClusterApplierService() {
 
  public static boolean assertClusterOrClusterManagerStateThread() {
  assert Thread.currentThread().getName().contains(ClusterApplierService.CLUSTER_UPDATE_THREAD_NAME)
- || Thread.currentThread().getName().contains(MasterService.CLUSTER_MANAGER_UPDATE_THREAD_NAME)
+ || Thread.currentThread().getName().contains(ClusterManagerService.CLUSTER_MANAGER_UPDATE_THREAD_NAME)
  : "not called from the master/cluster state update thread";
  return true;
  }
@@ -349,6 +355,6 @@ public <T> void submitStateUpdateTasks(
  final ClusterStateTaskConfig config,
  final ClusterStateTaskExecutor<T> executor
  ) {
- masterService.submitStateUpdateTasks(source, tasks, config, executor);
+ clusterManagerService.submitStateUpdateTasks(source, tasks, config, executor);
  }
 }

server/src/main/java/org/opensearch/cluster/service/MasterService.java

@@ -87,7 +87,9 @@
  * Main Master Node Service
  *
  * @opensearch.internal
+ * @deprecated As of 2.2, because supporting inclusive language, replaced by {@link ClusterManagerService}.
  */
+@Deprecated
 public class MasterService extends AbstractLifecycleComponent {
  private static final Logger logger = LogManager.getLogger(MasterService.class);
 

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -82,7 +82,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.MasterService;
+import org.opensearch.cluster.service.ClusterManagerService;
 import org.opensearch.common.logging.Loggers;
 import org.opensearch.common.network.NetworkModule;
 import org.opensearch.common.network.NetworkService;
@@ -335,8 +335,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,
- MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, // deprecated
- MasterService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING,
+ ClusterManagerService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING, // deprecated
+ ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING,
  SearchService.DEFAULT_SEARCH_TIMEOUT_SETTING,
  SearchService.DEFAULT_ALLOW_PARTIAL_SEARCH_RESULTS,
  TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,

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.MasterService;
+import org.opensearch.cluster.service.ClusterManagerService;
 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)
- && MasterService.assertNotClusterManagerUpdateThread(BLOCKING_OP_REASON);
+ && ClusterManagerService.assertNotClusterManagerUpdateThread(BLOCKING_OP_REASON);
  }
 
  @Override