/*
 * 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.client;

import static org.apache.hadoop.hbase.favored.FavoredNodesPlan.Position.PRIMARY;
import static org.apache.hadoop.hbase.favored.FavoredNodesPlan.Position.SECONDARY;
import static org.apache.hadoop.hbase.favored.FavoredNodesPlan.Position.TERTIARY;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseTestingUtil;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.favored.FavoredNodeAssignmentHelper;
import org.apache.hadoop.hbase.favored.FavoredNodesManager;
import org.apache.hadoop.hbase.master.LoadBalancer;
import org.apache.hadoop.hbase.master.ServerManager;
import org.apache.hadoop.hbase.master.balancer.LoadOnlyFavoredStochasticBalancer;
import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.Threads;
import org.junit.jupiter.api.AfterAll;
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.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;

@Tag(ClientTests.TAG)
@Tag(MediumTests.TAG)
public class TestTableFavoredNodes {

  private static final Logger LOG = LoggerFactory.getLogger(TestTableFavoredNodes.class);

  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
  private final static int WAIT_TIMEOUT = 60000;
  private final static int SLAVES = 8;
  private FavoredNodesManager fnm;
  private Admin admin;

  private final byte[][] splitKeys = new byte[][] { Bytes.toBytes(1), Bytes.toBytes(9) };

  private String methodName;

  @BeforeEach
  public void setupMethodName(TestInfo testInfo) {
    this.methodName = testInfo.getTestMethod().get().getName();
  }

  @BeforeAll
  public static void setupBeforeClass() throws Exception {
    Configuration conf = TEST_UTIL.getConfiguration();
    // Setting FavoredNodeBalancer will enable favored nodes
    conf.setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
      LoadOnlyFavoredStochasticBalancer.class, LoadBalancer.class);
    conf.set(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, "" + SLAVES);

    TEST_UTIL.startMiniCluster(SLAVES);
    TEST_UTIL.getMiniHBaseCluster().waitForActiveAndReadyMaster(WAIT_TIMEOUT);
  }

  @AfterAll
  public static void tearDownAfterClass() throws Exception {
    TEST_UTIL.shutdownMiniCluster();
    TEST_UTIL.cleanupTestDir();
  }

  @BeforeEach
  public void setup() throws IOException {
    fnm = TEST_UTIL.getMiniHBaseCluster().getMaster().getFavoredNodesManager();
    admin = TEST_UTIL.getAdmin();
    admin.balancerSwitch(false, true);
    admin.catalogJanitorSwitch(false);
  }

  /*
   * Create a table with FN enabled and check if all its regions have favored nodes set.
   */
  @Test
  public void testCreateTable() throws Exception {
    final TableName tableName = TableName.valueOf(methodName);
    TEST_UTIL.createTable(tableName, Bytes.toBytes("f"), splitKeys);
    TEST_UTIL.waitUntilAllRegionsAssigned(tableName);

    // All regions should have favored nodes
    checkIfFavoredNodeInformationIsCorrect(tableName);

    List<RegionInfo> regions = admin.getRegions(tableName);

    TEST_UTIL.deleteTable(tableName);

    checkNoFNForDeletedTable(regions);
  }

  /*
   * Checks if favored node information is removed on table truncation.
   */
  @Test
  public void testTruncateTable() throws Exception {
    final TableName tableName = TableName.valueOf(methodName);
    TEST_UTIL.createTable(tableName, Bytes.toBytes("f"), splitKeys);
    TEST_UTIL.waitUntilAllRegionsAssigned(tableName);

    // All regions should have favored nodes
    checkIfFavoredNodeInformationIsCorrect(tableName);

    List<RegionInfo> regions = admin.getRegions(tableName);
    TEST_UTIL.truncateTable(tableName, true);

    checkNoFNForDeletedTable(regions);
    checkIfFavoredNodeInformationIsCorrect(tableName);

    regions = admin.getRegions(tableName);
    TEST_UTIL.truncateTable(tableName, false);
    checkNoFNForDeletedTable(regions);

    TEST_UTIL.deleteTable(tableName);
  }

  /*
   * Check if daughters inherit at-least 2 FN from parent after region split.
   */
  @Test
  public void testSplitTable() throws Exception {
    final TableName tableName = TableName.valueOf(methodName);
    Table t = TEST_UTIL.createTable(tableName, Bytes.toBytes("f"), splitKeys);
    TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
    final int numberOfRegions = admin.getRegions(t.getName()).size();

    checkIfFavoredNodeInformationIsCorrect(tableName);

    byte[] splitPoint = Bytes.toBytes(0);
    RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName);
    RegionInfo parent = locator.getRegionLocation(splitPoint).getRegion();
    List<ServerName> parentFN = fnm.getFavoredNodes(parent);
    assertNotNull(parentFN, "FN should not be null for region: " + parent);

    LOG.info("SPLITTING TABLE");
    admin.split(tableName, splitPoint);

    TEST_UTIL.waitUntilNoRegionsInTransition(WAIT_TIMEOUT);
    LOG.info("FINISHED WAITING ON RIT");
    waitUntilTableRegionCountReached(tableName, numberOfRegions + 1);

    // All regions should have favored nodes checkIfFavoredNodeInformationIsCorrect(tableName);

    // Get the daughters of parent.
    RegionInfo daughter1 = locator.getRegionLocation(parent.getStartKey(), true).getRegion();
    List<ServerName> daughter1FN = fnm.getFavoredNodes(daughter1);

    RegionInfo daughter2 = locator.getRegionLocation(splitPoint, true).getRegion();
    List<ServerName> daughter2FN = fnm.getFavoredNodes(daughter2);

    checkIfDaughterInherits2FN(parentFN, daughter1FN);
    checkIfDaughterInherits2FN(parentFN, daughter2FN);

    assertEquals(parentFN.get(PRIMARY.ordinal()), daughter1FN.get(PRIMARY.ordinal()),
      "Daughter's PRIMARY FN should be PRIMARY of parent");
    assertEquals(parentFN.get(SECONDARY.ordinal()), daughter1FN.get(SECONDARY.ordinal()),
      "Daughter's SECONDARY FN should be SECONDARY of parent");

    assertEquals(parentFN.get(PRIMARY.ordinal()), daughter2FN.get(PRIMARY.ordinal()),
      "Daughter's PRIMARY FN should be PRIMARY of parent");
    assertEquals(parentFN.get(TERTIARY.ordinal()), daughter2FN.get(SECONDARY.ordinal()),
      "Daughter's SECONDARY FN should be TERTIARY of parent");

    // Major compact table and run catalog janitor. Parent's FN should be removed
    TEST_UTIL.getMiniHBaseCluster().compact(tableName, true);
    admin.runCatalogJanitor();
    // Catalog cleanup is async. Wait on procedure to finish up.
    ProcedureTestingUtility
      .waitAllProcedures(TEST_UTIL.getMiniHBaseCluster().getMaster().getMasterProcedureExecutor());
    // assertEquals("Parent region should have been cleaned", 1, admin.runCatalogScan());
    assertNull(fnm.getFavoredNodes(parent), "Parent FN should be null");

    List<RegionInfo> regions = admin.getRegions(tableName);
    // Split and Table Disable interfere with each other around region replicas
    // TODO. Meantime pause a few seconds.
    Threads.sleep(2000);
    LOG.info("STARTING DELETE");
    TEST_UTIL.deleteTable(tableName);

    checkNoFNForDeletedTable(regions);
  }

  /*
   * Check if merged region inherits FN from one of its regions.
   */
  @Test
  public void testMergeTable() throws Exception {
    final TableName tableName = TableName.valueOf(methodName);
    TEST_UTIL.createTable(tableName, Bytes.toBytes("f"), splitKeys);
    TEST_UTIL.waitUntilAllRegionsAssigned(tableName);

    checkIfFavoredNodeInformationIsCorrect(tableName);

    RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName);
    RegionInfo regionA = locator.getRegionLocation(HConstants.EMPTY_START_ROW).getRegion();
    RegionInfo regionB = locator.getRegionLocation(splitKeys[0]).getRegion();

    List<ServerName> regionAFN = fnm.getFavoredNodes(regionA);
    LOG.info("regionA: " + regionA.getEncodedName() + " with FN: " + fnm.getFavoredNodes(regionA));
    LOG.info("regionB: " + regionA.getEncodedName() + " with FN: " + fnm.getFavoredNodes(regionB));

    int countOfRegions = TEST_UTIL.getMiniHBaseCluster().getRegions(tableName).size();
    admin.mergeRegionsAsync(regionA.getEncodedNameAsBytes(), regionB.getEncodedNameAsBytes(), false)
      .get(60, TimeUnit.SECONDS);

    TEST_UTIL.waitUntilNoRegionsInTransition(WAIT_TIMEOUT);
    waitUntilTableRegionCountReached(tableName, countOfRegions - 1);

    // All regions should have favored nodes
    checkIfFavoredNodeInformationIsCorrect(tableName);

    RegionInfo mergedRegion = locator.getRegionLocation(HConstants.EMPTY_START_ROW).getRegion();
    List<ServerName> mergedFN = fnm.getFavoredNodes(mergedRegion);

    assertArrayEquals(regionAFN.toArray(), mergedFN.toArray(),
      "Merged region doesn't match regionA's FN");

    // Major compact table and run catalog janitor. Parent FN should be removed
    TEST_UTIL.getMiniHBaseCluster().compact(tableName, true);
    assertEquals(1, admin.runCatalogJanitor(), "Merge parents should have been cleaned");
    // Catalog cleanup is async. Wait on procedure to finish up.
    ProcedureTestingUtility
      .waitAllProcedures(TEST_UTIL.getMiniHBaseCluster().getMaster().getMasterProcedureExecutor());
    assertNull(fnm.getFavoredNodes(regionA), "Parent FN should be null");
    assertNull(fnm.getFavoredNodes(regionB), "Parent FN should be null");

    List<RegionInfo> regions = admin.getRegions(tableName);

    TEST_UTIL.deleteTable(tableName);

    checkNoFNForDeletedTable(regions);
  }

  private void checkNoFNForDeletedTable(List<RegionInfo> regions) {
    for (RegionInfo region : regions) {
      LOG.info("Testing if FN data for " + region);
      assertNull(fnm.getFavoredNodes(region), "FN not null for deleted table's region: " + region);
    }
  }

  /*
   * This checks the following: 1. Do all regions of the table have favored nodes updated in master?
   * 2. Is the number of favored nodes correct for a region? Is the start code -1? 3. Is the FN
   * information consistent between Master and the respective RegionServer?
   */
  private void checkIfFavoredNodeInformationIsCorrect(TableName tableName) throws Exception {
    /*
     * Since we need HRegionServer to check for consistency of FN between Master and RS, lets
     * construct a map for each serverName lookup. Makes it easy later.
     */
    Map<ServerName, HRegionServer> snRSMap = Maps.newHashMap();
    for (JVMClusterUtil.RegionServerThread rst : TEST_UTIL.getMiniHBaseCluster()
      .getLiveRegionServerThreads()) {
      snRSMap.put(rst.getRegionServer().getServerName(), rst.getRegionServer());
    }

    int dnPort = FavoredNodeAssignmentHelper.getDataNodePort(TEST_UTIL.getConfiguration());
    RegionLocator regionLocator = admin.getConnection().getRegionLocator(tableName);
    for (HRegionLocation regionLocation : regionLocator.getAllRegionLocations()) {

      RegionInfo regionInfo = regionLocation.getRegion();
      List<ServerName> fnList = fnm.getFavoredNodes(regionInfo);

      // 1. Does each region have favored node?
      assertNotNull(fnList, "Favored nodes should not be null for region:" + regionInfo);

      // 2. Do we have the right number of favored nodes? Is start code -1?
      assertEquals(FavoredNodeAssignmentHelper.FAVORED_NODES_NUM, fnList.size(),
        "Incorrect favored nodes for region:" + regionInfo + " fnlist: " + fnList);
      for (ServerName sn : fnList) {
        assertEquals(-1, sn.getStartCode(), "FN should not have startCode, fnlist:" + fnList);
      }

      // 3. Check if the regionServers have all the FN updated and in sync with Master
      HRegionServer regionServer = snRSMap.get(regionLocation.getServerName());
      assertNotNull(regionServer, "RS should not be null for regionLocation: " + regionLocation);

      InetSocketAddress[] rsFavNodes =
        regionServer.getFavoredNodesForRegion(regionInfo.getEncodedName());
      assertNotNull(rsFavNodes,
        "RS " + regionLocation.getServerName() + " does not have FN for region: " + regionInfo);
      assertEquals(FavoredNodeAssignmentHelper.FAVORED_NODES_NUM, rsFavNodes.length,
        "Incorrect FN for region:" + regionInfo.getEncodedName() + " on server:"
          + regionLocation.getServerName());

      // 4. Does DN port match all FN node list?
      for (ServerName sn : fnm.getFavoredNodesWithDNPort(regionInfo)) {
        assertEquals(-1, sn.getStartCode(), "FN should not have startCode, fnlist:" + fnList);
        assertEquals(dnPort, sn.getPort(), "FN port should belong to DN port, fnlist:" + fnList);
      }
    }
  }

  /*
   * Check favored nodes for system tables
   */
  @Test
  public void testSystemTables() throws Exception {
    final TableName tableName = TableName.valueOf(methodName);
    TEST_UTIL.createTable(tableName, Bytes.toBytes("f"), splitKeys);
    TEST_UTIL.waitUntilAllRegionsAssigned(tableName);

    // All regions should have favored nodes
    checkIfFavoredNodeInformationIsCorrect(tableName);

    for (TableName sysTable : admin
      .listTableNamesByNamespace(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) {
      List<RegionInfo> regions = admin.getRegions(sysTable);
      for (RegionInfo region : regions) {
        assertNull(fnm.getFavoredNodes(region), "FN should be null for sys region");
      }
    }

    TEST_UTIL.deleteTable(tableName);
  }

  private void checkIfDaughterInherits2FN(List<ServerName> parentFN, List<ServerName> daughterFN) {

    assertNotNull(parentFN);
    assertNotNull(daughterFN);

    List<ServerName> favoredNodes = Lists.newArrayList(daughterFN);
    favoredNodes.removeAll(parentFN);

    /*
     * With a small cluster its likely some FN might accidentally get shared. Its likely the 3rd FN
     * the balancer chooses might still belong to the parent in which case favoredNodes size would
     * be 0.
     */
    assertTrue(favoredNodes.size() <= 1,
      "Daughter FN:" + daughterFN + " should have inherited 2 FN from parent FN:" + parentFN);
  }

  private void waitUntilTableRegionCountReached(final TableName tableName, final int numRegions)
    throws Exception {
    TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        try (RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName)) {
          return locator.getAllRegionLocations().size() == numRegions;
        }
      }
    });
  }
}