001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.rsgroup;
019
020import java.io.ByteArrayInputStream;
021import java.io.IOException;
022import java.util.ArrayList;
023import java.util.Collection;
024import java.util.Collections;
025import java.util.HashMap;
026import java.util.HashSet;
027import java.util.LinkedList;
028import java.util.List;
029import java.util.Map;
030import java.util.Optional;
031import java.util.OptionalLong;
032import java.util.Set;
033import java.util.SortedSet;
034import java.util.TreeSet;
035import java.util.concurrent.Future;
036import java.util.function.Function;
037import java.util.regex.Pattern;
038import java.util.regex.PatternSyntaxException;
039import java.util.stream.Collectors;
040import org.apache.commons.lang3.StringUtils;
041import org.apache.hadoop.conf.Configuration;
042import org.apache.hadoop.hbase.Coprocessor;
043import org.apache.hadoop.hbase.DoNotRetryIOException;
044import org.apache.hadoop.hbase.HConstants;
045import org.apache.hadoop.hbase.NamespaceDescriptor;
046import org.apache.hadoop.hbase.ServerName;
047import org.apache.hadoop.hbase.TableDescriptors;
048import org.apache.hadoop.hbase.TableName;
049import org.apache.hadoop.hbase.client.AsyncClusterConnection;
050import org.apache.hadoop.hbase.client.AsyncTable;
051import org.apache.hadoop.hbase.client.BalanceRequest;
052import org.apache.hadoop.hbase.client.BalanceResponse;
053import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
054import org.apache.hadoop.hbase.client.CoprocessorDescriptorBuilder;
055import org.apache.hadoop.hbase.client.Delete;
056import org.apache.hadoop.hbase.client.Get;
057import org.apache.hadoop.hbase.client.Mutation;
058import org.apache.hadoop.hbase.client.Put;
059import org.apache.hadoop.hbase.client.RegionInfo;
060import org.apache.hadoop.hbase.client.Result;
061import org.apache.hadoop.hbase.client.ResultScanner;
062import org.apache.hadoop.hbase.client.TableDescriptor;
063import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
064import org.apache.hadoop.hbase.client.TableState;
065import org.apache.hadoop.hbase.constraint.ConstraintException;
066import org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint;
067import org.apache.hadoop.hbase.exceptions.DeserializationException;
068import org.apache.hadoop.hbase.master.LoadBalancer;
069import org.apache.hadoop.hbase.master.MasterServices;
070import org.apache.hadoop.hbase.master.RegionPlan;
071import org.apache.hadoop.hbase.master.RegionState;
072import org.apache.hadoop.hbase.master.ServerListener;
073import org.apache.hadoop.hbase.master.ServerManager;
074import org.apache.hadoop.hbase.master.TableStateManager;
075import org.apache.hadoop.hbase.master.assignment.RegionStateNode;
076import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
077import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
078import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil;
079import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
080import org.apache.hadoop.hbase.net.Address;
081import org.apache.hadoop.hbase.procedure2.Procedure;
082import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
083import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
084import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
085import org.apache.hadoop.hbase.util.Bytes;
086import org.apache.hadoop.hbase.util.FutureUtils;
087import org.apache.hadoop.hbase.util.Pair;
088import org.apache.hadoop.hbase.util.Threads;
089import org.apache.hadoop.hbase.zookeeper.ZKUtil;
090import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
091import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
092import org.apache.hadoop.util.Shell;
093import org.apache.yetus.audience.InterfaceAudience;
094import org.apache.zookeeper.KeeperException;
095import org.slf4j.Logger;
096import org.slf4j.LoggerFactory;
097
098import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
099import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
100import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
101
102import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
103import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto;
104import org.apache.hadoop.hbase.shaded.protobuf.generated.MultiRowMutationProtos.MultiRowMutationService;
105import org.apache.hadoop.hbase.shaded.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest;
106import org.apache.hadoop.hbase.shaded.protobuf.generated.MultiRowMutationProtos.MutateRowsResponse;
107import org.apache.hadoop.hbase.shaded.protobuf.generated.RSGroupProtos;
108
109/**
110 * This is an implementation of {@link RSGroupInfoManager} which makes use of an HBase table as the
111 * persistence store for the group information. It also makes use of zookeeper to store group
112 * information needed for bootstrapping during offline mode.
113 * <h2>Admin-managed and auto-managed groups</h2> An <b>admin-managed</b> group is a non-default
114 * group that is not regex-governed (see {@code hbase.rsgroup.regex.<group>}). Its servers are set
115 * by an administrator and are stored in hbase:rsgroup and ZooKeeper. A group whose regex is invalid
116 * is admin-managed until the regex is fixed. In that time the servers that the group holds in
117 * memory count as admin-managed when the membership is computed
118 * ({@code resolveRegexBasedRSGroupMembership}). After a master restart the group holds only what
119 * was stored at the last flush.
120 * <p/>
121 * An <b>auto-managed</b> group is 'default' or a regex-governed group. Its servers are computed
122 * from the live servers and, while the regex is valid, are not stored in hbase:rsgroup and
123 * ZooKeeper.
124 * <p/>
125 * Auto-managed groups exist to make RSGroup work with auto-scaling when HBase runs in a cloud
126 * environment such as AWS or GCP. An admin-managed group does not give the elasticity that
127 * automatic upscaling and downscaling need: each new RegionServer must be moved into the group by
128 * hand. It fits a fixed number of RegionServers, for example to host internal system tables like
129 * hbase:meta and hbase:namespace. End user facing tables must be placed on an elastic group, so
130 * RegionServers join and leave it automatically. That is the need for auto-managed groups.
131 * <h2>Concurrency</h2> RSGroup state is kept locally in Maps. There is a rsgroup name to cached
132 * RSGroupInfo Map at {@link RSGroupInfoHolder#groupName2Group}. These Maps are persisted to the
133 * hbase:rsgroup table (and cached in zk) on each modification.
134 * <p/>
135 * Mutations on state are synchronized but reads can continue without having to wait on an instance
136 * monitor, mutations do wholesale replace of the Maps on update -- Copy-On-Write; the local Maps of
137 * state are read-only, just-in-case (see flushConfig).
138 * <p/>
139 * Reads must not block else there is a danger we'll deadlock.
140 * <p/>
141 * Clients of this class, the {@link RSGroupAdminEndpoint} for example, want to query and then act
142 * on the results of the query modifying cache in zookeeper without another thread making
143 * intermediate modifications. These clients synchronize on the 'this' instance so no other has
144 * access concurrently. Reads must be able to continue concurrently.
145 */
146@InterfaceAudience.Private
147final class RSGroupInfoManagerImpl implements RSGroupInfoManager {
148  private static final Logger LOG = LoggerFactory.getLogger(RSGroupInfoManagerImpl.class);
149
150  // Assigned before user tables
151  static final TableName RSGROUP_TABLE_NAME =
152    TableName.valueOf(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR, "rsgroup");
153
154  static final String KEEP_ONE_SERVER_IN_DEFAULT_ERROR_MESSAGE =
155    "should keep at least " + "one server in 'default' RSGroup.";
156
157  /** Define the config key of retries threshold when movements failed */
158  static final String FAILED_MOVE_MAX_RETRY = "hbase.rsgroup.move.max.retry";
159
160  /** Define the default number of retries */
161  static final int DEFAULT_MAX_RETRY_VALUE = 50;
162
163  /** Config key prefix; {@code hbase.rsgroup.regex.<groupname>=<regex>} */
164  static final String RS_GROUP_REGEX_PREFIX = "hbase.rsgroup.regex.";
165  private static final Pattern GROUP_NAME_PATTERN = Pattern.compile("[a-zA-Z0-9_]+");
166
167  private static final String RS_GROUP_ZNODE = "rsgroup";
168
169  static final byte[] META_FAMILY_BYTES = Bytes.toBytes("m");
170
171  static final byte[] META_QUALIFIER_BYTES = Bytes.toBytes("i");
172
173  static final String MIGRATE_THREAD_NAME = "Migrate-RSGroup-Tables";
174
175  private static final byte[] ROW_KEY = { 0 };
176
177  /** Table descriptor for <code>hbase:rsgroup</code> catalog table */
178  private static final TableDescriptor RSGROUP_TABLE_DESC;
179  static {
180    TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(RSGROUP_TABLE_NAME)
181      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(META_FAMILY_BYTES))
182      .setRegionSplitPolicyClassName(DisabledRegionSplitPolicy.class.getName());
183    try {
184      builder.setCoprocessor(
185        CoprocessorDescriptorBuilder.newBuilder(MultiRowMutationEndpoint.class.getName())
186          .setPriority(Coprocessor.PRIORITY_SYSTEM).build());
187    } catch (IOException ex) {
188      throw new Error(ex);
189    }
190    RSGROUP_TABLE_DESC = builder.build();
191  }
192
193  // There two Maps are immutable and wholesale replaced on each modification
194  // so are safe to access concurrently. See class comment.
195  private static final class RSGroupInfoHolder {
196    final ImmutableMap<String, RSGroupInfo> groupName2Group;
197    final ImmutableMap<TableName, RSGroupInfo> tableName2Group;
198
199    RSGroupInfoHolder() {
200      this(Collections.emptyMap());
201    }
202
203    RSGroupInfoHolder(Map<String, RSGroupInfo> rsGroupMap) {
204      ImmutableMap.Builder<String, RSGroupInfo> group2Name2GroupBuilder = ImmutableMap.builder();
205      ImmutableMap.Builder<TableName, RSGroupInfo> tableName2GroupBuilder = ImmutableMap.builder();
206      rsGroupMap.forEach((groupName, rsGroupInfo) -> {
207        group2Name2GroupBuilder.put(groupName, rsGroupInfo);
208        if (!groupName.equals(RSGroupInfo.DEFAULT_GROUP)) {
209          rsGroupInfo.getTables()
210            .forEach(tableName -> tableName2GroupBuilder.put(tableName, rsGroupInfo));
211        }
212      });
213      this.groupName2Group = group2Name2GroupBuilder.build();
214      this.tableName2Group = tableName2GroupBuilder.build();
215    }
216  }
217
218  private volatile RSGroupInfoHolder holder = new RSGroupInfoHolder();
219
220  private final MasterServices masterServices;
221  private final AsyncClusterConnection conn;
222  private final ZKWatcher watcher;
223  private final RSGroupStartupWorker rsGroupStartupWorker;
224  private final ServerEventsListener serverEventsListener = new ServerEventsListener();
225  // contains list of groups that were last flushed to persistent store
226  private Set<String> prevRSGroups = new HashSet<>();
227
228  // Package visibility for testing
229  static class RSGroupMappingScript {
230    static final String RS_GROUP_MAPPING_SCRIPT = "hbase.rsgroup.table.mapping.script";
231    static final String RS_GROUP_MAPPING_SCRIPT_TIMEOUT =
232      "hbase.rsgroup.table.mapping.script.timeout";
233
234    private final String script;
235    private final long scriptTimeout;
236
237    RSGroupMappingScript(Configuration conf) {
238      script = conf.get(RS_GROUP_MAPPING_SCRIPT);
239      scriptTimeout = conf.getLong(RS_GROUP_MAPPING_SCRIPT_TIMEOUT, 5000); // 5 seconds
240    }
241
242    String getRSGroup(String namespace, String tablename) {
243      if (script == null || script.isEmpty()) {
244        return null;
245      }
246      Shell.ShellCommandExecutor rsgroupMappingScript =
247        new Shell.ShellCommandExecutor(new String[] { script, "", "" }, null, null, scriptTimeout);
248
249      String[] exec = rsgroupMappingScript.getExecString();
250      exec[1] = namespace;
251      exec[2] = tablename;
252      try {
253        rsgroupMappingScript.execute();
254      } catch (IOException e) {
255        // This exception may happen, like process doesn't have permission to run this script.
256        LOG.error("{}, placing {} back to default rsgroup", e.getMessage(),
257          TableName.valueOf(namespace, tablename));
258        return RSGroupInfo.DEFAULT_GROUP;
259      }
260      return rsgroupMappingScript.getOutput().trim();
261    }
262  }
263
264  private RSGroupMappingScript script;
265
266  private RSGroupInfoManagerImpl(MasterServices masterServices) {
267    this.masterServices = masterServices;
268    this.watcher = masterServices.getZooKeeper();
269    this.conn = masterServices.getAsyncClusterConnection();
270    this.rsGroupStartupWorker = new RSGroupStartupWorker();
271    this.script = new RSGroupMappingScript(masterServices.getConfiguration());
272  }
273
274  /**
275   * Called synchronously from {@code ServerListener#serverAdded}/{@code #serverRemoved}. Server
276   * membership of 'default' and of every regex-governed group is never persisted, so it is
277   * recomputed from the live servers and applied in memory only -- no {@code hbase:rsgroup}/ZK I/O.
278   */
279  private synchronized void handleServerEvent() {
280    Map<String, RSGroupInfo> currentGroups = holder.groupName2Group;
281    Map<String, SortedSet<Address>> autoManagedServers =
282      computeAutoManagedRSGroupServers(currentGroups.values());
283    Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(currentGroups);
284    applyAutoManagedRSGroupServers(newGroupMap, autoManagedServers);
285    dropRegexGovernedServersFromOtherGroups(newGroupMap, autoManagedServers);
286    for (Map.Entry<String, RSGroupInfo> entry : newGroupMap.entrySet()) {
287      if (!entry.getValue().getServers().equals(currentGroups.get(entry.getKey()).getServers())) {
288        resetRSGroupMap(newGroupMap);
289        LOG.info("Updated auto-managed RSGroup servers, {} servers",
290          autoManagedServers.values().stream().mapToInt(SortedSet::size).sum());
291        return;
292      }
293    }
294  }
295
296  private synchronized void init() throws IOException {
297    refresh(false);
298    masterServices.getServerManager().registerListener(serverEventsListener);
299  }
300
301  static RSGroupInfoManager getInstance(MasterServices masterServices) throws IOException {
302    RSGroupInfoManagerImpl instance = new RSGroupInfoManagerImpl(masterServices);
303    instance.init();
304    return instance;
305  }
306
307  public void start() {
308    // create system table of rsgroup
309    rsGroupStartupWorker.start();
310  }
311
312  @Override
313  public synchronized void addRSGroup(RSGroupInfo rsGroupInfo) throws IOException {
314    checkGroupName(rsGroupInfo.getName());
315    Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group;
316    if (
317      rsGroupMap.get(rsGroupInfo.getName()) != null
318        || rsGroupInfo.getName().equals(RSGroupInfo.DEFAULT_GROUP)
319    ) {
320      throw new ConstraintException("Group already exists: " + rsGroupInfo.getName());
321    }
322    Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
323    newGroupMap.put(rsGroupInfo.getName(), rsGroupInfo);
324    flushConfig(newGroupMap);
325    LOG.info("Add group {} done.", rsGroupInfo.getName());
326  }
327
328  private RSGroupInfo getRSGroupInfo(final String groupName) throws ConstraintException {
329    RSGroupInfo rsGroupInfo = holder.groupName2Group.get(groupName);
330    if (rsGroupInfo == null) {
331      throw new ConstraintException("RSGroup " + groupName + " does not exist");
332    }
333    return rsGroupInfo;
334  }
335
336  /** Returns Set of online Servers named for their hostname and port (not ServerName). */
337  private Set<Address> getOnlineServers() {
338    return masterServices.getServerManager().getOnlineServers().keySet().stream()
339      .map(ServerName::getAddress).collect(Collectors.toSet());
340  }
341
342  /**
343   * Returns a copy of {@code src}, safe to mutate without affecting the live {@code
344   * holder.groupName2Group} entry -- callers must not mutate the object returned by
345   * {@link #getRSGroupInfo} or {@link #getRSGroupOfServer} in place, since a subsequent
346   * {@link #flushConfig(Map)} can still reject the change and the live state must stay untouched
347   * until the change is actually persisted.
348   */
349  private static RSGroupInfo copyOf(RSGroupInfo src) {
350    RSGroupInfo copy = new RSGroupInfo(src);
351    // The copy constructor does not copy configuration, so do it by hand.
352    src.getConfiguration().forEach(copy::setConfiguration);
353    return copy;
354  }
355
356  public synchronized Set<Address> moveServers(Set<Address> servers, String srcGroup,
357    String dstGroup) throws IOException {
358    RSGroupInfo src = copyOf(getRSGroupInfo(srcGroup));
359    RSGroupInfo dst = copyOf(getRSGroupInfo(dstGroup));
360    Set<Address> movedServers = new HashSet<>();
361    // If destination is 'default' rsgroup, only add servers that are online. If not online, drop
362    // it. If not 'default' group, add server to 'dst' rsgroup EVEN IF IT IS NOT online (could be a
363    // rsgroup of dead servers that are to come back later).
364    Set<Address> onlineServers =
365      dst.getName().equals(RSGroupInfo.DEFAULT_GROUP) ? getOnlineServers() : null;
366    for (Address el : servers) {
367      src.removeServer(el);
368      if (onlineServers != null) {
369        if (!onlineServers.contains(el)) {
370          if (LOG.isDebugEnabled()) {
371            LOG.debug("Dropping " + el + " during move-to-default RSGroup because not online");
372          }
373          continue;
374        }
375      }
376      dst.addServer(el);
377      movedServers.add(el);
378    }
379    checkMoveAgreesWithRegex(movedServers, dst.getName());
380    Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(holder.groupName2Group);
381    newGroupMap.put(src.getName(), src);
382    newGroupMap.put(dst.getName(), dst);
383    flushConfig(newGroupMap);
384    return movedServers;
385  }
386
387  /**
388   * Regex-based membership always wins: the next server event or refresh puts a server back where
389   * its hostname says, so an explicit move that disagrees would only make regions flap. Such a move
390   * is rejected. A server whose hostname matches no regex (or an ambiguous or dangling one, which
391   * fall back to ordinary placement) may go anywhere except into a regex-governed group.
392   */
393  private void checkMoveAgreesWithRegex(Set<Address> movedServers, String dstGroup)
394    throws ConstraintException {
395    Map<String, Pattern> regexGroupMap = getRegexGroupMap(masterServices.getConfiguration());
396    if (regexGroupMap.isEmpty() || movedServers.isEmpty()) {
397      return;
398    }
399    Map<Address, String> expectedGroups =
400      resolveServerAddrToRSGroupName(movedServers, regexGroupMap, holder.groupName2Group.keySet());
401    for (Address server : movedServers) {
402      String expectedGroup = expectedGroups.get(server);
403      boolean isWrongMove = expectedGroup != null
404        ? !expectedGroup.equals(dstGroup)
405        : regexGroupMap.containsKey(dstGroup);
406      if (isWrongMove) {
407        String reason = expectedGroup != null
408          ? "its membership is decided by " + RS_GROUP_REGEX_PREFIX + expectedGroup
409            + ", under which its hostname belongs in '" + expectedGroup + "'"
410          : "its hostname does not match " + RS_GROUP_REGEX_PREFIX + dstGroup
411            + ", which decides the membership of that RSGroup";
412        throw new ConstraintException("Server " + server + " cannot be moved to RSGroup '"
413          + dstGroup + "': " + reason + ". Change that config instead.");
414      }
415    }
416  }
417
418  @Override
419  public RSGroupInfo getRSGroupOfServer(Address serverHostPort) {
420    for (RSGroupInfo info : holder.groupName2Group.values()) {
421      if (info.containsServer(serverHostPort)) {
422        return info;
423      }
424    }
425    return null;
426  }
427
428  @Override
429  public RSGroupInfo getRSGroup(String groupName) {
430    return holder.groupName2Group.get(groupName);
431  }
432
433  @Override
434  public synchronized void removeRSGroup(String groupName) throws IOException {
435    RSGroupInfo rsGroupInfo = getRSGroupInfo(groupName);
436    int serverCount = rsGroupInfo.getServers().size();
437    if (serverCount > 0) {
438      throw new ConstraintException("RSGroup " + groupName + " has " + serverCount
439        + " servers; you must remove these servers from the RSGroup before"
440        + " the RSGroup can be removed.");
441    }
442    for (TableDescriptor td : masterServices.getTableDescriptors().getAll().values()) {
443      if (td.getRegionServerGroup().map(groupName::equals).orElse(false)) {
444        throw new ConstraintException("RSGroup " + groupName + " is already referenced by "
445          + td.getTableName() + "; you must remove all the tables from the RSGroup before "
446          + "the RSGroup can be removed.");
447      }
448    }
449    for (NamespaceDescriptor ns : masterServices.getClusterSchema().getNamespaces()) {
450      String nsGroup = ns.getConfigurationValue(RSGroupInfo.NAMESPACE_DESC_PROP_GROUP);
451      if (nsGroup != null && nsGroup.equals(groupName)) {
452        throw new ConstraintException(
453          "RSGroup " + groupName + " is referenced by namespace: " + ns.getName());
454      }
455    }
456    Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group;
457    if (!rsGroupMap.containsKey(groupName) || groupName.equals(RSGroupInfo.DEFAULT_GROUP)) {
458      throw new ConstraintException(
459        "Group " + groupName + " does not exist or is a reserved " + "group");
460    }
461    Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
462    newGroupMap.remove(groupName);
463    flushConfig(newGroupMap);
464    LOG.info("Remove group {} done", groupName);
465  }
466
467  @Override
468  public List<RSGroupInfo> listRSGroups() {
469    return Lists.newArrayList(holder.groupName2Group.values());
470  }
471
472  @Override
473  public boolean isOnline() {
474    return rsGroupStartupWorker.isOnline();
475  }
476
477  @Override
478  public synchronized void removeServers(Set<Address> servers) throws IOException {
479    if (servers == null || servers.isEmpty()) {
480      throw new ConstraintException("The set of servers to remove cannot be null or empty.");
481    }
482
483    // check the set of servers
484    checkForDeadOrOnlineServers(servers);
485
486    Map<String, RSGroupInfo> rsGroupInfos = new HashMap<String, RSGroupInfo>();
487    for (Address el : servers) {
488      RSGroupInfo rsGroupInfo = getRSGroupOfServer(el);
489      if (rsGroupInfo != null) {
490        RSGroupInfo newRsGroupInfo = rsGroupInfos.get(rsGroupInfo.getName());
491        if (newRsGroupInfo == null) {
492          newRsGroupInfo = copyOf(rsGroupInfo);
493        }
494        newRsGroupInfo.removeServer(el);
495        rsGroupInfos.put(newRsGroupInfo.getName(), newRsGroupInfo);
496      } else {
497        LOG.warn("Server " + el + " does not belong to any rsgroup.");
498      }
499    }
500
501    if (rsGroupInfos.size() > 0) {
502      Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(holder.groupName2Group);
503      newGroupMap.putAll(rsGroupInfos);
504      flushConfig(newGroupMap);
505    }
506    LOG.info("Remove decommissioned servers {} from RSGroup done", servers);
507  }
508
509  private List<RSGroupInfo> retrieveGroupListFromGroupTable() throws IOException {
510    List<RSGroupInfo> rsGroupInfoList = Lists.newArrayList();
511    AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME);
512    try (ResultScanner scanner = table.getScanner(META_FAMILY_BYTES, META_QUALIFIER_BYTES)) {
513      for (Result result;;) {
514        result = scanner.next();
515        if (result == null) {
516          break;
517        }
518        RSGroupProtos.RSGroupInfo proto = RSGroupProtos.RSGroupInfo
519          .parseFrom(result.getValue(META_FAMILY_BYTES, META_QUALIFIER_BYTES));
520        rsGroupInfoList.add(ProtobufUtil.toGroupInfo(proto));
521      }
522    }
523    return rsGroupInfoList;
524  }
525
526  private List<RSGroupInfo> retrieveGroupListFromZookeeper() throws IOException {
527    String groupBasePath = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE);
528    List<RSGroupInfo> RSGroupInfoList = Lists.newArrayList();
529    // Overwrite any info stored by table, this takes precedence
530    try {
531      if (ZKUtil.checkExists(watcher, groupBasePath) != -1) {
532        List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(watcher, groupBasePath);
533        if (children == null) {
534          return RSGroupInfoList;
535        }
536        for (String znode : children) {
537          byte[] data = ZKUtil.getData(watcher, ZNodePaths.joinZNode(groupBasePath, znode));
538          if (data != null && data.length > 0) {
539            ProtobufUtil.expectPBMagicPrefix(data);
540            ByteArrayInputStream bis =
541              new ByteArrayInputStream(data, ProtobufUtil.lengthOfPBMagic(), data.length);
542            RSGroupInfoList.add(ProtobufUtil.toGroupInfo(RSGroupProtos.RSGroupInfo.parseFrom(bis)));
543          }
544        }
545        LOG.debug("Read ZK GroupInfo count:" + RSGroupInfoList.size());
546      }
547    } catch (KeeperException | DeserializationException | InterruptedException e) {
548      throw new IOException("Failed to read rsGroupZNode", e);
549    }
550    return RSGroupInfoList;
551  }
552
553  private void migrate(Collection<RSGroupInfo> groupList) {
554    TableDescriptors tds = masterServices.getTableDescriptors();
555    ProcedureExecutor<MasterProcedureEnv> procExec = masterServices.getMasterProcedureExecutor();
556    for (RSGroupInfo groupInfo : groupList) {
557      if (groupInfo.getName().equals(RSGroupInfo.DEFAULT_GROUP)) {
558        continue;
559      }
560      SortedSet<TableName> failedTables = new TreeSet<>();
561      List<MigrateRSGroupProcedure> procs = new ArrayList<>();
562      for (TableName tableName : groupInfo.getTables()) {
563        LOG.debug("Migrating {} in group {}", tableName, groupInfo.getName());
564        TableDescriptor oldTd;
565        try {
566          oldTd = tds.get(tableName);
567        } catch (IOException e) {
568          LOG.warn("Failed to migrate {} in group {}", tableName, groupInfo.getName(), e);
569          failedTables.add(tableName);
570          continue;
571        }
572        if (oldTd == null) {
573          continue;
574        }
575        if (oldTd.getRegionServerGroup().isPresent()) {
576          // either we have already migrated it or that user has set the rs group using the new
577          // code which will set the group directly on table descriptor, skip.
578          LOG.debug("Skip migrating {} since it is already in group {}", tableName,
579            oldTd.getRegionServerGroup().get());
580          continue;
581        }
582        // This is a bit tricky. Since we know that the region server group config in
583        // TableDescriptor will only be used at master side, it is fine to just update the table
584        // descriptor on file system and also the cache, without reopening all the regions. This
585        // will be much faster than the normal modifyTable. And when upgrading, we will update
586        // master first and then region server, so after all the region servers has been reopened,
587        // the new TableDescriptor will be loaded.
588        MigrateRSGroupProcedure proc =
589          new MigrateRSGroupProcedure(procExec.getEnvironment(), tableName);
590        procExec.submitProcedure(proc);
591        procs.add(proc);
592      }
593      for (MigrateRSGroupProcedure proc : procs) {
594        try {
595          ProcedureSyncWait.waitForProcedureToComplete(procExec, proc, 60000);
596        } catch (IOException e) {
597          LOG.warn("Failed to migrate rs group {} for table {}", groupInfo.getName(),
598            proc.getTableName());
599          failedTables.add(proc.getTableName());
600        }
601      }
602      LOG.debug("Done migrating {}, failed tables {}", groupInfo.getName(), failedTables);
603      synchronized (RSGroupInfoManagerImpl.this) {
604        Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group;
605        RSGroupInfo currentInfo = rsGroupMap.get(groupInfo.getName());
606        if (currentInfo != null) {
607          RSGroupInfo newInfo =
608            new RSGroupInfo(currentInfo.getName(), currentInfo.getServers(), failedTables);
609          Map<String, RSGroupInfo> newGroupMap = new HashMap<>(rsGroupMap);
610          newGroupMap.put(groupInfo.getName(), newInfo);
611          try {
612            flushConfig(newGroupMap);
613          } catch (IOException e) {
614            LOG.warn("Failed to persist rs group {}", newInfo.getName(), e);
615          }
616        }
617      }
618    }
619  }
620
621  // Migrate the table rs group info from RSGroupInfo into the table descriptor
622  // Notice that we do not want to block the initialize so this will be done in background, and
623  // during the migrating, the rs group info maybe incomplete and cause region to be misplaced.
624  private void migrate() {
625    Thread migrateThread = new Thread(MIGRATE_THREAD_NAME) {
626
627      @Override
628      public void run() {
629        LOG.info("Start migrating table rs group config");
630        while (!masterServices.isStopped()) {
631          Collection<RSGroupInfo> groups = holder.groupName2Group.values();
632          boolean hasTables = groups.stream().anyMatch(r -> !r.getTables().isEmpty());
633          if (!hasTables) {
634            break;
635          }
636          migrate(groups);
637        }
638        LOG.info("Done migrating table rs group info");
639      }
640    };
641    migrateThread.setDaemon(true);
642    migrateThread.start();
643  }
644
645  /**
646   * Read rsgroup info from the source of truth, the hbase:rsgroup table. Update zk cache. Called on
647   * startup of the manager.
648   */
649  private synchronized void refresh(boolean forceOnline) throws IOException {
650    LOG.debug("Refreshing RSGroup info from source of truth: forceOnline={}, isOnline={}",
651      forceOnline, isOnline());
652    List<RSGroupInfo> groupList = new ArrayList<>();
653
654    // Overwrite anything read from zk, group table is source of truth
655    // if online read from GROUP table
656    if (forceOnline || isOnline()) {
657      LOG.debug("Refreshing in Online mode.");
658      groupList.addAll(retrieveGroupListFromGroupTable());
659    } else {
660      LOG.debug("Refreshing in Offline mode.");
661      groupList.addAll(retrieveGroupListFromZookeeper());
662    }
663
664    // This is added to the last of the list so it overwrites the 'default' rsgroup loaded
665    // from region group table or zk
666    groupList.add(new RSGroupInfo(RSGroupInfo.DEFAULT_GROUP));
667
668    // populate the data
669    HashMap<String, RSGroupInfo> newGroupMap = Maps.newHashMap();
670    for (RSGroupInfo group : groupList) {
671      newGroupMap.put(group.getName(), group);
672    }
673    // Server membership for 'default' and for every regex-governed group is always recomputed
674    // on the fly -- never persisted nor trusted from storage.
675    Map<String, SortedSet<Address>> autoManagedServers =
676      computeAutoManagedRSGroupServers(groupList);
677    for (Map.Entry<String, SortedSet<Address>> entry : autoManagedServers.entrySet()) {
678      RSGroupInfo stored = newGroupMap.get(entry.getKey());
679      if (!RSGroupInfo.DEFAULT_GROUP.equals(entry.getKey()) && !stored.getServers().isEmpty()) {
680        LOG.info(
681          "Ignoring servers {} stored for regex-governed RSGroup '{}'; its membership is "
682            + "computed on the fly and the stored copy is cleared on the next flush",
683          stored.getServers(), entry.getKey());
684      }
685    }
686    applyAutoManagedRSGroupServers(newGroupMap, autoManagedServers);
687    dropRegexGovernedServersFromOtherGroups(newGroupMap, autoManagedServers);
688    resetRSGroupMap(newGroupMap);
689    updateCacheOfRSGroups(newGroupMap.keySet());
690    LOG.info("Refresh completed successfully");
691  }
692
693  private void flushConfigTable(Map<String, RSGroupInfo> groupMap) throws IOException {
694    List<Mutation> mutations = Lists.newArrayList();
695
696    // populate deletes
697    for (String groupName : prevRSGroups) {
698      if (!groupMap.containsKey(groupName)) {
699        Delete d = new Delete(Bytes.toBytes(groupName));
700        mutations.add(d);
701      }
702    }
703
704    // populate puts
705    for (RSGroupInfo gi : groupMap.values()) {
706      if (!gi.getName().equals(RSGroupInfo.DEFAULT_GROUP)) {
707        RSGroupProtos.RSGroupInfo proto = ProtobufUtil.toProtoGroupInfo(gi);
708        Put p = new Put(Bytes.toBytes(gi.getName()));
709        p.addColumn(META_FAMILY_BYTES, META_QUALIFIER_BYTES, proto.toByteArray());
710        mutations.add(p);
711      }
712    }
713
714    if (mutations.size() > 0) {
715      multiMutate(mutations);
716    }
717  }
718
719  private synchronized void flushConfig() throws IOException {
720    flushConfig(holder.groupName2Group);
721  }
722
723  private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) throws IOException {
724    // For offline mode persistence is still unavailable
725    // We're refreshing in-memory state but only for servers in default group
726    if (!isOnline()) {
727      if (newGroupMap == holder.groupName2Group) {
728        // When newGroupMap is this.rsGroupMap itself,
729        // do not need to check default group and other groups as followed
730        return;
731      }
732
733      LOG.debug("Offline mode, cannot persist to {}", RSGROUP_TABLE_NAME);
734
735      Map<String, RSGroupInfo> oldGroupMap = Maps.newHashMap(holder.groupName2Group);
736      RSGroupInfo oldDefaultGroup = oldGroupMap.remove(RSGroupInfo.DEFAULT_GROUP);
737      RSGroupInfo newDefaultGroup = newGroupMap.remove(RSGroupInfo.DEFAULT_GROUP);
738      // Servers of regex-governed groups are computed on the fly and never persisted, so their
739      // changes are allowed here too.
740      if (
741        !withoutRegexGovernedServers(oldGroupMap).equals(withoutRegexGovernedServers(newGroupMap))
742          /* compare both tables and servers in other groups */ || !oldDefaultGroup.getTables()
743            .equals(newDefaultGroup.getTables())
744        /* compare tables in default group */
745      ) {
746        throw new IOException("Only servers in the default group and in regex-governed groups can "
747          + "be updated during offline mode");
748      }
749
750      // Restore newGroupMap by putting its default group back
751      newGroupMap.put(RSGroupInfo.DEFAULT_GROUP, newDefaultGroup);
752
753      // Refresh rsGroupMap
754      // according to the inputted newGroupMap (an updated copy of rsGroupMap)
755      this.holder = new RSGroupInfoHolder(newGroupMap);
756
757      LOG.debug("New RSGroup map: {}", newGroupMap);
758
759      // Do not need to update tableMap
760      // because only the update on servers in default group is allowed above,
761      // or IOException will be thrown
762      return;
763    }
764
765    /* For online mode, persist to hbase:rsgroup and Zookeeper */
766    LOG.debug("Online mode, persisting to {} and ZK", RSGROUP_TABLE_NAME);
767    // Servers of regex-governed groups are computed on the fly and never persisted
768    Map<String, RSGroupInfo> persistedGroupMap = withoutRegexGovernedServers(newGroupMap);
769    flushConfigTable(persistedGroupMap);
770
771    // Make changes visible after having been persisted to the source of truth
772    resetRSGroupMap(newGroupMap);
773    saveRSGroupMapToZK(persistedGroupMap);
774    updateCacheOfRSGroups(newGroupMap.keySet());
775    LOG.info("Flush config done, new RSGroup map: {}", newGroupMap);
776  }
777
778  /**
779   * Returns a copy of {@code groupMap} where every regex-governed group has an empty server set.
780   * Their membership is recomputed from the live servers (see {@link #refresh}), so persisting it
781   * would only add I/O and leave stale data behind.
782   */
783  private Map<String, RSGroupInfo> withoutRegexGovernedServers(Map<String, RSGroupInfo> groupMap) {
784    Set<String> regexGovernedGroups = getRegexGroupMap(masterServices.getConfiguration()).keySet();
785    Map<String, RSGroupInfo> result = Maps.newHashMap(groupMap);
786    for (String groupName : regexGovernedGroups) {
787      RSGroupInfo info = groupMap.get(groupName);
788      if (info == null || info.getServers().isEmpty()) {
789        continue;
790      }
791      RSGroupInfo stripped = new RSGroupInfo(groupName);
792      stripped.addAllTables(info.getTables());
793      info.getConfiguration().forEach(stripped::setConfiguration);
794      result.put(groupName, stripped);
795    }
796    return result;
797  }
798
799  private void saveRSGroupMapToZK(Map<String, RSGroupInfo> newGroupMap) throws IOException {
800    LOG.debug("Saving RSGroup info to ZK");
801    try {
802      String groupBasePath =
803        ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE);
804      ZKUtil.createAndFailSilent(watcher, groupBasePath, ProtobufMagic.PB_MAGIC);
805
806      List<ZKUtil.ZKUtilOp> zkOps = new ArrayList<>(newGroupMap.size());
807      for (String groupName : prevRSGroups) {
808        if (!newGroupMap.containsKey(groupName)) {
809          String znode = ZNodePaths.joinZNode(groupBasePath, groupName);
810          zkOps.add(ZKUtil.ZKUtilOp.deleteNodeFailSilent(znode));
811        }
812      }
813
814      for (RSGroupInfo gi : newGroupMap.values()) {
815        if (!gi.getName().equals(RSGroupInfo.DEFAULT_GROUP)) {
816          String znode = ZNodePaths.joinZNode(groupBasePath, gi.getName());
817          RSGroupProtos.RSGroupInfo proto = ProtobufUtil.toProtoGroupInfo(gi);
818          LOG.debug("Updating znode: " + znode);
819          ZKUtil.createAndFailSilent(watcher, znode);
820          zkOps.add(ZKUtil.ZKUtilOp.deleteNodeFailSilent(znode));
821          zkOps.add(ZKUtil.ZKUtilOp.createAndFailSilent(znode,
822            ProtobufUtil.prependPBMagic(proto.toByteArray())));
823        }
824      }
825      LOG.debug("Writing ZK GroupInfo count: " + zkOps.size());
826
827      ZKUtil.multiOrSequential(watcher, zkOps, false);
828    } catch (KeeperException e) {
829      LOG.error("Failed to write to rsGroupZNode", e);
830      masterServices.abort("Failed to write to rsGroupZNode", e);
831      throw new IOException("Failed to write to rsGroupZNode", e);
832    }
833  }
834
835  /**
836   * Make changes visible. Caller must be synchronized on 'this'.
837   */
838  private void resetRSGroupMap(Map<String, RSGroupInfo> newRSGroupMap) {
839    this.holder = new RSGroupInfoHolder(newRSGroupMap);
840  }
841
842  /**
843   * Update cache of rsgroups. Caller must be synchronized on 'this'.
844   * @param currentGroups Current list of Groups.
845   */
846  private void updateCacheOfRSGroups(final Set<String> currentGroups) {
847    this.prevRSGroups.clear();
848    this.prevRSGroups.addAll(currentGroups);
849  }
850
851  /**
852   * Parses {@code hbase.rsgroup.regex.<groupname>=<regex>} entries out of {@code conf}. Entries
853   * with an invalid group name, targeting the reserved {@link RSGroupInfo#DEFAULT_GROUP}, or with
854   * an unparsable regex are logged and skipped.
855   */
856  static Map<String, Pattern> getRegexGroupMap(Configuration conf) {
857    Map<String, String> rsGroupNameToRegexMap = conf.getPropsWithPrefix(RS_GROUP_REGEX_PREFIX);
858    Map<String, Pattern> rsGroupNameToPatternMap = new HashMap<>();
859    for (Map.Entry<String, String> e : rsGroupNameToRegexMap.entrySet()) {
860      if (e.getValue() == null) {
861        // getPropsWithPrefix reads names and values separately, so an entry removed by a concurrent
862        // configuration reload can show up here with a null value. Treat it as unset.
863        continue;
864      }
865      if (!GROUP_NAME_PATTERN.matcher(e.getKey()).matches()) {
866        LOG.warn("Ignoring {}{} -- '{}' is not a valid RSGroup name (it must match {})",
867          RS_GROUP_REGEX_PREFIX, e.getKey(), e.getKey(), GROUP_NAME_PATTERN.pattern());
868        continue;
869      }
870      if (RSGroupInfo.DEFAULT_GROUP.equals(e.getKey())) {
871        LOG.warn("Ignoring {}{} -- regex-based membership cannot target the reserved '{}' group",
872          RS_GROUP_REGEX_PREFIX, e.getKey(), RSGroupInfo.DEFAULT_GROUP);
873        continue;
874      }
875      try {
876        rsGroupNameToPatternMap.put(e.getKey(), Pattern.compile(e.getValue()));
877      } catch (PatternSyntaxException ex) {
878        LOG.warn("Invalid regex '{}' for RSGroup '{}' ({}{}); ignoring this entry", e.getValue(),
879          e.getKey(), RS_GROUP_REGEX_PREFIX, e.getKey());
880      }
881    }
882    LOG.debug("Resolved regex-based RSGroup membership config: {}", rsGroupNameToPatternMap);
883    return rsGroupNameToPatternMap;
884  }
885
886  private static List<String> getMatchingRSGroupNames(String hostname,
887    Map<String, Pattern> rsGroupNameToPatternMap) {
888    List<String> rsGroupNames = new ArrayList<>();
889    for (Map.Entry<String, Pattern> e : rsGroupNameToPatternMap.entrySet()) {
890      if (e.getValue().matcher(hostname).matches()) {
891        rsGroupNames.add(e.getKey());
892      }
893    }
894    return rsGroupNames;
895  }
896
897  /**
898   * Resolves every server in {@code onlineServers} whose hostname unambiguously matches exactly one
899   * entry of {@code rsGroupNameToPatternMap} naming a group in {@code existingGroupNames}. A server
900   * matching multiple regexes, or matching a regex for a group that does not (yet) exist, is
901   * omitted (warn-only) rather than resolved.
902   */
903  static Map<Address, String> resolveServerAddrToRSGroupName(Set<Address> onlineServers,
904    Map<String, Pattern> rsGroupNameToPatternMap, Set<String> existingGroupNames) {
905    Map<Address, String> serverAddrToRSGroupName = new HashMap<>();
906    if (rsGroupNameToPatternMap.isEmpty()) {
907      return serverAddrToRSGroupName;
908    }
909    Set<String> nonExistingRSGroupNames = new HashSet<>();
910    for (Address server : onlineServers) {
911      List<String> matchingRSGroupNames =
912        getMatchingRSGroupNames(server.getHostName(), rsGroupNameToPatternMap);
913      if (matchingRSGroupNames.isEmpty()) {
914        continue;
915      }
916      if (matchingRSGroupNames.size() > 1) {
917        LOG.warn("Server {} hostname matches regexes for multiple RSGroups {}; treating it as "
918          + "unmatched by regex (falls back to default) until the overlapping "
919          + "hbase.rsgroup.regex.* entries are fixed", server, matchingRSGroupNames);
920        continue;
921      }
922      String rsGroupName = matchingRSGroupNames.get(0);
923      if (existingGroupNames.contains(rsGroupName)) {
924        serverAddrToRSGroupName.put(server, rsGroupName);
925      } else if (nonExistingRSGroupNames.add(rsGroupName)) {
926        LOG.warn(
927          "Config {}{} matches server {} but RSGroup '{}' does not exist -- create it with "
928            + "addRSGroup first; treating this server as unmatched by regex until then",
929          RS_GROUP_REGEX_PREFIX, rsGroupName, server, rsGroupName);
930      }
931    }
932    LOG.debug("Resolved server address to RSGroup name map: {}", serverAddrToRSGroupName);
933    return serverAddrToRSGroupName;
934  }
935
936  /** Returns {@code true} if every online server is either admin-managed or regex-matched. */
937  static boolean wouldEmptyDefaultGroup(Set<Address> onlineServers,
938    Set<Address> adminManagedServers, Set<Address> regexMatchedServers) {
939    for (Address server : onlineServers) {
940      if (!adminManagedServers.contains(server) && !regexMatchedServers.contains(server)) {
941        return false;
942      }
943    }
944    return !onlineServers.isEmpty();
945  }
946
947  static final class RegexBasedRSGroupMembershipResolution {
948    final Set<Address> adminManagedServers;
949    final Map<Address, String> regexMatchedServers;
950
951    RegexBasedRSGroupMembershipResolution(Set<Address> adminManagedServers,
952      Map<Address, String> regexMatchedServers) {
953      this.adminManagedServers = adminManagedServers;
954      this.regexMatchedServers = regexMatchedServers;
955    }
956  }
957
958  static RegexBasedRSGroupMembershipResolution resolveRegexBasedRSGroupMembership(
959    Map<String, Pattern> rsGroupNameToPatternMap, Set<Address> onlineServers,
960    Collection<RSGroupInfo> existingGroups) {
961    Set<String> existingGroupNames =
962      existingGroups.stream().map(RSGroupInfo::getName).collect(Collectors.toSet());
963    Set<Address> adminManagedServers = new HashSet<>();
964    for (RSGroupInfo existingGroup : existingGroups) {
965      if (
966        !RSGroupInfo.DEFAULT_GROUP.equals(existingGroup.getName())
967          && !rsGroupNameToPatternMap.containsKey(existingGroup.getName())
968      ) {
969        adminManagedServers.addAll(existingGroup.getServers());
970      }
971    }
972    Map<Address, String> regexMatchedServers =
973      resolveServerAddrToRSGroupName(onlineServers, rsGroupNameToPatternMap, existingGroupNames);
974    // A server stored in an admin-managed group that a regex now claims belongs to the regex group.
975    adminManagedServers.removeAll(regexMatchedServers.keySet());
976    if (wouldEmptyDefaultGroup(onlineServers, adminManagedServers, regexMatchedServers.keySet())) {
977      // Regex placement is still applied: every server lands in its correct group. Operators are
978      // expected to configure regexes carefully, so this is a warning, not an error.
979      LOG.warn(
980        "Regex-based RSGroup membership leaves RSGroup '{}' with no online servers -- every "
981          + "online server is either admin-managed or matches a {}* entry. Tables in RSGroup '{}' "
982          + "(including system tables, unless they are in another RSGroup) have no server of "
983          + "their own to be assigned to; set {} to let them fall back to servers of other "
984          + "RSGroups",
985        RSGroupInfo.DEFAULT_GROUP, RS_GROUP_REGEX_PREFIX, RSGroupInfo.DEFAULT_GROUP,
986        RSGroupBasedLoadBalancer.FALLBACK_GROUP_ENABLE_KEY);
987    }
988    LOG.debug(
989      "Regex-based RSGroup membership resolution: onlineServers={}, "
990        + "adminManagedServers={}, regexMatchedServers={}",
991      onlineServers, adminManagedServers, regexMatchedServers);
992    return new RegexBasedRSGroupMembershipResolution(adminManagedServers, regexMatchedServers);
993  }
994
995  private Map<String, SortedSet<Address>>
996    computeAutoManagedRSGroupServers(Collection<RSGroupInfo> existingGroups) {
997    Set<Address> onlineServers = getOnlineServers();
998    Map<String, Pattern> rsGroupNameToPatternMap =
999      getRegexGroupMap(masterServices.getConfiguration());
1000    RegexBasedRSGroupMembershipResolution resolution =
1001      resolveRegexBasedRSGroupMembership(rsGroupNameToPatternMap, onlineServers, existingGroups);
1002
1003    Set<String> existingGroupNames =
1004      existingGroups.stream().map(RSGroupInfo::getName).collect(Collectors.toSet());
1005    Map<String, SortedSet<Address>> result = new HashMap<>();
1006    for (String rsGroupName : rsGroupNameToPatternMap.keySet()) {
1007      if (existingGroupNames.contains(rsGroupName)) {
1008        result.put(rsGroupName, new TreeSet<>());
1009      }
1010    }
1011    result.put(RSGroupInfo.DEFAULT_GROUP, new TreeSet<>());
1012
1013    for (Address server : onlineServers) {
1014      String regexGroupName = resolution.regexMatchedServers.get(server);
1015      if (regexGroupName != null) {
1016        result.get(regexGroupName).add(server);
1017        continue;
1018      }
1019      if (resolution.adminManagedServers.contains(server)) {
1020        continue;
1021      }
1022      result.get(RSGroupInfo.DEFAULT_GROUP).add(server);
1023    }
1024    for (Map.Entry<String, SortedSet<Address>> entry : result.entrySet()) {
1025      if (entry.getValue().isEmpty() && !RSGroupInfo.DEFAULT_GROUP.equals(entry.getKey())) {
1026        LOG.warn(
1027          "Regex-governed RSGroup '{}' has no online servers; any table assigned to it has "
1028            + "nowhere to go (check {}{} and the online servers' hostnames)",
1029          entry.getKey(), RS_GROUP_REGEX_PREFIX, entry.getKey());
1030      }
1031    }
1032    LOG.debug("Computed auto-managed RSGroup server membership: {}", result);
1033    return result;
1034  }
1035
1036  private static void applyAutoManagedRSGroupServers(Map<String, RSGroupInfo> groupMap,
1037    Map<String, SortedSet<Address>> newRSGroupToServers) {
1038    for (Map.Entry<String, SortedSet<Address>> entry : newRSGroupToServers.entrySet()) {
1039      RSGroupInfo oldInfo = groupMap.get(entry.getKey());
1040      if (oldInfo == null) {
1041        continue;
1042      }
1043      RSGroupInfo newInfo = new RSGroupInfo(entry.getKey(), entry.getValue());
1044      newInfo.addAllTables(oldInfo.getTables());
1045      oldInfo.getConfiguration().forEach(newInfo::setConfiguration);
1046      groupMap.put(entry.getKey(), newInfo);
1047    }
1048  }
1049
1050  /**
1051   * Stored membership of admin-managed groups is not recomputed on refresh, so a server that was
1052   * placed in one of them before a regex claimed it would end up in two groups. Unlike the runtime
1053   * paths, refresh cannot reject this: it runs in the startup worker, which retries forever on
1054   * failure and would keep the manager offline. Instead, resolve it the way
1055   * {@link #computeAutoManagedRSGroupServers} already does -- the regex wins -- in memory only;
1056   * storage is corrected by the next flush.
1057   */
1058  private static void dropRegexGovernedServersFromOtherGroups(Map<String, RSGroupInfo> groupMap,
1059    Map<String, SortedSet<Address>> autoManagedServers) {
1060    Map<Address, String> regexGovernedServers = new HashMap<>();
1061    for (Map.Entry<String, SortedSet<Address>> entry : autoManagedServers.entrySet()) {
1062      if (!RSGroupInfo.DEFAULT_GROUP.equals(entry.getKey())) {
1063        entry.getValue().forEach(server -> regexGovernedServers.put(server, entry.getKey()));
1064      }
1065    }
1066    if (regexGovernedServers.isEmpty()) {
1067      return;
1068    }
1069    for (Map.Entry<String, RSGroupInfo> entry : groupMap.entrySet()) {
1070      RSGroupInfo info = entry.getValue();
1071      if (autoManagedServers.containsKey(info.getName())) {
1072        continue;
1073      }
1074      Set<Address> overlap = new TreeSet<>();
1075      for (Address server : info.getServers()) {
1076        if (regexGovernedServers.containsKey(server)) {
1077          overlap.add(server);
1078        }
1079      }
1080      if (overlap.isEmpty()) {
1081        continue;
1082      }
1083      LOG.warn(
1084        "Servers {} are stored in RSGroup '{}' but match a {}* entry; placing them in the "
1085          + "regex-governed RSGroup instead and dropping the stored copy",
1086        overlap, info.getName(), RS_GROUP_REGEX_PREFIX);
1087      SortedSet<Address> kept = new TreeSet<>(info.getServers());
1088      kept.removeAll(overlap);
1089      RSGroupInfo newInfo = new RSGroupInfo(info.getName(), kept);
1090      newInfo.addAllTables(info.getTables());
1091      info.getConfiguration().forEach(newInfo::setConfiguration);
1092      entry.setValue(newInfo);
1093    }
1094  }
1095
1096  private class ServerEventsListener implements ServerListener {
1097    private final Logger LOG = LoggerFactory.getLogger(ServerEventsListener.class);
1098
1099    @Override
1100    public void serverAdded(ServerName serverName) {
1101      LOG.debug("Server added: {}", serverName);
1102      RSGroupInfoManagerImpl.this.handleServerEvent();
1103    }
1104
1105    @Override
1106    public void serverRemoved(ServerName serverName) {
1107      LOG.debug("Server removed: {}", serverName);
1108      RSGroupInfoManagerImpl.this.handleServerEvent();
1109    }
1110  }
1111
1112  private class RSGroupStartupWorker extends Thread {
1113    private final Logger LOG = LoggerFactory.getLogger(RSGroupStartupWorker.class);
1114    private volatile boolean online = false;
1115
1116    RSGroupStartupWorker() {
1117      super(RSGroupStartupWorker.class.getName() + "-" + masterServices.getServerName());
1118      setDaemon(true);
1119    }
1120
1121    @Override
1122    public void run() {
1123      if (waitForGroupTableOnline()) {
1124        LOG.info("GroupBasedLoadBalancer is now online");
1125      } else {
1126        LOG.warn("Quit without making region group table online");
1127      }
1128    }
1129
1130    private boolean waitForGroupTableOnline() {
1131      while (isMasterRunning(masterServices)) {
1132        try {
1133          TableStateManager tsm = masterServices.getTableStateManager();
1134          if (!tsm.isTablePresent(RSGROUP_TABLE_NAME)) {
1135            createRSGroupTable();
1136          }
1137          // try reading from the table
1138          FutureUtils.get(conn.getTable(RSGROUP_TABLE_NAME).get(new Get(ROW_KEY)));
1139          LOG.info("RSGroup table={} is online, refreshing cached information", RSGROUP_TABLE_NAME);
1140          RSGroupInfoManagerImpl.this.refresh(true);
1141          online = true;
1142          // flush any inconsistencies between ZK and HTable
1143          RSGroupInfoManagerImpl.this.flushConfig();
1144          // migrate after we are online.
1145          migrate();
1146          return true;
1147        } catch (Exception e) {
1148          LOG.warn("Failed to perform check", e);
1149          // 100ms is short so let's just ignore the interrupt
1150          Threads.sleepWithoutInterrupt(100);
1151        }
1152      }
1153      return false;
1154    }
1155
1156    private void createRSGroupTable() throws IOException {
1157      OptionalLong optProcId = masterServices.getProcedures().stream()
1158        .filter(p -> p instanceof CreateTableProcedure).map(p -> (CreateTableProcedure) p)
1159        .filter(p -> p.getTableName().equals(RSGROUP_TABLE_NAME)).mapToLong(Procedure::getProcId)
1160        .findFirst();
1161      long procId;
1162      if (optProcId.isPresent()) {
1163        procId = optProcId.getAsLong();
1164      } else {
1165        LOG.debug("Creating group table {}", RSGROUP_TABLE_NAME);
1166        procId = masterServices.createSystemTable(RSGROUP_TABLE_DESC);
1167      }
1168      // wait for region to be online
1169      int tries = 600;
1170      while (
1171        !(masterServices.getMasterProcedureExecutor().isFinished(procId))
1172          && masterServices.getMasterProcedureExecutor().isRunning() && tries > 0
1173      ) {
1174        try {
1175          Thread.sleep(100);
1176        } catch (InterruptedException e) {
1177          throw new IOException("Wait interrupted ", e);
1178        }
1179        tries--;
1180      }
1181      if (tries <= 0) {
1182        throw new IOException("Failed to create group table in a given time.");
1183      } else {
1184        Procedure<?> result = masterServices.getMasterProcedureExecutor().getResult(procId);
1185        if (result != null && result.isFailed()) {
1186          throw new IOException(
1187            "Failed to create group table. " + MasterProcedureUtil.unwrapRemoteIOException(result));
1188        }
1189      }
1190    }
1191
1192    public boolean isOnline() {
1193      return online;
1194    }
1195  }
1196
1197  private static boolean isMasterRunning(MasterServices masterServices) {
1198    return !masterServices.isAborted() && !masterServices.isStopped();
1199  }
1200
1201  private void multiMutate(List<Mutation> mutations) throws IOException {
1202    MutateRowsRequest.Builder builder = MutateRowsRequest.newBuilder();
1203    for (Mutation mutation : mutations) {
1204      if (mutation instanceof Put) {
1205        builder
1206          .addMutationRequest(ProtobufUtil.toMutation(MutationProto.MutationType.PUT, mutation));
1207      } else if (mutation instanceof Delete) {
1208        builder
1209          .addMutationRequest(ProtobufUtil.toMutation(MutationProto.MutationType.DELETE, mutation));
1210      } else {
1211        throw new DoNotRetryIOException(
1212          "multiMutate doesn't support " + mutation.getClass().getName());
1213      }
1214    }
1215    MutateRowsRequest request = builder.build();
1216    AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME);
1217    LOG.debug("Multimutating {} with {} mutations", RSGROUP_TABLE_NAME, mutations.size());
1218    FutureUtils.get(table.<MultiRowMutationService, MutateRowsResponse> coprocessorService(
1219      MultiRowMutationService::newStub,
1220      (stub, controller, done) -> stub.mutateRows(controller, request, done), ROW_KEY));
1221    LOG.info("Multimutating {} with {} mutations done", RSGROUP_TABLE_NAME, mutations.size());
1222  }
1223
1224  private void checkGroupName(String groupName) throws ConstraintException {
1225    if (!GROUP_NAME_PATTERN.matcher(groupName).matches()) {
1226      throw new ConstraintException(
1227        "RSGroup name '" + groupName + "' must match " + GROUP_NAME_PATTERN.pattern());
1228    }
1229  }
1230
1231  @Override
1232  public RSGroupInfo getRSGroupForTable(TableName tableName) throws IOException {
1233    return holder.tableName2Group.get(tableName);
1234  }
1235
1236  /**
1237   * Check if the set of servers are belong to dead servers list or online servers list.
1238   * @param servers servers to remove
1239   */
1240  private void checkForDeadOrOnlineServers(Set<Address> servers) throws IOException {
1241    // This ugliness is because we only have Address, not ServerName.
1242    Set<Address> onlineServers = new HashSet<>();
1243    List<ServerName> drainingServers = masterServices.getServerManager().getDrainingServersList();
1244    for (ServerName server : masterServices.getServerManager().getOnlineServers().keySet()) {
1245      // Only online but not decommissioned servers are really online
1246      if (!drainingServers.contains(server)) {
1247        onlineServers.add(server.getAddress());
1248      }
1249    }
1250
1251    Set<Address> deadServers = new HashSet<>();
1252    for (ServerName server : masterServices.getServerManager().getDeadServers().copyServerNames()) {
1253      deadServers.add(server.getAddress());
1254    }
1255
1256    for (Address address : servers) {
1257      if (onlineServers.contains(address)) {
1258        throw new DoNotRetryIOException(
1259          "Server " + address + " is an online server, not allowed to remove.");
1260      }
1261      if (deadServers.contains(address)) {
1262        throw new DoNotRetryIOException("Server " + address + " is on the dead servers list,"
1263          + " Maybe it will come back again, not allowed to remove.");
1264      }
1265    }
1266  }
1267
1268  private void checkOnlineServersOnly(Set<Address> servers) throws IOException {
1269    // This uglyness is because we only have Address, not ServerName.
1270    // Online servers are keyed by ServerName.
1271    Set<Address> onlineServers = new HashSet<>();
1272    for (ServerName server : masterServices.getServerManager().getOnlineServers().keySet()) {
1273      onlineServers.add(server.getAddress());
1274    }
1275    for (Address address : servers) {
1276      if (!onlineServers.contains(address)) {
1277        throw new DoNotRetryIOException(
1278          "Server " + address + " is not an online server in 'default' RSGroup.");
1279      }
1280    }
1281  }
1282
1283  /** Returns List of Regions associated with this <code>server</code>. */
1284  private List<RegionInfo> getRegions(final Address server) {
1285    LinkedList<RegionInfo> regions = new LinkedList<>();
1286    for (Map.Entry<RegionInfo, ServerName> el : masterServices.getAssignmentManager()
1287      .getRegionStates().getRegionAssignments().entrySet()) {
1288      if (el.getValue() == null) {
1289        continue;
1290      }
1291
1292      if (el.getValue().getAddress().equals(server)) {
1293        addRegion(regions, el.getKey());
1294      }
1295    }
1296    for (RegionStateNode state : masterServices.getAssignmentManager().getRegionsInTransition()) {
1297      if (
1298        state.getRegionLocation() != null && state.getRegionLocation().getAddress().equals(server)
1299      ) {
1300        addRegion(regions, state.getRegionInfo());
1301      }
1302    }
1303    return regions;
1304  }
1305
1306  private void addRegion(final LinkedList<RegionInfo> regions, RegionInfo hri) {
1307    // If meta, move it last otherwise other unassigns fail because meta is not
1308    // online for them to update state in. This is dodgy. Needs to be made more
1309    // robust. See TODO below.
1310    if (hri.isMetaRegion()) {
1311      regions.addLast(hri);
1312    } else {
1313      regions.addFirst(hri);
1314    }
1315  }
1316
1317  /**
1318   * Move every region from servers which are currently located on these servers, but should not be
1319   * located there.
1320   * @param movedServers    the servers that are moved to new group
1321   * @param srcGrpServers   all servers in the source group, excluding the movedServers
1322   * @param targetGroupName the target group
1323   * @param sourceGroupName the source group
1324   * @throws IOException if moving the server and tables fail
1325   */
1326  private void moveServerRegionsFromGroup(Set<Address> movedServers, Set<Address> srcGrpServers,
1327    String targetGroupName, String sourceGroupName) throws IOException {
1328    moveRegionsBetweenGroups(movedServers, srcGrpServers, targetGroupName, sourceGroupName,
1329      rs -> getRegions(rs), info -> {
1330        try {
1331          String groupName = RSGroupUtil.getRSGroupInfo(masterServices, this, info.getTable())
1332            .map(RSGroupInfo::getName).orElse(RSGroupInfo.DEFAULT_GROUP);
1333          return groupName.equals(targetGroupName);
1334        } catch (IOException e) {
1335          LOG.warn("Failed to test group for region {} and target group {}", info, targetGroupName);
1336          return false;
1337        }
1338      });
1339  }
1340
1341  private <T> void moveRegionsBetweenGroups(Set<T> regionsOwners, Set<Address> newRegionsOwners,
1342    String targetGroupName, String sourceGroupName, Function<T, List<RegionInfo>> getRegionsInfo,
1343    Function<RegionInfo, Boolean> validation) throws IOException {
1344    // Get server names corresponding to given Addresses
1345    List<ServerName> movedServerNames = new ArrayList<>(regionsOwners.size());
1346    List<ServerName> srcGrpServerNames = new ArrayList<>(newRegionsOwners.size());
1347    for (ServerName serverName : masterServices.getServerManager().getOnlineServers().keySet()) {
1348      // In case region move failed in previous attempt, regionsOwners and newRegionsOwners
1349      // can have the same servers. So for all servers below both conditions to be checked
1350      if (newRegionsOwners.contains(serverName.getAddress())) {
1351        srcGrpServerNames.add(serverName);
1352      }
1353      if (regionsOwners.contains(serverName.getAddress())) {
1354        movedServerNames.add(serverName);
1355      }
1356    }
1357    List<Pair<RegionInfo, Future<byte[]>>> assignmentFutures = new ArrayList<>();
1358    int retry = 0;
1359    Set<String> failedRegions = new HashSet<>();
1360    IOException toThrow = null;
1361    do {
1362      assignmentFutures.clear();
1363      failedRegions.clear();
1364      for (ServerName owner : movedServerNames) {
1365        // Get regions that are associated with this server and filter regions by group tables.
1366        for (RegionInfo region : getRegionsInfo.apply((T) owner.getAddress())) {
1367          if (!validation.apply(region)) {
1368            LOG.info("Moving region {}, which does not belong to RSGroup {}",
1369              region.getShortNameToLog(), targetGroupName);
1370            // Move region back to source RSGroup servers
1371            ServerName dest =
1372              masterServices.getLoadBalancer().randomAssignment(region, srcGrpServerNames);
1373            if (dest == null) {
1374              failedRegions.add(region.getRegionNameAsString());
1375              continue;
1376            }
1377            RegionPlan rp = new RegionPlan(region, owner, dest);
1378            try {
1379              Future<byte[]> future = masterServices.getAssignmentManager().moveAsync(rp);
1380              assignmentFutures.add(Pair.newPair(region, future));
1381            } catch (IOException ioe) {
1382              failedRegions.add(region.getRegionNameAsString());
1383              LOG.debug("Move region {} failed, will retry, current retry time is {}",
1384                region.getShortNameToLog(), retry, ioe);
1385              toThrow = ioe;
1386            }
1387          }
1388        }
1389      }
1390      waitForRegionMovement(assignmentFutures, failedRegions, sourceGroupName, retry);
1391      if (failedRegions.isEmpty()) {
1392        LOG.info("All regions from {} are moved back to {}", movedServerNames, sourceGroupName);
1393        return;
1394      } else {
1395        try {
1396          wait(1000);
1397        } catch (InterruptedException e) {
1398          LOG.warn("Sleep interrupted", e);
1399          Thread.currentThread().interrupt();
1400        }
1401        retry++;
1402      }
1403    } while (
1404      !failedRegions.isEmpty() && retry <= masterServices.getConfiguration()
1405        .getInt(FAILED_MOVE_MAX_RETRY, DEFAULT_MAX_RETRY_VALUE)
1406    );
1407
1408    // has up to max retry time or there are no more regions to move
1409    if (!failedRegions.isEmpty()) {
1410      // print failed moved regions, for later process conveniently
1411      String msg = String.format("move regions for group %s failed, failed regions: %s",
1412        sourceGroupName, failedRegions);
1413      LOG.error(msg);
1414      throw new DoNotRetryIOException(
1415        msg + ", just record the last failed region's cause, more details in server log", toThrow);
1416    }
1417  }
1418
1419  /**
1420   * Wait for all the region move to complete. Keep waiting for other region movement completion
1421   * even if some region movement fails.
1422   */
1423  private void waitForRegionMovement(List<Pair<RegionInfo, Future<byte[]>>> regionMoveFutures,
1424    Set<String> failedRegions, String sourceGroupName, int retryCount) {
1425    LOG.info("Moving {} region(s) to group {}, current retry={}", regionMoveFutures.size(),
1426      sourceGroupName, retryCount);
1427    for (Pair<RegionInfo, Future<byte[]>> pair : regionMoveFutures) {
1428      try {
1429        pair.getSecond().get();
1430        if (
1431          masterServices.getAssignmentManager().getRegionStates().getRegionState(pair.getFirst())
1432            .isFailedOpen()
1433        ) {
1434          failedRegions.add(pair.getFirst().getRegionNameAsString());
1435        }
1436      } catch (InterruptedException e) {
1437        // Dont return form there lets wait for other regions to complete movement.
1438        failedRegions.add(pair.getFirst().getRegionNameAsString());
1439        LOG.warn("Sleep interrupted", e);
1440      } catch (Exception e) {
1441        failedRegions.add(pair.getFirst().getRegionNameAsString());
1442        LOG.error("Move region {} to group {} failed, will retry on next attempt",
1443          pair.getFirst().getShortNameToLog(), sourceGroupName, e);
1444      }
1445    }
1446  }
1447
1448  private boolean isTableInGroup(TableName tableName, String groupName,
1449    Set<TableName> tablesInGroupCache) throws IOException {
1450    if (tablesInGroupCache.contains(tableName)) {
1451      return true;
1452    }
1453    if (
1454      RSGroupUtil.getRSGroupInfo(masterServices, this, tableName).map(RSGroupInfo::getName)
1455        .orElse(RSGroupInfo.DEFAULT_GROUP).equals(groupName)
1456    ) {
1457      tablesInGroupCache.add(tableName);
1458      return true;
1459    }
1460    return false;
1461  }
1462
1463  private Map<String, RegionState> rsGroupGetRegionsInTransition(String groupName)
1464    throws IOException {
1465    Map<String, RegionState> rit = Maps.newTreeMap();
1466    Set<TableName> tablesInGroupCache = new HashSet<>();
1467    for (RegionStateNode regionNode : masterServices.getAssignmentManager()
1468      .getRegionsInTransition()) {
1469      TableName tn = regionNode.getTable();
1470      if (isTableInGroup(tn, groupName, tablesInGroupCache)) {
1471        rit.put(regionNode.getRegionInfo().getEncodedName(), regionNode.toRegionState());
1472      }
1473    }
1474    return rit;
1475  }
1476
1477  /**
1478   * This is an EXPENSIVE clone. Cloning though is the safest thing to do. Can't let out original
1479   * since it can change and at least the load balancer wants to iterate this exported list. Load
1480   * balancer should iterate over this list because cloned list will ignore disabled table and split
1481   * parent region cases. This method is invoked by {@link #balanceRSGroup}
1482   * @return A clone of current assignments for this group.
1483   */
1484  Map<TableName, Map<ServerName, List<RegionInfo>>> getRSGroupAssignmentsByTable(
1485    TableStateManager tableStateManager, String groupName) throws IOException {
1486    Map<TableName, Map<ServerName, List<RegionInfo>>> result = Maps.newHashMap();
1487    Set<TableName> tablesInGroupCache = new HashSet<>();
1488    for (Map.Entry<RegionInfo, ServerName> entry : masterServices.getAssignmentManager()
1489      .getRegionStates().getRegionAssignments().entrySet()) {
1490      RegionInfo region = entry.getKey();
1491      TableName tn = region.getTable();
1492      ServerName server = entry.getValue();
1493      if (isTableInGroup(tn, groupName, tablesInGroupCache)) {
1494        if (
1495          tableStateManager.isTableState(tn, TableState.State.DISABLED, TableState.State.DISABLING)
1496        ) {
1497          continue;
1498        }
1499        if (region.isSplitParent()) {
1500          continue;
1501        }
1502        result.computeIfAbsent(tn, k -> new HashMap<>())
1503          .computeIfAbsent(server, k -> new ArrayList<>()).add(region);
1504      }
1505    }
1506    RSGroupInfo rsGroupInfo = getRSGroupInfo(groupName);
1507    for (ServerName serverName : masterServices.getServerManager().getOnlineServers().keySet()) {
1508      if (rsGroupInfo.containsServer(serverName.getAddress())) {
1509        for (Map<ServerName, List<RegionInfo>> map : result.values()) {
1510          map.computeIfAbsent(serverName, k -> Collections.emptyList());
1511        }
1512      }
1513    }
1514    return result;
1515  }
1516
1517  @Override
1518  public BalanceResponse balanceRSGroup(String groupName, BalanceRequest request)
1519    throws IOException {
1520    ServerManager serverManager = masterServices.getServerManager();
1521    LoadBalancer balancer = masterServices.getLoadBalancer();
1522    getRSGroupInfo(groupName);
1523
1524    BalanceResponse.Builder responseBuilder = BalanceResponse.newBuilder();
1525
1526    synchronized (balancer) {
1527      // If balance not true, don't run balancer.
1528      if (!masterServices.isBalancerOn() && !request.isDryRun()) {
1529        return responseBuilder.build();
1530      }
1531
1532      // Only allow one balance run at at time.
1533      Map<String, RegionState> groupRIT = rsGroupGetRegionsInTransition(groupName);
1534      if (groupRIT.size() > 0 && !request.isIgnoreRegionsInTransition()) {
1535        LOG.debug("Not running balancer because {} region(s) in transition: {}", groupRIT.size(),
1536          StringUtils.abbreviate(
1537            masterServices.getAssignmentManager().getRegionsInTransition().toString(), 256));
1538        return responseBuilder.build();
1539      }
1540
1541      if (serverManager.areDeadServersInProgress()) {
1542        LOG.debug("Not running balancer because processing dead regionserver(s): {}",
1543          serverManager.getDeadServers());
1544        return responseBuilder.build();
1545      }
1546
1547      // We balance per group instead of per table
1548      Map<TableName, Map<ServerName, List<RegionInfo>>> assignmentsByTable =
1549        getRSGroupAssignmentsByTable(masterServices.getTableStateManager(), groupName);
1550      List<RegionPlan> plans = balancer.balanceCluster(assignmentsByTable);
1551      boolean balancerRan = !plans.isEmpty();
1552
1553      responseBuilder.setBalancerRan(balancerRan).setMovesCalculated(plans.size());
1554
1555      if (balancerRan && !request.isDryRun()) {
1556        LOG.info("RSGroup balance {} starting with plan count: {}", groupName, plans.size());
1557        List<RegionPlan> executed = masterServices.executeRegionPlansWithThrottling(plans);
1558        responseBuilder.setMovesExecuted(executed.size());
1559        LOG.info("RSGroup balance " + groupName + " completed");
1560      }
1561
1562      return responseBuilder.build();
1563    }
1564  }
1565
1566  private void moveTablesAndWait(Set<TableName> tables, String targetGroup) throws IOException {
1567    LOG.debug("Moving {} tables to target group {}", tables.size(), targetGroup);
1568    List<Long> procIds = new ArrayList<Long>();
1569    for (TableName tableName : tables) {
1570      TableDescriptor oldTd = masterServices.getTableDescriptors().get(tableName);
1571      if (oldTd == null) {
1572        continue;
1573      }
1574      TableDescriptor newTd =
1575        TableDescriptorBuilder.newBuilder(oldTd).setRegionServerGroup(targetGroup).build();
1576      procIds.add(
1577        masterServices.modifyTable(tableName, newTd, HConstants.NO_NONCE, HConstants.NO_NONCE));
1578    }
1579    for (long procId : procIds) {
1580      Procedure<?> proc = masterServices.getMasterProcedureExecutor().getProcedure(procId);
1581      if (proc == null) {
1582        continue;
1583      }
1584      ProcedureSyncWait.waitForProcedureToCompleteIOE(masterServices.getMasterProcedureExecutor(),
1585        proc, Long.MAX_VALUE);
1586    }
1587    LOG.info("Move tables done: moved {} tables to {}", tables.size(), targetGroup);
1588    if (LOG.isDebugEnabled()) {
1589      LOG.debug("Tables moved to {}: {}", targetGroup, tables);
1590    }
1591  }
1592
1593  @Override
1594  public void setRSGroup(Set<TableName> tables, String groupName) throws IOException {
1595    getRSGroupInfo(groupName);
1596    moveTablesAndWait(tables, groupName);
1597  }
1598
1599  public void moveServers(Set<Address> servers, String targetGroupName) throws IOException {
1600    if (servers == null) {
1601      throw new ConstraintException("The list of servers to move cannot be null.");
1602    }
1603    if (servers.isEmpty()) {
1604      // For some reason this difference between null servers and isEmpty is important distinction.
1605      // TODO. Why? Stuff breaks if I equate them.
1606      return;
1607    }
1608    if (StringUtils.isEmpty(targetGroupName)) {
1609      throw new ConstraintException("RSGroup cannot be null.");
1610    }
1611
1612    // Hold a lock on the manager instance while moving servers to prevent
1613    // another writer changing our state while we are working.
1614    synchronized (this) {
1615      // Presume first server's source group. Later ensure all servers are from this group.
1616      Address firstServer = servers.iterator().next();
1617      RSGroupInfo srcGrp = getRSGroupOfServer(firstServer);
1618      if (srcGrp == null) {
1619        // Be careful. This exception message is tested for in TestRSGroupAdmin2...
1620        throw new ConstraintException(
1621          "Server " + firstServer + " is either offline or it does not exist.");
1622      }
1623
1624      // Only move online servers (when moving from 'default') or servers from other
1625      // groups. This prevents bogus servers from entering groups
1626      if (RSGroupInfo.DEFAULT_GROUP.equals(srcGrp.getName())) {
1627        if (srcGrp.getServers().size() <= servers.size()) {
1628          throw new ConstraintException(KEEP_ONE_SERVER_IN_DEFAULT_ERROR_MESSAGE);
1629        }
1630        checkOnlineServersOnly(servers);
1631      }
1632      // Ensure all servers are of same rsgroup.
1633      for (Address server : servers) {
1634        String tmpGroup = getRSGroupOfServer(server).getName();
1635        if (!tmpGroup.equals(srcGrp.getName())) {
1636          throw new ConstraintException("Move server request should only come from one source "
1637            + "RSGroup. Expecting only " + srcGrp.getName() + " but contains " + tmpGroup);
1638        }
1639      }
1640      if (srcGrp.getServers().size() <= servers.size()) {
1641        // check if there are still tables reference this group
1642        for (TableDescriptor td : masterServices.getTableDescriptors().getAll().values()) {
1643          Optional<String> optGroupName = td.getRegionServerGroup();
1644          if (optGroupName.isPresent() && optGroupName.get().equals(srcGrp.getName())) {
1645            throw new ConstraintException(
1646              "Cannot leave a RSGroup " + srcGrp.getName() + " that contains tables('"
1647                + td.getTableName() + "' at least) without servers to host them.");
1648          }
1649        }
1650      }
1651
1652      // MovedServers may be < passed in 'servers'.
1653      Set<Address> movedServers = moveServers(servers, srcGrp.getName(), targetGroupName);
1654      // moveServers (above) no longer mutates srcGrp in place -- re-fetch to see the post-move
1655      // remaining server set.
1656      RSGroupInfo refreshedSrcGrp = getRSGroupInfo(srcGrp.getName());
1657      moveServerRegionsFromGroup(movedServers, refreshedSrcGrp.getServers(), targetGroupName,
1658        srcGrp.getName());
1659      LOG.info("Move servers done: moved {} servers from {} to {}", movedServers.size(),
1660        srcGrp.getName(), targetGroupName);
1661      if (LOG.isDebugEnabled()) {
1662        LOG.debug("Servers moved from {} to {}: {}", srcGrp.getName(), targetGroupName,
1663          movedServers);
1664      }
1665    }
1666  }
1667
1668  @Override
1669  public String determineRSGroupInfoForTable(TableName tableName) {
1670    return script.getRSGroup(tableName.getNamespaceAsString(), tableName.getQualifierAsString());
1671  }
1672
1673  @Override
1674  public synchronized void renameRSGroup(String oldName, String newName) throws IOException {
1675    if (oldName.equals(RSGroupInfo.DEFAULT_GROUP)) {
1676      throw new ConstraintException(RSGroupInfo.DEFAULT_GROUP + " can't be rename");
1677    }
1678    checkGroupName(newName);
1679    // getRSGroupInfo validates old RSGroup existence.
1680    RSGroupInfo oldRSG = getRSGroupInfo(oldName);
1681    Map<String, RSGroupInfo> rsGroupMap = holder.groupName2Group;
1682    if (rsGroupMap.containsKey(newName)) {
1683      throw new ConstraintException("Group already exists: " + newName);
1684    }
1685
1686    Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
1687    newGroupMap.remove(oldRSG.getName());
1688    RSGroupInfo newRSG = new RSGroupInfo(newName, oldRSG.getServers());
1689    newGroupMap.put(newName, newRSG);
1690    flushConfig(newGroupMap);
1691    Set<TableName> updateTables = masterServices.getTableDescriptors().getAll().values().stream()
1692      .filter(t -> oldName.equals(t.getRegionServerGroup().orElse(null)))
1693      .map(TableDescriptor::getTableName).collect(Collectors.toSet());
1694    setRSGroup(updateTables, newName);
1695    LOG.info("Rename RSGroup done: {} => {}", oldName, newName);
1696  }
1697
1698  @Override
1699  public synchronized void updateRSGroupConfig(String groupName, Map<String, String> configuration)
1700    throws IOException {
1701    if (RSGroupInfo.DEFAULT_GROUP.equals(groupName)) {
1702      // We do not persist anything of default group, therefore, it is not supported to update
1703      // default group's configuration which lost once master down.
1704      throw new ConstraintException(
1705        "configuration of " + RSGroupInfo.DEFAULT_GROUP + " can't be stored persistently");
1706    }
1707    RSGroupInfo rsGroupInfo = copyOf(getRSGroupInfo(groupName));
1708    // Iterate over a snapshot of the keys, as removeConfiguration modifies the backing map.
1709    new ArrayList<>(rsGroupInfo.getConfiguration().keySet())
1710      .forEach(rsGroupInfo::removeConfiguration);
1711    configuration.forEach((k, v) -> rsGroupInfo.setConfiguration(k, v));
1712    Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(holder.groupName2Group);
1713    newGroupMap.put(groupName, rsGroupInfo);
1714    flushConfig(newGroupMap);
1715  }
1716}