Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-25926 Cleanup MetaTableAccessor references in FavoredNodeBalancer related code (#3313) 

Signed-off-by: Yulin Niu <niuyulin@apache.org>
This commit is contained in:
Duo Zhang 2021-05-27 16:05:14 +08:00
parent e265eccf20
commit 3cf0678aee
3 changed files with 149 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
hbase-server/src/main/java/org/apache/hadoop/hbase/favored/FavoredNodeAssignmentHelper.java
View File

@ -28,12 +28,11 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell.Type;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellBuilderFactory;
import org.apache.hadoop.hbase.CellBuilderType;
import org.apache.hadoop.hbase.HBaseIOException;
@ -55,6 +54,8 @@ import org.slf4j.LoggerFactory;
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.FavoredNodes;
@ -75,7 +76,6 @@ public class FavoredNodeAssignmentHelper {
// This map serves as a cache for rack to sn lookups. The num of
// region server entries might not match with that is in servers.
private Map<String, String> regionServerToRackMap;
private Random random;
private List<ServerName> servers;
public static final byte [] FAVOREDNODES_QUALIFIER = Bytes.toBytes("fn");
public final static short FAVORED_NODES_NUM = 3;
@ -92,7 +92,6 @@ public class FavoredNodeAssignmentHelper {
this.rackToRegionServerMap = new HashMap<>();
this.regionServerToRackMap = new HashMap<>();
this.uniqueRackList = new ArrayList<>();
this.random = new Random();
}
// Always initialize() when FavoredNodeAssignmentHelper is constructed.
@ -121,80 +120,58 @@ public class FavoredNodeAssignmentHelper {
* Update meta table with favored nodes info
* @param regionToFavoredNodes map of RegionInfo's to their favored nodes
* @param connection connection to be used
* @throws IOException
*/
public static void updateMetaWithFavoredNodesInfo(
Map<RegionInfo, List<ServerName>> regionToFavoredNodes,
Connection connection) throws IOException {
Map<RegionInfo, List<ServerName>> regionToFavoredNodes, Connection connection)
throws IOException {
List<Put> puts = new ArrayList<>();
for (Map.Entry<RegionInfo, List<ServerName>> entry : regionToFavoredNodes.entrySet()) {
Put put = makePutFromRegionInfo(entry.getKey(), entry.getValue());
Put put = makePut(entry.getKey(), entry.getValue());
if (put != null) {
puts.add(put);
}
}
MetaTableAccessor.putsToMetaTable(connection, puts);
LOG.info("Added " + puts.size() + " regions in META");
try (Table table = connection.getTable(TableName.META_TABLE_NAME)) {
table.put(puts);
}
LOG.info("Added " + puts.size() + " region favored nodes in META");
}
/**
* Update meta table with favored nodes info
* @param regionToFavoredNodes
* @param conf
* @throws IOException
*/
public static void updateMetaWithFavoredNodesInfo(
Map<RegionInfo, List<ServerName>> regionToFavoredNodes,
Configuration conf) throws IOException {
List<Put> puts = new ArrayList<>();
for (Map.Entry<RegionInfo, List<ServerName>> entry : regionToFavoredNodes.entrySet()) {
Put put = makePutFromRegionInfo(entry.getKey(), entry.getValue());
if (put != null) {
puts.add(put);
}
}
Map<RegionInfo, List<ServerName>> regionToFavoredNodes, Configuration conf) throws IOException {
// Write the region assignments to the meta table.
// TODO: See above overrides take a Connection rather than a Configuration only the
// Connection is a short circuit connection. That is not going to good in all cases, when
// master and meta are not colocated. Fix when this favored nodes feature is actually used
// someday.
try (Connection connection = ConnectionFactory.createConnection(conf)) {
try (Table metaTable = connection.getTable(TableName.META_TABLE_NAME)) {
metaTable.put(puts);
}
try (Connection conn = ConnectionFactory.createConnection(conf)) {
updateMetaWithFavoredNodesInfo(regionToFavoredNodes, conn);
}
LOG.info("Added " + puts.size() + " regions in META");
}
/**
* Generates and returns a Put containing the region info for the catalog table and the servers
* @return Put object
*/
private static Put makePutFromRegionInfo(RegionInfo regionInfo, List<ServerName> favoredNodeList)
throws IOException {
Put put = null;
if (favoredNodeList != null) {
long time = EnvironmentEdgeManager.currentTime();
put = MetaTableAccessor.makePutFromRegionInfo(regionInfo, time);
byte[] favoredNodes = getFavoredNodes(favoredNodeList);
put.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY)
.setRow(put.getRow())
.setFamily(HConstants.CATALOG_FAMILY)
.setQualifier(FAVOREDNODES_QUALIFIER)
.setTimestamp(time)
.setType(Type.Put)
.setValue(favoredNodes)
.build());
LOG.debug("Create the region {} with favored nodes {}", regionInfo.getRegionNameAsString(),
favoredNodeList);
private static Put makePut(RegionInfo regionInfo, List<ServerName> favoredNodeList)
throws IOException {
if (CollectionUtils.isEmpty(favoredNodeList)) {
return null;
}
long time = EnvironmentEdgeManager.currentTime();
Put put = new Put(MetaTableAccessor.getMetaKeyForRegion(regionInfo), time);
byte[] favoredNodes = getFavoredNodes(favoredNodeList);
put.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setRow(put.getRow())
.setFamily(HConstants.CATALOG_FAMILY).setQualifier(FAVOREDNODES_QUALIFIER).setTimestamp(time)
.setType(Cell.Type.Put).setValue(favoredNodes).build());
LOG.debug("Create the region {} with favored nodes {}", regionInfo.getRegionNameAsString(),
favoredNodeList);
return put;
}
/**
* Convert PB bytes to ServerName.
* @param favoredNodes The PB'ed bytes of favored nodes
* @return the array of {@link ServerName} for the byte array of favored nodes.
* @throws IOException
*/
public static ServerName[] getFavoredNodesList(byte[] favoredNodes) throws IOException {
FavoredNodes f = FavoredNodes.parseFrom(favoredNodes);
@ -236,7 +213,7 @@ public class FavoredNodeAssignmentHelper {
Map<RegionInfo, ServerName> primaryRSMap, List<RegionInfo> regions) {
List<String> rackList = new ArrayList<>(rackToRegionServerMap.size());
rackList.addAll(rackToRegionServerMap.keySet());
int rackIndex = random.nextInt(rackList.size());
int rackIndex = ThreadLocalRandom.current().nextInt(rackList.size());
int maxRackSize = 0;
for (Map.Entry<String,List<ServerName>> r : rackToRegionServerMap.entrySet()) {
if (r.getValue().size() > maxRackSize) {
@ -245,7 +222,7 @@ public class FavoredNodeAssignmentHelper {
}
int numIterations = 0;
// Initialize the current processing host index.
int serverIndex = random.nextInt(maxRackSize);
int serverIndex = ThreadLocalRandom.current().nextInt(maxRackSize);
for (RegionInfo regionInfo : regions) {
List<ServerName> currentServerList;
String rackName;
@ -590,7 +567,7 @@ public class FavoredNodeAssignmentHelper {
}
ServerName randomServer = null;
int randomIndex = random.nextInt(serversToChooseFrom.size());
int randomIndex = ThreadLocalRandom.current().nextInt(serversToChooseFrom.size());
int j = 0;
for (StartcodeAgnosticServerName sn : serversToChooseFrom) {
if (j == randomIndex) {
@ -611,14 +588,14 @@ public class FavoredNodeAssignmentHelper {
return this.getOneRandomServer(rack, null);
}
protected String getOneRandomRack(Set<String> skipRackSet) throws IOException {
String getOneRandomRack(Set<String> skipRackSet) throws IOException {
if (skipRackSet == null || uniqueRackList.size() <= skipRackSet.size()) {
throw new IOException("Cannot randomly pick another random server");
}
String randomRack;
do {
int randomIndex = random.nextInt(this.uniqueRackList.size());
int randomIndex = ThreadLocalRandom.current().nextInt(this.uniqueRackList.size());
randomRack = this.uniqueRackList.get(randomIndex);
} while (skipRackSet.contains(randomRack));
@ -772,7 +749,7 @@ public class FavoredNodeAssignmentHelper {
public List<ServerName> generateFavoredNodes(RegionInfo hri) throws IOException {
List<ServerName> favoredNodesForRegion = new ArrayList<>(FAVORED_NODES_NUM);
ServerName primary = servers.get(random.nextInt(servers.size()));
ServerName primary = servers.get(ThreadLocalRandom.current().nextInt(servers.size()));
favoredNodesForRegion.add(ServerName.valueOf(primary.getAddress(), ServerName.NON_STARTCODE));
Map<RegionInfo, ServerName> primaryRSMap = new HashMap<>(1);

149
hbase-server/src/main/java/org/apache/hadoop/hbase/master/SnapshotOfRegionAssignmentFromMeta.java
View File

@ -31,17 +31,17 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.MetaTableAccessor;
import org.apache.hadoop.hbase.MetaTableAccessor.Visitor;
import org.apache.hadoop.hbase.RegionLocations;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.favored.FavoredNodeAssignmentHelper;
import org.apache.hadoop.hbase.favored.FavoredNodesPlan;
import org.apache.yetus.audience.InterfaceAudience;
@ -98,78 +98,97 @@ public class SnapshotOfRegionAssignmentFromMeta {
this.excludeOfflinedSplitParents = excludeOfflinedSplitParents;
}
private void processMetaRecord(Result result) throws IOException {
if (result == null || result.isEmpty()) {
return;
}
RegionLocations rl = MetaTableAccessor.getRegionLocations(result);
if (rl == null) {
return;
}
RegionInfo hri = rl.getRegionLocation(0).getRegion();
if (hri == null) {
return;
}
if (hri.getTable() == null) {
return;
}
if (disabledTables.contains(hri.getTable())) {
return;
}
// Are we to include split parents in the list?
if (excludeOfflinedSplitParents && hri.isSplit()) {
return;
}
HRegionLocation[] hrls = rl.getRegionLocations();
// Add the current assignment to the snapshot for all replicas
for (int i = 0; i < hrls.length; i++) {
if (hrls[i] == null) {
continue;
}
hri = hrls[i].getRegion();
if (hri == null) {
continue;
}
addAssignment(hri, hrls[i].getServerName());
addRegion(hri);
}
hri = rl.getRegionLocation(0).getRegion();
// the code below is to handle favored nodes
byte[] favoredNodes = result.getValue(HConstants.CATALOG_FAMILY,
FavoredNodeAssignmentHelper.FAVOREDNODES_QUALIFIER);
if (favoredNodes == null) {
return;
}
// Add the favored nodes into assignment plan
ServerName[] favoredServerList = FavoredNodeAssignmentHelper.getFavoredNodesList(favoredNodes);
// Add the favored nodes into assignment plan
existingAssignmentPlan.updateFavoredNodesMap(hri, Arrays.asList(favoredServerList));
/*
* Typically there should be FAVORED_NODES_NUM favored nodes for a region in meta. If there is
* less than FAVORED_NODES_NUM, lets use as much as we can but log a warning.
*/
if (favoredServerList.length != FavoredNodeAssignmentHelper.FAVORED_NODES_NUM) {
LOG.warn("Insufficient favored nodes for region " + hri + " fn: " +
Arrays.toString(favoredServerList));
}
for (int i = 0; i < favoredServerList.length; i++) {
if (i == PRIMARY.ordinal()) {
addPrimaryAssignment(hri, favoredServerList[i]);
}
if (i == SECONDARY.ordinal()) {
addSecondaryAssignment(hri, favoredServerList[i]);
}
if (i == TERTIARY.ordinal()) {
addTeritiaryAssignment(hri, favoredServerList[i]);
}
}
}
/**
* Initialize the region assignment snapshot by scanning the hbase:meta table
* @throws IOException
*/
public void initialize() throws IOException {
LOG.info("Start to scan the hbase:meta for the current region assignment " +
"snappshot");
// TODO: at some point this code could live in the MetaTableAccessor
Visitor v = new Visitor() {
@Override
public boolean visit(Result result) throws IOException {
LOG.info("Start to scan the hbase:meta for the current region assignment " + "snappshot");
// Scan hbase:meta to pick up user regions
try (Table metaTable = connection.getTable(TableName.META_TABLE_NAME);
ResultScanner scanner = metaTable.getScanner(HConstants.CATALOG_FAMILY)) {
for (;;) {
Result result = scanner.next();
if (result == null) {
break;
}
try {
if (result == null || result.isEmpty()) return true;
RegionLocations rl = MetaTableAccessor.getRegionLocations(result);
if (rl == null) return true;
RegionInfo hri = rl.getRegionLocation(0).getRegionInfo();
if (hri == null) return true;
if (hri.getTable() == null) return true;
if (disabledTables.contains(hri.getTable())) {
return true;
}
// Are we to include split parents in the list?
if (excludeOfflinedSplitParents && hri.isSplit()) return true;
HRegionLocation[] hrls = rl.getRegionLocations();
// Add the current assignment to the snapshot for all replicas
for (int i = 0; i < hrls.length; i++) {
if (hrls[i] == null) continue;
hri = hrls[i].getRegionInfo();
if (hri == null) continue;
addAssignment(hri, hrls[i].getServerName());
addRegion(hri);
}
hri = rl.getRegionLocation(0).getRegionInfo();
// the code below is to handle favored nodes
byte[] favoredNodes = result.getValue(HConstants.CATALOG_FAMILY,
FavoredNodeAssignmentHelper.FAVOREDNODES_QUALIFIER);
if (favoredNodes == null) return true;
// Add the favored nodes into assignment plan
ServerName[] favoredServerList =
FavoredNodeAssignmentHelper.getFavoredNodesList(favoredNodes);
// Add the favored nodes into assignment plan
existingAssignmentPlan.updateFavoredNodesMap(hri,
Arrays.asList(favoredServerList));
/*
* Typically there should be FAVORED_NODES_NUM favored nodes for a region in meta. If
* there is less than FAVORED_NODES_NUM, lets use as much as we can but log a warning.
*/
if (favoredServerList.length != FavoredNodeAssignmentHelper.FAVORED_NODES_NUM) {
LOG.warn("Insufficient favored nodes for region " + hri + " fn: " + Arrays
.toString(favoredServerList));
}
for (int i = 0; i < favoredServerList.length; i++) {
if (i == PRIMARY.ordinal()) addPrimaryAssignment(hri, favoredServerList[i]);
if (i == SECONDARY.ordinal()) addSecondaryAssignment(hri, favoredServerList[i]);
if (i == TERTIARY.ordinal()) addTeritiaryAssignment(hri, favoredServerList[i]);
}
return true;
processMetaRecord(result);
} catch (RuntimeException e) {
LOG.error("Catche remote exception " + e.getMessage() +
" when processing" + result);
LOG.error("Catch remote exception " + e.getMessage() + " when processing" + result);
throw e;
}
}
};
// Scan hbase:meta to pick up user regions
MetaTableAccessor.fullScanRegions(connection, v);
//regionToRegionServerMap = regions;
LOG.info("Finished to scan the hbase:meta for the current region assignment" +
"snapshot");
}
LOG.info("Finished to scan the hbase:meta for the current region assignment" + "snapshot");
}
private void addRegion(RegionInfo regionInfo) {

63
hbase-server/src/test/java/org/apache/hadoop/hbase/favored/TestFavoredNodeAssignmentHelper.java
View File

@ -22,6 +22,8 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.ArrayList;
@ -31,7 +33,6 @@ import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.ServerName;
@ -39,8 +40,8 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.master.RackManager;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Triple;
import org.junit.BeforeClass;
@ -54,7 +55,7 @@ import org.mockito.Mockito;
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
@Category({MasterTests.class, LargeTests.class})
@Category({ MasterTests.class, MediumTests.class })
public class TestFavoredNodeAssignmentHelper {
@ClassRule
@ -71,35 +72,36 @@ public class TestFavoredNodeAssignmentHelper {
@Rule
public TestName name = new TestName();
private static String getRack(int index) {
if (index < 10) {
return "rack1";
} else if (index < 20) {
return "rack2";
} else if (index < 30) {
return "rack3";
} else {
return RackManager.UNKNOWN_RACK;
}
}
@BeforeClass
public static void setupBeforeClass() throws Exception {
// Set up some server -> rack mappings
// Have three racks in the cluster with 10 hosts each.
when(rackManager.getRack(any(ServerName.class))).then(invocation -> {
ServerName sn = invocation.getArgument(0, ServerName.class);
try {
int i = Integer.parseInt(sn.getHostname().substring("foo".length()));
return getRack(i);
} catch (NumberFormatException e) {
return RackManager.UNKNOWN_RACK;
}
});
for (int i = 0; i < 40; i++) {
ServerName server = ServerName.valueOf("foo" + i + ":1234", -1);
if (i < 10) {
Mockito.when(rackManager.getRack(server)).thenReturn("rack1");
if (rackToServers.get("rack1") == null) {
List<ServerName> servers = new ArrayList<>();
rackToServers.put("rack1", servers);
}
rackToServers.get("rack1").add(server);
}
if (i >= 10 && i < 20) {
Mockito.when(rackManager.getRack(server)).thenReturn("rack2");
if (rackToServers.get("rack2") == null) {
List<ServerName> servers = new ArrayList<>();
rackToServers.put("rack2", servers);
}
rackToServers.get("rack2").add(server);
}
if (i >= 20 && i < 30) {
Mockito.when(rackManager.getRack(server)).thenReturn("rack3");
if (rackToServers.get("rack3") == null) {
List<ServerName> servers = new ArrayList<>();
rackToServers.put("rack3", servers);
}
rackToServers.get("rack3").add(server);
ServerName server = ServerName.valueOf("foo" + i, 1234, System.currentTimeMillis());
String rack = getRack(i);
if (!rack.equals(RackManager.UNKNOWN_RACK)) {
rackToServers.computeIfAbsent(rack, k -> new ArrayList<>()).add(server);
}
servers.add(server);
}
@ -107,7 +109,7 @@ public class TestFavoredNodeAssignmentHelper {
// The tests decide which racks to work with, and how many machines to
// work with from any given rack
// Return a rondom 'count' number of servers from 'rack'
// Return a random 'count' number of servers from 'rack'
private static List<ServerName> getServersFromRack(Map<String, Integer> rackToServerCount) {
List<ServerName> chosenServers = new ArrayList<>();
for (Map.Entry<String, Integer> entry : rackToServerCount.entrySet()) {
@ -123,11 +125,10 @@ public class TestFavoredNodeAssignmentHelper {
public void testSmallCluster() {
// Test the case where we cannot assign favored nodes (because the number
// of nodes in the cluster is too less)
Map<String,Integer> rackToServerCount = new HashMap<>();
Map<String, Integer> rackToServerCount = new HashMap<>();
rackToServerCount.put("rack1", 2);
List<ServerName> servers = getServersFromRack(rackToServerCount);
FavoredNodeAssignmentHelper helper = new FavoredNodeAssignmentHelper(servers,
new Configuration());
FavoredNodeAssignmentHelper helper = new FavoredNodeAssignmentHelper(servers, rackManager);
helper.initialize();
assertFalse(helper.canPlaceFavoredNodes());
}