/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase.rsgroup;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.IOException;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.LocalHBaseCluster;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.SingleProcessHBaseCluster;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.net.Address;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.RSGroupTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.DNS;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.collect.Sets;

import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.RSGroupProtos;

/**
 * Black-box tests for regex-based automatic RSGroup membership
 * (hbase.rsgroup.regex.<groupname>=<regex>), driven entirely through the public
 * {@link org.apache.hadoop.hbase.client.Admin} RSGroup API, live RegionServer add/remove against
 * the minicluster, and captured log output -- no reflection or visibility changes against
 * {@link RSGroupInfoManagerImpl}.
 */
@Tag(RSGroupTests.TAG)
@Tag(MediumTests.TAG)
public class TestRegexBasedRSGroupMembership extends TestRSGroupsBase {

  private static final Logger LOG = LoggerFactory.getLogger(TestRegexBasedRSGroupMembership.class);
  private static final String REGEX_PREFIX = "hbase.rsgroup.regex.";

  private final List<JVMClusterUtil.RegionServerThread> fakeRegionServers = new ArrayList<>();
  private final Set<String> regexConfigKeysSet = new HashSet<>();

  @BeforeAll
  public static void setUp() throws Exception {
    setUpTestBeforeClass();
  }

  @AfterAll
  public static void tearDown() throws Exception {
    tearDownAfterClass();
  }

  @BeforeEach
  public void beforeMethod(TestInfo testInfo) throws Exception {
    setUpBeforeMethod(testInfo);
  }

  @AfterEach
  public void afterMethod() throws Exception {
    // Stop any fake-hostname RS this test started before handing back to the shared teardown,
    // which only ever starts servers to top back up to NUM_SLAVES_BASE, never stops extras.
    for (JVMClusterUtil.RegionServerThread rst : new ArrayList<>(fakeRegionServers)) {
      stopFakeRegionServer(rst);
    }
    // The MASTER/ADMIN Configuration objects are shared across every test method in this class;
    // undo any regex config this test added to both so it can't leak into the next method.
    for (String key : regexConfigKeysSet) {
      MASTER.getConfiguration().unset(key);
      ADMIN.getConfiguration().unset(key);
    }
    regexConfigKeysSet.clear();
    tearDownAfterMethod();
    RSGroupBasedLoadBalancer balancer = (RSGroupBasedLoadBalancer) MASTER.getLoadBalancer();
    balancer.resetAssignmentCallFlagsForTest();
    balancer.setFallbackEnabledForTest(false);
  }

  // ============================== helpers ==============================

  /**
   * Starts a genuinely new RegionServer process (thread) in the shared minicluster that reports
   * itself under a caller-chosen hostname, so regex membership rules can target it individually.
   * The base cluster's real RS all share one real, resolvable hostname (this machine's address);
   * the RPC services constructor unconditionally resolves whatever hostname is configured
   * (independent of any RPC bind-address override), so the caller-chosen hostname here must itself
   * be a real, resolvable, bindable literal -- "127.0.0.1" and "localhost" both work and are
   * guaranteed distinct from the base cluster's real hostname, which is exactly what every test in
   * this class relies on to get a controllable, non-shared identity.
   */
  private JVMClusterUtil.RegionServerThread startFakeHostnameRS(String hostname) throws Exception {
    Configuration rsConf = new Configuration(TEST_UTIL.getConfiguration());
    rsConf.set(DNS.UNSAFE_RS_HOSTNAME_KEY, hostname);
    rsConf.set("hbase.regionserver.ipc.address", "localhost");
    // The shared TEST_UTIL configuration's regionserver port is never rewritten to an ephemeral
    // value by minicluster startup (the base RS get their distinct ports through a separate,
    // per-instance mechanism), so without this override every fake RS started here would try to
    // bind the literal default port (16020) and collide with any other fake RS still alive.
    rsConf.set(HConstants.REGIONSERVER_PORT, "0");
    // Same issue for the embedded web UI (default port 16030); these fake RS don't need one.
    rsConf.set(HConstants.REGIONSERVER_INFO_PORT, "-1");
    LocalHBaseCluster localCluster = ((SingleProcessHBaseCluster) CLUSTER).hbaseCluster;
    JVMClusterUtil.RegionServerThread rst =
      localCluster.addRegionServer(rsConf, localCluster.getRegionServers().size());
    rst.start();
    rst.waitForServerOnline();
    fakeRegionServers.add(rst);
    final ServerName sn = rst.getRegionServer().getServerName();
    TEST_UTIL.waitFor(WAIT_TIMEOUT,
      () -> MASTER.getServerManager().getOnlineServersList().contains(sn));
    LOG.info("Started fake-hostname RS {}", sn);
    return rst;
  }

  private void stopFakeRegionServer(JVMClusterUtil.RegionServerThread rst) throws Exception {
    fakeRegionServers.remove(rst);
    final ServerName sn = rst.getRegionServer().getServerName();
    rst.getRegionServer().stop("test cleanup");
    TEST_UTIL.waitFor(WAIT_TIMEOUT,
      () -> !MASTER.getServerManager().getOnlineServersList().contains(sn));
    LOG.info("Stopped fake-hostname RS {}", sn);
  }

  private void killFakeRegionServer(JVMClusterUtil.RegionServerThread rst) throws Exception {
    fakeRegionServers.remove(rst);
    final ServerName sn = rst.getRegionServer().getServerName();
    ((SingleProcessHBaseCluster) CLUSTER).killRegionServer(sn);
    TEST_UTIL.waitFor(WAIT_TIMEOUT,
      () -> !MASTER.getServerManager().getOnlineServersList().contains(sn));
    LOG.info("Killed fake-hostname RS {}", sn);
  }

  /**
   * Server-membership recompute happens synchronously in the {@code serverAdded}/
   * {@code serverRemoved} callback -- it is only ever triggered by an actual server add/remove
   * event, a bare config mutation is not enough. Forces such an event (add+remove a bystander RS)
   * so a config change just made takes effect without requiring an explicit RSGroup admin RPC. Only
   * used by tests that *want* the automatic reconciliation to run and correct things; the trigger
   * RS's own transient membership doesn't affect assertions elsewhere since those only check for
   * specific other servers' addresses.
   */
  private void triggerListenerCycle() throws Exception {
    JVMClusterUtil.RegionServerThread trigger = startFakeHostnameRS("127.0.0.1");
    stopFakeRegionServer(trigger);
  }

  /** Reads the server set of a group straight from its row in the hbase:rsgroup table. */
  private Set<Address> readStoredServers(String groupName) throws IOException {
    try (
      Table table = TEST_UTIL.getConnection().getTable(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME)) {
      Result result = table.get(new Get(Bytes.toBytes(groupName)));
      byte[] value = result.getValue(RSGroupInfoManagerImpl.META_FAMILY_BYTES,
        RSGroupInfoManagerImpl.META_QUALIFIER_BYTES);
      if (value == null) {
        return new HashSet<>();
      }
      return new HashSet<>(
        ProtobufUtil.toGroupInfo(RSGroupProtos.RSGroupInfo.parseFrom(value)).getServers());
    }
  }

  /**
   * Fails over to a freshly started master (the minicluster's masters are threads): a backup is
   * started, the active master is stopped, and the shared MASTER/ADMIN handles are re-pointed. The
   * base class keeps the master's embedded RS in a "master" group, whose stored address would go
   * stale, so the master is first moved back to 'default' (whose membership is live-computed); the
   * next cleanup places the new master in the "master" group again.
   */
  private void restartMaster() throws Exception {
    SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
    int oldIndex = -1;
    List<JVMClusterUtil.MasterThread> masters = cluster.getMasterThreads();
    for (int i = 0; i < masters.size(); i++) {
      if (masters.get(i).getMaster() == MASTER) {
        oldIndex = i;
      }
    }
    cluster.startMaster();
    cluster.stopMaster(oldIndex);
    cluster.waitOnMaster(oldIndex);
    assertTrue(cluster.waitForActiveAndReadyMaster(WAIT_TIMEOUT));
    // Clients bootstrap from the configured master address, which is now the dead master's.
    TEST_UTIL.getConfiguration().set(HConstants.MASTER_ADDRS_KEY,
      cluster.getMaster().getServerName().getAddress().toString());
    TEST_UTIL.closeConnection();
    ADMIN.close();
    initialize();
  }

  private void setRegex(String groupName, String regex) {
    String key = REGEX_PREFIX + groupName;
    MASTER.getConfiguration().set(key, regex);
    // ADMIN's own Configuration is a distinct object from MASTER's (the minicluster deep-copies
    // configuration per daemon), so mirror the change there too -- VerifyingRSGroupAdmin#verify
    // reads it to know which servers' membership is regex-driven.
    ADMIN.getConfiguration().set(key, regex);
    regexConfigKeysSet.add(key);
  }

  /**
   * Tears down a regex-governed group. Explicit moves that contradict an active regex are rejected,
   * so the regex is first dropped on the master only -- ADMIN's copy still tells
   * VerifyingRSGroupAdmin which servers' membership is regex-driven -- and then the group is
   * emptied and removed.
   */
  private void removeRegexGroup(String groupName) throws Exception {
    String key = REGEX_PREFIX + groupName;
    MASTER.getConfiguration().unset(key);
    try {
      removeGroup(groupName);
    } finally {
      clearRegex(groupName);
    }
  }

  private void clearRegex(String groupName) {
    String key = REGEX_PREFIX + groupName;
    MASTER.getConfiguration().unset(key);
    ADMIN.getConfiguration().unset(key);
    regexConfigKeysSet.remove(key);
  }

  private Address addressOf(JVMClusterUtil.RegionServerThread rst) {
    return rst.getRegionServer().getServerName().getAddress();
  }

  private static class LogCapturer {
    private final StringWriter sw = new StringWriter();
    private final org.apache.logging.log4j.core.appender.WriterAppender appender;
    private final org.apache.logging.log4j.core.Logger logger;

    LogCapturer(org.apache.logging.log4j.core.Logger logger) {
      this.logger = logger;
      this.appender = org.apache.logging.log4j.core.appender.WriterAppender.newBuilder()
        .setName("test-regex-membership").setTarget(sw).build();
      this.appender.start();
      this.logger.addAppender(this.appender);
    }

    String getOutput() {
      return sw.toString();
    }

    void stopCapturing() {
      this.logger.removeAppender(this.appender);
      this.appender.stop();
    }
  }

  private static LogCapturer captureRSGroupInfoManagerLog() {
    return new LogCapturer(
      (org.apache.logging.log4j.core.Logger) org.apache.logging.log4j.LogManager
        .getLogger(RSGroupInfoManagerImpl.class));
  }

  // ============================== tests ==============================

  @Test
  public void testNoRegexConfigBaseline() throws Exception {
    // No hbase.rsgroup.regex.* configured at all: every server must remain in default, exactly
    // as it would without this feature. This is also implicitly exercised by every other test in
    // this module, but pin it down explicitly for this feature.
    RSGroupInfo defaultGroup = ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP);
    assertEquals(NUM_SLAVES_BASE, defaultGroup.getServers().size());
  }

  @Test
  public void testSingleRegexAutoJoinOnLiveAdd() throws Exception {
    String groupName = getGroupName("auto");
    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\.1");

    JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
    Address addr = addressOf(rst);
    try {
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr));
      // No explicit moveServersToRSGroup call was made -- the server auto-joined purely from the
      // ServerListener reacting to its own arrival.
      assertFalse(ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testRegexWithQuantifiersMatches() throws Exception {
    String groupName = getGroupName("quant");
    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\.\\d{1,3}");

    JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
    Address addr = addressOf(rst);
    try {
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr));
      // No explicit moveServersToRSGroup call was made -- the server auto-joined purely from the
      // ServerListener reacting to its own arrival.
      assertFalse(ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testFullStringMatchSemantics() throws Exception {
    // The regex is a strict prefix of the hostname; full-string Pattern#matches semantics must
    // NOT treat this as a match.
    String groupName = getGroupName("prefix");
    ADMIN.addRSGroup(groupName);
    // Strict prefix of "127.0.0.1" -- must NOT match under full-string Pattern#matches semantics.
    setRegex(groupName, "127\\.0\\.0");

    JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
    Address addr = addressOf(rst);
    try {
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      assertFalse(ADMIN.getRSGroup(groupName).getServers().contains(addr));
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testRegexTargetingDefaultGroupIsNoOpAndWarns() throws Exception {
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      setRegex(RSGroupInfo.DEFAULT_GROUP, "127\\.0\\.0\\..*");
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> capturer.getOutput()
        .contains("cannot target the reserved '" + RSGroupInfo.DEFAULT_GROUP + "' group"));
    } finally {
      capturer.stopCapturing();
      clearRegex(RSGroupInfo.DEFAULT_GROUP);
    }
  }

  @Test
  public void testInvalidRegexSyntaxIgnoredAndWarns() throws Exception {
    String groupName = getGroupName("badregex");
    ADMIN.addRSGroup(groupName);
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      setRegex(groupName, "[");
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> capturer.getOutput().contains("Invalid regex '[' for RSGroup '" + groupName + "'"));
    } finally {
      capturer.stopCapturing();
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testInvalidGroupNameKeyIgnoredAndWarns() throws Exception {
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      setRegex("not a valid name", "127\\.0\\.0\\..*");
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> capturer.getOutput().contains("is not a valid RSGroup name"));
    } finally {
      capturer.stopCapturing();
      clearRegex("not a valid name");
    }
  }

  @Test
  public void testListenerDoesNotStickToStaleServer() throws Exception {
    // A server leaves, then a *different* server with a matching hostname joins later; the
    // recompute must not get stuck comparing against a stale previous-assignments snapshot.
    String groupName = getGroupName("stale");
    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\..*");
    try {
      JVMClusterUtil.RegionServerThread first = startFakeHostnameRS("127.0.0.1");
      Address firstAddr = addressOf(first);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(firstAddr));

      stopFakeRegionServer(first);

      // Different startcode (fresh ServerName) under the same matching hostname.
      JVMClusterUtil.RegionServerThread second = startFakeHostnameRS("127.0.0.1");
      Address secondAddr = addressOf(second);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(secondAddr));
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testDanglingGroupFallsBackThenResolvesOnceGroupCreated() throws Exception {
    String groupName = getGroupName("dangling");
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      // Group does not exist yet when the regex is configured and the matching RS joins.
      setRegex(groupName, "127\\.0\\.0\\..*");
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> capturer.getOutput()
        .contains("RSGroup '" + groupName + "' does not exist -- create it with addRSGroup"));

      // Now create the group; on the next cycle the server should migrate in.
      ADMIN.addRSGroup(groupName);
      triggerListenerCycle();
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr));
    } finally {
      capturer.stopCapturing();
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testAmbiguousMatchFallsBackThenResolvesOnceOverlapFixed() throws Exception {
    String groupOne = getGroupName("ambig1");
    String groupTwo = getGroupName("ambig2");
    ADMIN.addRSGroup(groupOne);
    ADMIN.addRSGroup(groupTwo);
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      setRegex(groupOne, "127\\.0\\.0\\..*");
      setRegex(groupTwo, "127\\.0\\.0\\.1");

      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> capturer.getOutput().contains("matches regexes for multiple RSGroups"));
      assertFalse(ADMIN.getRSGroup(groupOne).getServers().contains(addr));
      assertFalse(ADMIN.getRSGroup(groupTwo).getServers().contains(addr));

      // Fix the overlap
      setRegex(groupOne, "10\\.0\\.0\\..*");
      triggerListenerCycle();
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> ADMIN.getRSGroup(groupTwo).getServers().contains(addr));
    } finally {
      capturer.stopCapturing();
      removeRegexGroup(groupOne);
      removeRegexGroup(groupTwo);
    }
  }

  @Test
  public void testRegexCoveringEveryServerEmptiesDefaultAndWarns() throws Exception {
    // Regex placement is always applied: a catch-all regex matches every online server (all base
    // RS share one real hostname), so they all move into the regex group and default is left
    // empty. That is a warning about an unusual configuration, not an error, and fallback is what
    // lets default's tables keep being assigned in that state.
    // (Explicit moveServersToRSGroup can never empty default -- an unrelated, longstanding
    // RSGroupInfoManagerImpl#moveServers guard keeps >=1 server there -- so the empty-default
    // state is driven purely through the automatic regex-reconciliation path.)
    String regexGroup = getGroupName("emptydefault");
    ADMIN.addRSGroup(regexGroup);
    RSGroupBasedLoadBalancer balancer = (RSGroupBasedLoadBalancer) MASTER.getLoadBalancer();
    balancer.setFallbackEnabledForTest(true);
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      setRegex(regexGroup, ".*");
      triggerListenerCycle();

      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> capturer.getOutput().contains("leaves RSGroup 'default' with no online servers"));
      assertTrue(capturer.getOutput().contains(RSGroupBasedLoadBalancer.FALLBACK_GROUP_ENABLE_KEY));
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(regexGroup).getServers().size() == NUM_SLAVES_BASE);
      assertTrue(ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().isEmpty());

      // With default empty and fallback on, a table in default is still assigned -- onto servers
      // of other groups.
      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 5);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
      assertEquals(RSGroupInfo.DEFAULT_GROUP, ADMIN.getRSGroup(tableName).getName());

      // Narrow the regex so it only targets a new server, not the base cluster's shared
      // hostname -- the base servers are unmatched again and return to default.
      setRegex(regexGroup, "127\\.0\\.0\\..*");
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> ADMIN.getRSGroup(regexGroup).getServers().contains(addr)
        && ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().size() == NUM_SLAVES_BASE);
    } finally {
      capturer.stopCapturing();
      removeRegexGroup(regexGroup);
    }
  }

  @Test
  public void testCatchAllRegexMovesEveryServerAndGroupTablesStayAssigned() throws Exception {
    // A regex group with a live member and a bound table is broadened to a catch-all. Every
    // server is placed in its regex group (default is left empty), and the group's table stays
    // fully assigned on its original member.
    String groupName = getGroupName("catchall");
    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\.1");
    RSGroupBasedLoadBalancer balancer = (RSGroupBasedLoadBalancer) MASTER.getLoadBalancer();
    balancer.setFallbackEnabledForTest(true);
    try {
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr));
      ServerName sn = getServerName(addr);

      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 5);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
      ADMIN.setRSGroup(Sets.newHashSet(tableName), groupName);
      assertEquals(5, getTableServerRegionMap().get(tableName).get(sn).size());

      setRegex(groupName, ".*");
      triggerListenerCycle();
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> {
        Set<Address> servers = ADMIN.getRSGroup(groupName).getServers();
        return servers.contains(addr) && servers.size() == NUM_SLAVES_BASE + 1
          && ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().isEmpty();
      });
      assertEquals(5, getTableServerRegionMap().get(tableName).get(sn).size());
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testRegexGovernedGroupMembershipIsNotPersisted() throws Exception {
    // Regex-governed membership is computed on the fly: it is visible even while hbase:rsgroup is
    // unavailable, and neither the table row nor any later flush ever carries it.
    String groupName = getGroupName("nopersist");
    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\.1");
    boolean tableDisabled = false;
    try {
      TEST_UTIL.getAdmin().disableTable(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME);
      tableDisabled = true;

      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr));

      TEST_UTIL.getAdmin().enableTable(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME);
      tableDisabled = false;
      // Any group mutation flushes every group.
      String flushGroup = getGroupName("nopersistflush");
      ADMIN.addRSGroup(flushGroup);
      removeGroup(flushGroup);

      assertTrue(ADMIN.getRSGroup(groupName).getServers().contains(addr));
      assertTrue(readStoredServers(groupName).isEmpty());
    } finally {
      if (tableDisabled) {
        TEST_UTIL.getAdmin().enableTable(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME);
      }
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testMoveContradictingRegexIsRejected() throws Exception {
    String groupName = getGroupName("reject");
    String otherGroup = getGroupName("rejectother");
    ADMIN.addRSGroup(groupName);
    ADMIN.addRSGroup(otherGroup);
    setRegex(groupName, "127\\.0\\.0\\..*");
    try {
      // A server whose hostname does not match the regex cannot be moved into the regex group ...
      JVMClusterUtil.RegionServerThread nonMatching = startFakeHostnameRS("localhost");
      Address nonMatchingAddr = addressOf(nonMatching);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(nonMatchingAddr));
      assertThrows(IOException.class,
        () -> ADMIN.moveServersToRSGroup(Sets.newHashSet(nonMatchingAddr), groupName));
      assertTrue(
        ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(nonMatchingAddr));
      assertFalse(ADMIN.getRSGroup(groupName).getServers().contains(nonMatchingAddr));

      // ... but can still be placed in any other group.
      ADMIN.moveServersToRSGroup(Sets.newHashSet(nonMatchingAddr), otherGroup);
      assertTrue(ADMIN.getRSGroup(otherGroup).getServers().contains(nonMatchingAddr));
      ADMIN.moveServersToRSGroup(Sets.newHashSet(nonMatchingAddr), RSGroupInfo.DEFAULT_GROUP);

      // A server the regex claims cannot be moved out of, or to any group other than, its group.
      JVMClusterUtil.RegionServerThread matching = startFakeHostnameRS("127.0.0.1");
      Address matchingAddr = addressOf(matching);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(matchingAddr));
      assertThrows(IOException.class,
        () -> ADMIN.moveServersToRSGroup(Sets.newHashSet(matchingAddr), RSGroupInfo.DEFAULT_GROUP));
      assertThrows(IOException.class,
        () -> ADMIN.moveServersToRSGroup(Sets.newHashSet(matchingAddr), otherGroup));
      assertTrue(ADMIN.getRSGroup(groupName).getServers().contains(matchingAddr));
    } finally {
      removeGroup(otherGroup);
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testUnrelatedAdminOpsSucceedUnderActiveRegex() throws Exception {
    String regexGroup = getGroupName("activeregex");
    String otherGroup = getGroupName("other");
    ADMIN.addRSGroup(regexGroup);
    setRegex(regexGroup, "127\\.0\\.0\\..*");
    try {
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(regexGroup).getServers().contains(addr));

      // Admin operations on admin-managed groups must keep working while a regex-governed group is
      // active. setRSGroup requires the target group to already have at least one server, so use a
      // base (non-regex-matched, admin-manageable) server rather than an empty new group.
      RSGroupInfo otherGroupInfo = addGroup(otherGroup, 1);
      Set<Address> otherGroupServers = otherGroupInfo.getServers();
      TEST_UTIL.createTable(tableName, Bytes.toBytes("f"));
      ADMIN.setRSGroup(Sets.newHashSet(tableName), otherGroup);
      // Table-to-group membership lives on the TableDescriptor (RegionServerGroup), not in
      // RSGroupInfo#getTables() -- the latter is only ever populated by flushConfig/refresh from
      // persisted storage and setRSGroup never touches it. ADMIN.getRSGroup(TableName) is the
      // live, authoritative direction to check this (see RSGroupUtil#getRSGroupInfo).
      assertEquals(otherGroup, ADMIN.getRSGroup(tableName).getName());

      String renamedGroup = otherGroup + "_renamed";
      ADMIN.renameRSGroup(otherGroup, renamedGroup);
      RSGroupInfo renamedGroupInfo = ADMIN.getRSGroup(renamedGroup);
      assertEquals(otherGroupServers, renamedGroupInfo.getServers());
      assertEquals(renamedGroup, ADMIN.getRSGroup(tableName).getName());
      assertFalse(ADMIN.listRSGroups().stream().anyMatch(g -> g.getName().equals(otherGroup)));

      ADMIN.setRSGroup(Sets.newHashSet(tableName), RSGroupInfo.DEFAULT_GROUP);
      assertEquals(RSGroupInfo.DEFAULT_GROUP, ADMIN.getRSGroup(tableName).getName());

      ADMIN.moveServersToRSGroup(ADMIN.getRSGroup(renamedGroup).getServers(),
        RSGroupInfo.DEFAULT_GROUP);
      Set<Address> defaultServersAfterMoveBack =
        ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers();
      for (Address server : otherGroupServers) {
        assertTrue(defaultServersAfterMoveBack.contains(server));
      }
      assertTrue(ADMIN.getRSGroup(renamedGroup).getServers().isEmpty());

      ADMIN.removeRSGroup(renamedGroup);
      assertFalse(ADMIN.listRSGroups().stream().anyMatch(g -> g.getName().equals(renamedGroup)));

      // Confirm the unrelated operations above never disturbed the active regex-governed group.
      assertTrue(ADMIN.getRSGroup(regexGroup).getServers().contains(addr));
    } finally {
      TEST_UTIL.deleteTable(tableName);
      removeRegexGroup(regexGroup);
    }
  }

  @Test
  public void testAutomaticMoveDoesNotFireCoprocessorHooks() throws Exception {
    String groupName = getGroupName("hooks");
    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\..*");
    try {
      assertFalse(OBSERVER.preMoveServersCalled);
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr));
      // Purely automatic, event-driven reassignment must not go through the MasterRpcServices
      // moveServers RPC path, so the moveServers coprocessor hooks must not fire.
      assertFalse(OBSERVER.preMoveServersCalled);
      assertFalse(OBSERVER.postMoveServersCalled);

      // An explicit admin-driven move, by contrast, does fire the hooks. Use one of the
      // base cluster's own (non-regex-matched) RS for this, rather than disturbing the
      // regex-matched fake RS started above, which keeps running under its active regex.
      String otherGroup = getGroupName("hooksother");
      ADMIN.addRSGroup(otherGroup);
      Address baseServerAddr =
        ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().iterator().next();
      ADMIN.moveServersToRSGroup(Sets.newHashSet(baseServerAddr), otherGroup);
      assertTrue(OBSERVER.preMoveServersCalled);
      assertTrue(OBSERVER.postMoveServersCalled);
      removeGroup(otherGroup);

      // The unrelated explicit move above must not have disturbed the regex-governed group.
      assertTrue(ADMIN.getRSGroup(groupName).getServers().contains(addr));
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testRefreshRemovesRegexMatchedServerFromStoredAdminGroup() throws Exception {
    // A server stored in an admin-managed group before a regex claimed it must not end up in two
    // groups after a manager restart. refresh cannot reject (it would wedge startup), so the regex
    // wins in memory and a WARN is logged.
    String adminGroup = getGroupName("refreshadmin");
    String regexGroup = getGroupName("refreshregex");
    ADMIN.addRSGroup(adminGroup);
    ADMIN.addRSGroup(regexGroup);
    LogCapturer capturer = captureRSGroupInfoManagerLog();
    try {
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      ADMIN.moveServersToRSGroup(Sets.newHashSet(addr), adminGroup);

      // The regex arrives while the server is already stored in the admin group. A new master
      // copies the minicluster configuration, so set it there too before the failover.
      setRegex(regexGroup, "127\\.0\\.0\\..*");
      TEST_UTIL.getMiniHBaseCluster().getConfiguration().set(REGEX_PREFIX + regexGroup,
        "127\\.0\\.0\\..*");
      restartMaster();
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(regexGroup).getServers().contains(addr));
      assertFalse(ADMIN.getRSGroup(adminGroup).getServers().contains(addr));
      assertFalse(ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));
      assertTrue(capturer.getOutput().contains("are stored in RSGroup '" + adminGroup + "'"));

      // The startup flush clears the stale stored copy: the admin group no longer holds the server
      // and the regex-governed group stores no servers at all.
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> !readStoredServers(adminGroup).contains(addr));
      assertTrue(readStoredServers(regexGroup).isEmpty());
    } finally {
      capturer.stopCapturing();
      TEST_UTIL.getMiniHBaseCluster().getConfiguration().unset(REGEX_PREFIX + regexGroup);
      removeRegexGroup(regexGroup);
      removeGroup(adminGroup);
    }
  }

  @Test
  public void testRegexConfigDriftDoesNotBlockUnrelatedAdminOps() throws Exception {
    // A regex that contradicts where a server was explicitly placed must not block any unrelated
    // admin operation.
    String groupName = getGroupName("drift");
    String bystanderGroup = getGroupName("bystander");
    ADMIN.addRSGroup(groupName);
    ADMIN.addRSGroup(bystanderGroup);
    String renamedBystander = bystanderGroup + "_renamed";
    try {
      JVMClusterUtil.RegionServerThread rst = startFakeHostnameRS("127.0.0.1");
      Address addr = addressOf(rst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr));

      // Legal at the time: no regex governs groupName yet.
      ADMIN.moveServersToRSGroup(Sets.newHashSet(addr), groupName);

      // Now introduce a regex for groupName that does NOT match the server already placed there.
      // No listener cycle runs, so the contradiction stays un-reconciled.
      setRegex(groupName, "10\\.0\\.0\\..*");

      ADMIN.renameRSGroup(bystanderGroup, renamedBystander);
      assertTrue(ADMIN.listRSGroups().stream().anyMatch(g -> g.getName().equals(renamedBystander)));
    } finally {
      removeRegexGroup(groupName);
      removeGroup(ADMIN.getRSGroup(bystanderGroup) != null ? bystanderGroup : renamedBystander);
    }
  }

  @Test
  public void testNewlyJoinedRegexMemberReceivesRegionsOnBalance() throws Exception {
    // A new RS auto-joins an already-regex-governed group while the group's sole existing
    // member holds every region of a table bound to that group; running the balancer must then
    // hand some of those regions to the newly-joined RS too. The first-member baseline is
    // established by setRSGroup itself: RSGroupInfoManagerImpl#setRSGroup synchronously relocates
    // every region of a table with no RSGroup-scoped descriptor yet onto the target group's
    // servers (see moveTablesAndWait, which blocks on each ModifyTableProcedure before returning),
    // so with sn1 as the sole group member at that point all regions are already there once
    // setRSGroup returns -- no separate balance step is needed for that part. Only the
    // second-member handoff genuinely needs an explicit balance: adding a group member does not
    // by itself retrigger any region movement.
    String groupName = getGroupName("balancenewmember");
    ADMIN.addRSGroup(groupName);
    // "127.0.0.1" and "localhost" are the only two distinct, real, independently-reachable
    // identities startFakeHostnameRS supports (its RPC bind address is always "localhost"
    // regardless of advertised hostname).
    setRegex(groupName, "127\\.0\\.0\\.1|localhost");
    try {
      JVMClusterUtil.RegionServerThread rst1 = startFakeHostnameRS("127.0.0.1");
      Address addr1 = addressOf(rst1);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr1));
      ServerName sn1 = getServerName(addr1);

      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 10);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
      ADMIN.setRSGroup(Sets.newHashSet(tableName), groupName);
      Map<ServerName, List<String>> perServerAfterMove = getTableServerRegionMap().get(tableName);
      assertEquals(10, perServerAfterMove.get(sn1).size());

      ADMIN.balancerSwitch(true, true);
      try {
        // A second RS matching the same regex joins while the group is already active; it must
        // auto-join with no explicit RPC.
        JVMClusterUtil.RegionServerThread rst2 = startFakeHostnameRS("localhost");
        Address addr2 = addressOf(rst2);
        TEST_UTIL.waitFor(WAIT_TIMEOUT,
          () -> ADMIN.getRSGroup(groupName).getServers().contains(addr2));
        ServerName sn2 = getServerName(addr2);

        ADMIN.balanceRSGroup(groupName);
        TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> {
          Map<ServerName, List<String>> perServer = getTableServerRegionMap().get(tableName);
          if (perServer == null) {
            return false;
          }
          List<String> onNewMember = perServer.get(sn2);
          if (onNewMember == null || onNewMember.isEmpty()) {
            return false;
          }
          int total = perServer.values().stream().mapToInt(List::size).sum();
          return total == 10;
        });
      } finally {
        ADMIN.balancerSwitch(false, true);
      }
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testRestartTriggeredMembershipConfinesTableToGroupServers() throws Exception {
    // RS join normally (no regex active yet), landing in default like any other RS. Only
    // afterward is a regex configured targeting their hostnames; a bare config change does not
    // reconcile membership by itself, so an actual lifecycle event -- restarting one of the two
    // RS -- is used to force the recompute to notice and migrate both into the new group
    // (the recompute triggered by one server's add/remove event considers every online server,
    // not just the one that triggered it). A table is then bound to that group, and its regions
    // must land only on the group's members, with those members hosting only that table's
    // regions -- a bidirectional isolation check.
    String groupName = getGroupName("restartisolation");
    // "127.0.0.1" and "localhost" are the only two distinct, real, independently-reachable
    // identities startFakeHostnameRS supports (its RPC bind address is always "localhost"
    // regardless of advertised hostname -- an arbitrary loopback alias like "127.0.0.2" would
    // advertise an address nothing is actually listening on, stalling any real region-open RPC).
    JVMClusterUtil.RegionServerThread rst1 = startFakeHostnameRS("127.0.0.1");
    JVMClusterUtil.RegionServerThread rst2 = startFakeHostnameRS("localhost");
    Address addr2 = addressOf(rst2);
    TEST_UTIL.waitFor(WAIT_TIMEOUT,
      () -> ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers().contains(addr2));

    ADMIN.addRSGroup(groupName);
    setRegex(groupName, "127\\.0\\.0\\.1|localhost");
    try {
      stopFakeRegionServer(rst1);
      JVMClusterUtil.RegionServerThread rst1Restarted = startFakeHostnameRS("127.0.0.1");
      Address addr1Restarted = addressOf(rst1Restarted);

      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr1Restarted));
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr2));

      ServerName sn1 = getServerName(addr1Restarted);
      ServerName sn2 = getServerName(addr2);

      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 6);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
      // setRSGroup synchronously relocates every region of a table with no RSGroup-scoped
      // descriptor yet onto one of the target group's servers (moveTablesAndWait blocks on each
      // ModifyTableProcedure before returning), so confinement to {sn1, sn2} is already guaranteed
      // once this call returns -- no separate balance step is needed.
      ADMIN.setRSGroup(Sets.newHashSet(tableName), groupName);
      Map<ServerName, List<String>> perServerAfterMove = getTableServerRegionMap().get(tableName);
      int totalAfterMove = 0;
      for (Map.Entry<ServerName, List<String>> entry : perServerAfterMove.entrySet()) {
        assertTrue(entry.getKey().equals(sn1) || entry.getKey().equals(sn2));
        totalAfterMove += entry.getValue().size();
      }
      assertEquals(6, totalAfterMove);

      // Bidirectional isolation: the group's own members must host nothing but this table.
      for (RegionInfo region : ADMIN.getRegions(sn1)) {
        assertEquals(tableName, region.getTable());
      }
      for (RegionInfo region : ADMIN.getRegions(sn2)) {
        assertEquals(tableName, region.getTable());
      }
    } finally {
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testUngracefulCrashOfRegexMemberReassignsRegionsOnlyToOtherGroupMembers()
    throws Exception {
    // ServerCrashProcedure recovers a dead server's regions by default with
    // hbase.master.scp.retain.assignment=false, i.e. forceNewPlan=true for every ASSIGN TRSP it
    // creates -- which routes through roundRobinAssignment rather than retainAssignment (see
    // ServerCrashProcedure#assignRegions). This test's job is to confirm
    // RSGroupBasedLoadBalancer#roundRobinAssignment still confines the
    // crashed regex member's regions to the *other* online members of its own group -- never to
    // default or an unrelated group -- rather than trusting the RSGroup-partitioning logic to
    // "just work". It also asserts, via the test-only call-tracking flags, exactly which
    // assignment methods actually ran for each step: retainAssignment during
    // setRSGroup, and roundRobinAssignment (not retainAssignment) during crash recovery.
    String groupName = getGroupName("crashroundrobin");
    RSGroupBasedLoadBalancer balancer = (RSGroupBasedLoadBalancer) MASTER.getLoadBalancer();
    ADMIN.addRSGroup(groupName);
    // "127.0.0.1" and "localhost" are the only two distinct, real, independently-reachable
    // identities startFakeHostnameRS supports (its RPC bind address is always "localhost"
    // regardless of advertised hostname).
    setRegex(groupName, "127\\.0\\.0\\.1|localhost");
    // A second, unrelated RSGroup holding one server borrowed from default -- proves crash
    // recovery's per-group candidate filtering excludes every other group, not just default.
    String otherGroupName = getGroupName("crashroundrobinother");
    RSGroupInfo otherGroupInfo = addGroup(otherGroupName, 1);
    ServerName otherGroupServer = getServerName(otherGroupInfo.getServers().iterator().next());
    try {
      JVMClusterUtil.RegionServerThread rst1 = startFakeHostnameRS("127.0.0.1");
      Address addr1 = addressOf(rst1);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr1));
      ServerName sn1 = getServerName(addr1);

      JVMClusterUtil.RegionServerThread rst2 = startFakeHostnameRS("localhost");
      Address addr2 = addressOf(rst2);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr2));
      ServerName sn2 = getServerName(addr2);

      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 6);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
      // setRSGroup synchronously relocates every region of a table with no RSGroup-scoped
      // descriptor yet onto the target group's servers (moveTablesAndWait blocks on the
      // ModifyTableProcedure, whose child ReopenTableRegionsProcedure blocks on the reopened
      // regions), so confinement to {sn1, sn2} is already guaranteed once this call returns.
      balancer.resetAssignmentCallFlagsForTest();
      ADMIN.setRSGroup(Sets.newHashSet(tableName), groupName);
      // ReopenTableRegionsProcedure closes and reassigns every region via a single batched
      // retainAssignment call; since none of these regions' previous hosts are group members,
      // the wrapped internal balancer's own retain logic (not this class's randomAssignment)
      // places them on random candidate hosts within the group -- see BaseLoadBalancer's
      // "assigned to random hosts" log path invoked from within retainAssignment.
      assertFalse(balancer.isRandomAssignmentInvoked());
      assertTrue(balancer.isRetainAssignmentInvoked());
      assertFalse(balancer.isRoundRobinAssignmentInvoked());
      Map<ServerName, List<String>> perServerBeforeCrash = getTableServerRegionMap().get(tableName);
      int totalBeforeCrash = 0;
      for (Map.Entry<ServerName, List<String>> entry : perServerBeforeCrash.entrySet()) {
        assertTrue(entry.getKey().equals(sn1) || entry.getKey().equals(sn2));
        totalBeforeCrash += entry.getValue().size();
      }
      assertEquals(6, totalBeforeCrash);

      // retainAssignment's random fallback (BaseLoadBalancer#randomAssignment) places each of
      // these regions independently, so it is possible -- roughly 3% of runs -- for all 6 to land
      // on one of the two servers alone. Crash whichever server actually has regions, so
      // ServerCrashProcedure always has at least one region to reassign and roundRobinAssignment
      // is guaranteed to fire; default to the usual sn1-crashes/sn2-survives case otherwise.
      boolean sn1HasNoRegions =
        !perServerBeforeCrash.containsKey(sn1) || perServerBeforeCrash.get(sn1).isEmpty();
      JVMClusterUtil.RegionServerThread rstToCrash = sn1HasNoRegions ? rst2 : rst1;
      ServerName survivor = sn1HasNoRegions ? sn1 : sn2;

      // Simulate an ungraceful crash
      balancer.resetAssignmentCallFlagsForTest();
      killFakeRegionServer(rstToCrash);

      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> {
        Map<ServerName, List<String>> perServer = getTableServerRegionMap().get(tableName);
        if (perServer == null) {
          return false;
        }
        int total = 0;
        for (Map.Entry<ServerName, List<String>> entry : perServer.entrySet()) {
          // Every region must have landed on the group's sole remaining member -- never on
          // default or any other group, and never back on the now-dead server.
          if (!entry.getKey().equals(survivor)) {
            return false;
          }
          total += entry.getValue().size();
        }
        return total == 6;
      });
      // Confirms the crash-recovery ASSIGN TRSPs went through roundRobinAssignment, not
      // retainAssignment -- expected since hbase.master.scp.retain.assignment defaults to false.
      assertTrue(balancer.isRoundRobinAssignmentInvoked());
      assertFalse(balancer.isRetainAssignmentInvoked());

      // The surviving group member must host nothing but this table's regions.
      for (RegionInfo region : ADMIN.getRegions(survivor)) {
        assertEquals(tableName, region.getTable());
      }
      // The unrelated group's own member must not have received any of the crashed member's
      // regions either -- recovery must stay confined to the crashed server's own group.
      for (RegionInfo region : ADMIN.getRegions(otherGroupServer)) {
        assertFalse(region.getTable().equals(tableName));
      }
    } finally {
      removeRegexGroup(groupName);
      removeGroup(otherGroupName);
    }
  }

  @Test
  public void testFallbackWhenAllRegexMembersCrash() throws Exception {
    // RSGroupBasedLoadBalancer#generateGroupAssignments only ever looks at a group's currently
    // online configured servers, regardless of how they became members -- regex-driven or
    // admin-driven membership is indistinguishable to it. This confirms the
    // hbase.rsgroup.fallback.enable=true fallback-to-default path works the same way when the
    // group that loses every one of its servers is regex-governed.
    String groupName = getGroupName("fallbackregex");
    ADMIN.addRSGroup(groupName);
    // "127.0.0.1" and "localhost" are the only two distinct, real, independently-reachable
    // identities startFakeHostnameRS supports (its RPC bind address is always "localhost"
    // regardless of advertised hostname).
    setRegex(groupName, "127\\.0\\.0\\.1|localhost");
    RSGroupBasedLoadBalancer balancer = (RSGroupBasedLoadBalancer) MASTER.getLoadBalancer();
    balancer.setFallbackEnabledForTest(true);
    try {
      JVMClusterUtil.RegionServerThread rst1 = startFakeHostnameRS("127.0.0.1");
      Address addr1 = addressOf(rst1);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr1));
      ServerName sn1 = getServerName(addr1);

      JVMClusterUtil.RegionServerThread rst2 = startFakeHostnameRS("localhost");
      Address addr2 = addressOf(rst2);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(groupName).getServers().contains(addr2));
      ServerName sn2 = getServerName(addr2);

      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 6);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
      // setRSGroup synchronously relocates every region of a table with no RSGroup-scoped
      // descriptor yet onto the target group's servers (moveTablesAndWait blocks on each
      // ModifyTableProcedure before returning), so confinement to {sn1, sn2} is already guaranteed
      // once this call returns.
      ADMIN.setRSGroup(Sets.newHashSet(tableName), groupName);

      // Both regex members must be hosting regions of the table before either is killed.
      Map<ServerName, List<String>> perServerBeforeCrash = getTableServerRegionMap().get(tableName);
      int totalBeforeCrash = 0;
      for (ServerName member : new ServerName[] { sn1, sn2 }) {
        List<String> regionsOnMember = perServerBeforeCrash.get(member);
        assertFalse(regionsOnMember == null || regionsOnMember.isEmpty());
        totalBeforeCrash += regionsOnMember.size();
      }
      assertEquals(6, totalBeforeCrash);

      // Kill both regex members -- the group is left with zero online servers. Unlike the base
      // cluster's servers (which remain in default throughout this test), these two fake RS were
      // never part of default, so default keeps its original online servers and the fallback
      // path lands there directly, with no need for a second-level fallback.
      killFakeRegionServer(rst1);
      killFakeRegionServer(rst2);

      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> {
        Map<ServerName, List<String>> perServer = getTableServerRegionMap().get(tableName);
        if (perServer == null) {
          return false;
        }
        RSGroupInfo defaultGroup;
        try {
          defaultGroup = ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP);
        } catch (IOException e) {
          return false;
        }
        int total = 0;
        for (Map.Entry<ServerName, List<String>> entry : perServer.entrySet()) {
          // Every region must have landed on a server that is currently in default -- never left
          // stranded on the now-dead sn1/sn2, and never on any unrelated group.
          if (!defaultGroup.getServers().contains(entry.getKey().getAddress())) {
            return false;
          }
          total += entry.getValue().size();
        }
        return total == 6;
      });
    } finally {
      balancer.setFallbackEnabledForTest(false);
      removeRegexGroup(groupName);
    }
  }

  @Test
  public void testExplicitMoveFromDefaultToAdminGroupTriggersRandomAssignmentBackToDefault()
    throws Exception {
    // Three concurrently active RSGroups -- default, a regex-governed group, and an
    // admin-managed group -- each with at least one member. Explicitly moving a server out of
    // default into the admin-managed group (RSGroupInfoManagerImpl#moveServers) leaves that
    // server still hosting regions of a table that belongs to default; moveServerRegionsFromGroup
    // detects the mismatch and relocates those orphaned regions back onto default's *remaining*
    // members via LoadBalancer#randomAssignment, then actually executes each relocation through
    // AssignmentManager#moveAsync -- which, because the destination is already decided, creates a
    // MOVE-type TransitRegionStateProcedure whose queueAssign sets a non-null region location, so
    // AssignmentManager#processAssignQueue routes it into retainMap and retainAssignment also
    // fires. roundRobinAssignment never fires, since no region in this flow is ever left with a
    // null location. This test confirms that exact combination of calls, plus the final,
    // fully-partitioned region placement across all three groups.
    RSGroupBasedLoadBalancer balancer = (RSGroupBasedLoadBalancer) MASTER.getLoadBalancer();
    String regexGroupName = getGroupName("regex3way");
    String adminGroupName = getGroupName("admin3way");
    ADMIN.addRSGroup(regexGroupName);
    // "127.0.0.1" is the only fake identity needed here; it is never a member of default, so it
    // cannot interfere with the base cluster's servers used below.
    setRegex(regexGroupName, "127\\.0\\.0\\.1");
    try {
      JVMClusterUtil.RegionServerThread regexRst = startFakeHostnameRS("127.0.0.1");
      Address regexAddr = addressOf(regexRst);
      TEST_UTIL.waitFor(WAIT_TIMEOUT,
        () -> ADMIN.getRSGroup(regexGroupName).getServers().contains(regexAddr));

      // Admin-managed group starts with one server borrowed from default -- NUM_SLAVES_BASE(4)
      // - 1 = 3 remain in default, enough that moving one more away below still satisfies
      // moveServers' KEEP_ONE_SERVER_IN_DEFAULT_ERROR_MESSAGE constraint (default must always
      // retain more servers than are being moved out of it).
      RSGroupInfo adminGroupInfo = addGroup(adminGroupName, 1);
      Address adminServerInitialAddr = adminGroupInfo.getServers().iterator().next();

      // A table with no explicit RSGroup is implicitly confined to default's current members --
      // here, the 3 servers remaining in default after the admin group's initial carve-out.
      TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("f"), 9);
      TEST_UTIL.waitUntilAllRegionsAssigned(tableName);

      // Pick a default-group server that is actually hosting some of this table's regions -- the
      // server we are about to move, so the orphaned-region relocation has something to do.
      RSGroupInfo defaultGroupBeforeMove = ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP);
      Map<ServerName, List<String>> perServerBeforeMove = getTableServerRegionMap().get(tableName);
      int totalRegionsBeforeMove = 0;
      ServerName movingServer = null;
      for (Map.Entry<ServerName, List<String>> entry : perServerBeforeMove.entrySet()) {
        assertTrue(defaultGroupBeforeMove.getServers().contains(entry.getKey().getAddress()));
        totalRegionsBeforeMove += entry.getValue().size();
        if (movingServer == null && !entry.getValue().isEmpty()) {
          movingServer = entry.getKey();
        }
      }
      assertEquals(9, totalRegionsBeforeMove);
      Address movingAddr = movingServer.getAddress();
      Set<Address> defaultServersBeforeMove = new HashSet<>(defaultGroupBeforeMove.getServers());

      // The action under test: explicitly move that one server out of default into the
      // admin-managed group.
      balancer.resetAssignmentCallFlagsForTest();
      ADMIN.moveServersToRSGroup(Sets.newHashSet(movingAddr), adminGroupName);

      // See the method comment above for why both randomAssignment (picking the relocation plan)
      // and retainAssignment (executing it via moveAsync) fire, while roundRobinAssignment never
      // does.
      assertTrue(balancer.isRandomAssignmentInvoked());
      assertTrue(balancer.isRetainAssignmentInvoked());
      assertFalse(balancer.isRoundRobinAssignmentInvoked());

      // Membership: the moved server now belongs to the admin-managed group, not default; the
      // regex group and the rest of default are untouched by this move.
      Set<Address> expectedDefaultServersAfterMove = new HashSet<>(defaultServersBeforeMove);
      expectedDefaultServersAfterMove.remove(movingAddr);
      assertEquals(expectedDefaultServersAfterMove,
        ADMIN.getRSGroup(RSGroupInfo.DEFAULT_GROUP).getServers());
      assertEquals(Sets.newHashSet(adminServerInitialAddr, movingAddr),
        ADMIN.getRSGroup(adminGroupName).getServers());
      assertEquals(Sets.newHashSet(regexAddr), ADMIN.getRSGroup(regexGroupName).getServers());

      // Regions: every one of the table's 9 regions must have moved off the relocated server,
      // landing only on default's remaining members -- never on the admin-managed or
      // regex-governed group's servers. moveServers blocks until this relocation completes, but
      // poll defensively rather than asserting immediately.
      TEST_UTIL.waitFor(WAIT_TIMEOUT, () -> {
        Map<ServerName, List<String>> perServer = getTableServerRegionMap().get(tableName);
        if (perServer == null) {
          return false;
        }
        int total = 0;
        for (Map.Entry<ServerName, List<String>> entry : perServer.entrySet()) {
          if (!expectedDefaultServersAfterMove.contains(entry.getKey().getAddress())) {
            return false;
          }
          total += entry.getValue().size();
        }
        return total == 9;
      });

      // Bidirectional isolation, now that the move has settled: neither the admin-managed group's
      // servers (the original member and the one just moved in) nor the regex group's server host
      // any region of this default-group table.
      for (Address addr : ADMIN.getRSGroup(adminGroupName).getServers()) {
        for (RegionInfo region : ADMIN.getRegions(getServerName(addr))) {
          assertFalse(region.getTable().equals(tableName));
        }
      }
      for (RegionInfo region : ADMIN.getRegions(getServerName(regexAddr))) {
        assertFalse(region.getTable().equals(tableName));
      }
    } finally {
      removeRegexGroup(regexGroupName);
      removeGroup(adminGroupName);
    }
  }
}