You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2017/02/14 20:18:28 UTC
[2/3] hbase git commit: HBASE-17624 Address late review of HBASE-6721,
rsgroups feature Addresses review comments by Sean Busbey and Appy that
happened to come in long after the commit of HBASE-6721,
the original rsgroup issue.
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java
index b83a308..3de1b1d 100644
--- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java
+++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -25,7 +23,6 @@ import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import com.google.common.net.HostAndPort;
import java.io.IOException;
import java.util.ArrayList;
@@ -55,6 +52,7 @@ import org.apache.hadoop.hbase.master.LoadBalancer;
import org.apache.hadoop.hbase.master.MasterServices;
import org.apache.hadoop.hbase.master.RegionPlan;
import org.apache.hadoop.hbase.master.balancer.StochasticLoadBalancer;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.util.ReflectionUtils;
/**
@@ -81,7 +79,9 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
private Configuration config;
private ClusterStatus clusterStatus;
private MasterServices masterServices;
- private RSGroupInfoManager RSGroupInfoManager;
+ // Synchronize on access until we take the time to cmoe up with a finer-grained
+ // locking regime.
+ private volatile RSGroupInfoManager rsGroupInfoManager;
private LoadBalancer internalBalancer;
//used during reflection by LoadBalancerFactory
@@ -91,8 +91,8 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
//This constructor should only be used for unit testing
@InterfaceAudience.Private
- public RSGroupBasedLoadBalancer(RSGroupInfoManager RSGroupInfoManager) {
- this.RSGroupInfoManager = RSGroupInfoManager;
+ public RSGroupBasedLoadBalancer(RSGroupInfoManager rsGroupInfoManager) {
+ this.rsGroupInfoManager = rsGroupInfoManager;
}
@Override
@@ -142,14 +142,15 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
regionPlans.add(new RegionPlan(regionInfo, null, null));
}
try {
- for (RSGroupInfo info : RSGroupInfoManager.listRSGroups()) {
+ List<RSGroupInfo> rsgi = rsGroupInfoManager.listRSGroups();
+ for (RSGroupInfo info: rsgi) {
Map<ServerName, List<HRegionInfo>> groupClusterState =
new HashMap<ServerName, List<HRegionInfo>>();
Map<TableName, Map<ServerName, List<HRegionInfo>>> groupClusterLoad =
new HashMap<TableName, Map<ServerName, List<HRegionInfo>>>();
- for (HostAndPort sName : info.getServers()) {
+ for (Address sName : info.getServers()) {
for(ServerName curr: clusterState.keySet()) {
- if(curr.getHostPort().equals(sName)) {
+ if(curr.getAddress().equals(sName)) {
groupClusterState.put(curr, correctedState.get(curr));
}
}
@@ -199,7 +200,7 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
Set<HRegionInfo> misplacedRegions = getMisplacedRegions(regions);
for (HRegionInfo region : regions.keySet()) {
if (!misplacedRegions.contains(region)) {
- String groupName = RSGroupInfoManager.getRSGroupOfTable(region.getTable());
+ String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
groupToRegion.put(groupName, region);
}
}
@@ -208,7 +209,7 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
for (String key : groupToRegion.keySet()) {
Map<HRegionInfo, ServerName> currentAssignmentMap = new TreeMap<HRegionInfo, ServerName>();
List<HRegionInfo> regionList = groupToRegion.get(key);
- RSGroupInfo info = RSGroupInfoManager.getRSGroup(key);
+ RSGroupInfo info = rsGroupInfoManager.getRSGroup(key);
List<ServerName> candidateList = filterOfflineServers(info, servers);
for (HRegionInfo region : regionList) {
currentAssignmentMap.put(region, regions.get(region));
@@ -220,9 +221,8 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
}
for (HRegionInfo region : misplacedRegions) {
- String groupName = RSGroupInfoManager.getRSGroupOfTable(
- region.getTable());
- RSGroupInfo info = RSGroupInfoManager.getRSGroup(groupName);
+ String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());;
+ RSGroupInfo info = rsGroupInfoManager.getRSGroup(groupName);
List<ServerName> candidateList = filterOfflineServers(info, servers);
ServerName server = this.internalBalancer.randomAssignment(region,
candidateList);
@@ -262,14 +262,14 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
ListMultimap<String, ServerName> serverMap) throws HBaseIOException {
try {
for (HRegionInfo region : regions) {
- String groupName = RSGroupInfoManager.getRSGroupOfTable(region.getTable());
- if(groupName == null) {
+ String groupName = rsGroupInfoManager.getRSGroupOfTable(region.getTable());
+ if (groupName == null) {
LOG.warn("Group for table "+region.getTable()+" is null");
}
regionMap.put(groupName, region);
}
for (String groupKey : regionMap.keySet()) {
- RSGroupInfo info = RSGroupInfoManager.getRSGroup(groupKey);
+ RSGroupInfo info = rsGroupInfoManager.getRSGroup(groupKey);
serverMap.putAll(groupKey, filterOfflineServers(info, servers));
if(serverMap.get(groupKey).size() < 1) {
serverMap.put(groupKey, LoadBalancer.BOGUS_SERVER_NAME);
@@ -285,7 +285,7 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
if (RSGroupInfo != null) {
return filterServers(RSGroupInfo.getServers(), onlineServers);
} else {
- LOG.debug("Group Information found to be null. Some regions might be unassigned.");
+ LOG.warn("RSGroup Information found to be null. Some regions might be unassigned.");
return Collections.EMPTY_LIST;
}
}
@@ -299,12 +299,12 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
* List of servers which are online.
* @return the list
*/
- private List<ServerName> filterServers(Collection<HostAndPort> servers,
+ private List<ServerName> filterServers(Collection<Address> servers,
Collection<ServerName> onlineServers) {
ArrayList<ServerName> finalList = new ArrayList<ServerName>();
- for (HostAndPort server : servers) {
+ for (Address server : servers) {
for(ServerName curr: onlineServers) {
- if(curr.getHostPort().equals(server)) {
+ if(curr.getAddress().equals(server)) {
finalList.add(curr);
}
}
@@ -312,31 +312,23 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
return finalList;
}
- private ListMultimap<String, HRegionInfo> groupRegions(
- List<HRegionInfo> regionList) throws IOException {
- ListMultimap<String, HRegionInfo> regionGroup = ArrayListMultimap
- .create();
- for (HRegionInfo region : regionList) {
- String groupName = RSGroupInfoManager.getRSGroupOfTable(region.getTable());
- regionGroup.put(groupName, region);
- }
- return regionGroup;
- }
-
private Set<HRegionInfo> getMisplacedRegions(
Map<HRegionInfo, ServerName> regions) throws IOException {
Set<HRegionInfo> misplacedRegions = new HashSet<HRegionInfo>();
for(Map.Entry<HRegionInfo, ServerName> region : regions.entrySet()) {
HRegionInfo regionInfo = region.getKey();
ServerName assignedServer = region.getValue();
- RSGroupInfo info =
- RSGroupInfoManager.getRSGroup(RSGroupInfoManager.getRSGroupOfTable(regionInfo.getTable()));
+ RSGroupInfo info = rsGroupInfoManager.getRSGroup(rsGroupInfoManager.
+ getRSGroupOfTable(regionInfo.getTable()));
if (assignedServer != null &&
- (info == null || !info.containsServer(assignedServer.getHostPort()))) {
+ (info == null || !info.containsServer(assignedServer.getAddress()))) {
+ RSGroupInfo otherInfo = null;
+ synchronized (this.rsGroupInfoManager) {
+ otherInfo = rsGroupInfoManager.getRSGroupOfServer(assignedServer.getAddress());
+ }
LOG.debug("Found misplaced region: " + regionInfo.getRegionNameAsString() +
" on server: " + assignedServer +
- " found in group: " +
- RSGroupInfoManager.getRSGroupOfServer(assignedServer.getHostPort()) +
+ " found in group: " + otherInfo +
" outside of group: " + (info == null ? "UNKNOWN" : info.getName()));
misplacedRegions.add(regionInfo);
}
@@ -357,13 +349,13 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
for (HRegionInfo region : regions) {
RSGroupInfo info = null;
try {
- info = RSGroupInfoManager.getRSGroup(
- RSGroupInfoManager.getRSGroupOfTable(region.getTable()));
+ info = rsGroupInfoManager.getRSGroup(
+ rsGroupInfoManager.getRSGroupOfTable(region.getTable()));
} catch (IOException exp) {
- LOG.debug("Group information null for region of table " + region.getTable(),
+ LOG.debug("RSGroup information null for region of table " + region.getTable(),
exp);
}
- if ((info == null) || (!info.containsServer(sName.getHostPort()))) {
+ if ((info == null) || (!info.containsServer(sName.getAddress()))) {
correctAssignments.get(LoadBalancer.BOGUS_SERVER_NAME).add(region);
} else {
correctAssignments.get(sName).add(region);
@@ -382,7 +374,7 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
@Override
public void initialize() throws HBaseIOException {
try {
- if (RSGroupInfoManager == null) {
+ if (rsGroupInfoManager == null) {
List<RSGroupAdminEndpoint> cps =
masterServices.getMasterCoprocessorHost().findCoprocessors(RSGroupAdminEndpoint.class);
if (cps.size() != 1) {
@@ -390,7 +382,7 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
LOG.error(msg);
throw new HBaseIOException(msg);
}
- RSGroupInfoManager = cps.get(0).getGroupInfoManager();
+ rsGroupInfoManager = cps.get(0).getGroupInfoManager();
}
} catch (IOException e) {
throw new HBaseIOException("Failed to initialize GroupInfoManagerImpl", e);
@@ -408,7 +400,8 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc
}
public boolean isOnline() {
- return RSGroupInfoManager != null && RSGroupInfoManager.isOnline();
+ if (this.rsGroupInfoManager == null) return false;
+ return this.rsGroupInfoManager.isOnline();
}
@Override
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java
index ce6699f..d5a971c 100644
--- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java
+++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -20,7 +18,6 @@
package org.apache.hadoop.hbase.rsgroup;
-import com.google.common.net.HostAndPort;
import java.io.IOException;
import java.util.List;
@@ -30,6 +27,7 @@ import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.Address;
/**
* Interface used to manage RSGroupInfo storage. An implementation
@@ -65,14 +63,14 @@ public interface RSGroupInfoManager {
void removeRSGroup(String groupName) throws IOException;
/**
- * move servers to a new group.
- * @param hostPorts list of servers, must be part of the same group
+ * Move servers to a new group.
+ * @param servers list of servers, must be part of the same group
* @param srcGroup groupName being moved from
* @param dstGroup groupName being moved to
- * @return Set of servers moved (May be a subset of {@code hostPorts}).
+ * @return Set of servers moved (May be a subset of {@code servers}).
* @throws java.io.IOException on move failure
*/
- Set<HostAndPort> moveServers(Set<HostAndPort> hostPorts,
+ Set<Address> moveServers(Set<Address> servers,
String srcGroup, String dstGroup) throws IOException;
/**
@@ -81,7 +79,7 @@ public interface RSGroupInfoManager {
* @param hostPort the server
* @return An instance of RSGroupInfo
*/
- RSGroupInfo getRSGroupOfServer(HostAndPort hostPort) throws IOException;
+ RSGroupInfo getRSGroupOfServer(Address hostPort) throws IOException;
/**
* Gets the group information.
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java
index fccf7bf..a89d9ed 100644
--- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java
+++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -20,16 +18,11 @@
package org.apache.hadoop.hbase.rsgroup;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.google.common.net.HostAndPort;
-import com.google.protobuf.ServiceException;
+import static org.apache.hadoop.hbase.rsgroup.Utility.getOnlineServers;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
@@ -37,12 +30,12 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Set;
+import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
@@ -52,9 +45,9 @@ import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MetaTableAccessor;
-import org.apache.hadoop.hbase.ProcedureInfo;
import org.apache.hadoop.hbase.MetaTableAccessor.DefaultVisitorBase;
import org.apache.hadoop.hbase.NamespaceDescriptor;
+import org.apache.hadoop.hbase.ProcedureInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -74,23 +67,40 @@ import org.apache.hadoop.hbase.master.ServerListener;
import org.apache.hadoop.hbase.master.TableStateManager;
import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
-import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos;
import org.apache.hadoop.hbase.protobuf.generated.RSGroupProtos;
import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
import org.apache.hadoop.hbase.security.access.AccessControlLists;
+import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
import org.apache.zookeeper.KeeperException;
-import static org.apache.hadoop.hbase.rsgroup.Utility.getOnlineServers;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.protobuf.ServiceException;
/**
- * This is an implementation of {@link RSGroupInfoManager}. Which makes
+ * This is an implementation of {@link RSGroupInfoManager} which makes
* use of an HBase table as the persistence store for the group information.
* It also makes use of zookeeper to store group information needed
* for bootstrapping during offline mode.
+ *
+ * <h2>Concurrency</h2>
+ * All methods are synchronized to protect against concurrent access on contained
+ * Maps and so as only one writer at a time to the backing zookeeper cache and rsgroup table.
+ *
+ * <p>Clients of this class, the {@link RSGroupAdminEndpoint} for example, want to query and
+ * then act on the results of the query modifying cache in zookeeper without another thread
+ * making intermediate modifications. These clients synchronize on the 'this' instance so
+ * no other has access concurrently.
+ *
+ * TODO: Spend time cleaning up this coarse locking that is prone to error if not carefully
+ * enforced everywhere.
*/
@InterfaceAudience.Private
public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListener {
@@ -111,18 +121,18 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
}
}
- private volatile Map<String, RSGroupInfo> rsGroupMap;
- private volatile Map<TableName, String> tableMap;
- private MasterServices master;
+ private Map<String, RSGroupInfo> rsGroupMap;
+ private Map<TableName, String> tableMap;
+ private final MasterServices master;
private Table rsGroupTable;
- private ClusterConnection conn;
- private ZooKeeperWatcher watcher;
+ private final ClusterConnection conn;
+ private final ZooKeeperWatcher watcher;
private RSGroupStartupWorker rsGroupStartupWorker;
// contains list of groups that were last flushed to persistent store
- private volatile Set<String> prevRSGroups;
- private RSGroupSerDe rsGroupSerDe;
+ private Set<String> prevRSGroups;
+ private final RSGroupSerDe rsGroupSerDe;
private DefaultServerUpdater defaultServerUpdater;
- private boolean isInit = false;
+ private boolean init = false;
public RSGroupInfoManagerImpl(MasterServices master) throws IOException {
@@ -135,18 +145,18 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
prevRSGroups = new HashSet<String>();
}
- public void init() throws IOException{
+ public synchronized void init() throws IOException{
rsGroupStartupWorker = new RSGroupStartupWorker(this, master, conn);
refresh();
rsGroupStartupWorker.start();
defaultServerUpdater = new DefaultServerUpdater(this);
master.getServerManager().registerListener(this);
defaultServerUpdater.start();
- isInit = true;
+ init = true;
}
- boolean isInit() {
- return isInit;
+ synchronized boolean isInit() {
+ return init;
}
/**
@@ -166,43 +176,48 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
flushConfig(newGroupMap);
}
- @Override
- public synchronized Set<HostAndPort> moveServers(Set<HostAndPort> hostPorts,
- String srcGroup, String dstGroup)
- throws IOException {
- if (!rsGroupMap.containsKey(srcGroup)) {
- throw new DoNotRetryIOException("RSGroup " + srcGroup + " does not exist");
+ private RSGroupInfo getRSGroupInfo(final String groupName) throws DoNotRetryIOException {
+ RSGroupInfo rsgi = null;
+ try {
+ rsgi = getRSGroup(groupName);
+ } catch (IOException ioe) {
+ // Will never happen
+ throw new DoNotRetryIOException(ioe);
}
- if (!rsGroupMap.containsKey(dstGroup)) {
- throw new DoNotRetryIOException("RSGroup " + dstGroup + " does not exist");
+ if (rsgi == null) {
+ throw new DoNotRetryIOException("RSGroup " + groupName + " does not exist");
}
- RSGroupInfo src = new RSGroupInfo(getRSGroup(srcGroup));
- RSGroupInfo dst = new RSGroupInfo(getRSGroup(dstGroup));
- // If destination is 'default' rsgroup, make sure servers is online.
- // If not, just drop it.
- Set<HostAndPort> onlineServers = dst.getName().equals(RSGroupInfo.DEFAULT_GROUP)?
+ return rsgi;
+ }
+
+ @Override
+ public synchronized Set<Address> moveServers(Set<Address> servers, String srcGroup, String dstGroup)
+ throws IOException {
+ RSGroupInfo src = getRSGroupInfo(srcGroup);
+ RSGroupInfo dst = getRSGroupInfo(dstGroup);
+ // If destination is 'default' rsgroup, only add servers that are online. If not online, drop it.
+ // If not 'default' group, add server to dst group EVEN IF IT IS NOT online (could be a group
+ // of dead servers that are to come back later).
+ Set<Address> onlineServers = dst.getName().equals(RSGroupInfo.DEFAULT_GROUP)?
getOnlineServers(this.master): null;
- Set<HostAndPort> result = new HashSet<>(hostPorts.size());
- for (HostAndPort el: hostPorts) {
+ for (Address el: servers) {
src.removeServer(el);
if (onlineServers != null) {
- // onlineServers is non-null if 'default' rsgroup.
- // If the server is not online, drop it.
+ // onlineServers is non-null if 'default' rsgroup. If the server is not online, drop it.
if (!onlineServers.contains(el)) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Dropping " + el + " during move-to-default rsgroup because it is not online");
+ LOG.debug("Dropping " + el + " during move-to-default rsgroup because not online");
}
continue;
}
}
dst.addServer(el);
- result.add(el);
}
Map<String,RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
newGroupMap.put(src.getName(), src);
newGroupMap.put(dst.getName(), dst);
flushConfig(newGroupMap);
- return result;
+ return dst.getServers();
}
/**
@@ -212,9 +227,10 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
* @return An instance of GroupInfo.
*/
@Override
- public RSGroupInfo getRSGroupOfServer(HostAndPort hostPort) throws IOException {
+ public synchronized RSGroupInfo getRSGroupOfServer(Address hostPort)
+ throws IOException {
for (RSGroupInfo info : rsGroupMap.values()) {
- if (info.containsServer(hostPort)){
+ if (info.containsServer(hostPort)) {
return info;
}
}
@@ -229,15 +245,14 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
* @return An instance of GroupInfo
*/
@Override
- public RSGroupInfo getRSGroup(String groupName) throws IOException {
- RSGroupInfo RSGroupInfo = rsGroupMap.get(groupName);
- return RSGroupInfo;
+ public synchronized RSGroupInfo getRSGroup(String groupName) throws IOException {
+ return this.rsGroupMap.get(groupName);
}
@Override
- public String getRSGroupOfTable(TableName tableName) throws IOException {
+ public synchronized String getRSGroupOfTable(TableName tableName) throws IOException {
return tableMap.get(tableName);
}
@@ -283,18 +298,17 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
}
@Override
- public List<RSGroupInfo> listRSGroups() throws IOException {
- List<RSGroupInfo> list = Lists.newLinkedList(rsGroupMap.values());
- return list;
+ public synchronized List<RSGroupInfo> listRSGroups() throws IOException {
+ return Lists.newLinkedList(rsGroupMap.values());
}
@Override
- public boolean isOnline() {
+ public synchronized boolean isOnline() {
return rsGroupStartupWorker.isOnline();
}
@Override
- public synchronized void refresh() throws IOException {
+ public void refresh() throws IOException {
refresh(false);
}
@@ -346,10 +360,9 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
// so it overwrites the default group loaded
// from region group table or zk
groupList.add(new RSGroupInfo(RSGroupInfo.DEFAULT_GROUP,
- Sets.newHashSet(getDefaultServers()),
+ Sets.newTreeSet(getDefaultServers()),
orphanTables));
-
// populate the data
HashMap<String, RSGroupInfo> newGroupMap = Maps.newHashMap();
HashMap<TableName, String> newTableMap = Maps.newHashMap();
@@ -395,7 +408,11 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
}
return newTableMap;
}
+ private synchronized void flushConfig() throws IOException {
+ flushConfig(rsGroupMap);
+ }
+ // Called from RSGroupStartupWorker thread so synchronize
private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) throws IOException {
Map<TableName, String> newTableMap;
@@ -457,6 +474,7 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
prevRSGroups.addAll(newGroupMap.keySet());
}
+ // Called by getDefaultServers. Presume it has lock in place.
private List<ServerName> getOnlineRS() throws IOException {
if (master != null) {
return master.getServerManager().getOnlineServersList();
@@ -473,29 +491,31 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
}
}
- private List<HostAndPort> getDefaultServers() throws IOException {
- List<HostAndPort> defaultServers = new LinkedList<HostAndPort>();
- for(ServerName server : getOnlineRS()) {
- HostAndPort hostPort = HostAndPort.fromParts(server.getHostname(), server.getPort());
+ // Called by DefaultServerUpdater. Presume it has lock on this manager when it runs.
+ private SortedSet<Address> getDefaultServers() throws IOException {
+ SortedSet<Address> defaultServers = Sets.newTreeSet();
+ for (ServerName server : getOnlineRS()) {
+ Address hostPort = Address.fromParts(server.getHostname(), server.getPort());
boolean found = false;
- for(RSGroupInfo RSGroupInfo : rsGroupMap.values()) {
+ for(RSGroupInfo RSGroupInfo: listRSGroups()) {
if(!RSGroupInfo.DEFAULT_GROUP.equals(RSGroupInfo.getName()) &&
RSGroupInfo.containsServer(hostPort)) {
found = true;
break;
}
}
- if(!found) {
+ if (!found) {
defaultServers.add(hostPort);
}
}
return defaultServers;
}
- private synchronized void updateDefaultServers(
- Set<HostAndPort> hostPort) throws IOException {
+ // Called by DefaultServerUpdater. Synchronize on this because redoing
+ // the rsGroupMap then writing it out.
+ private synchronized void updateDefaultServers(SortedSet<Address> servers) throws IOException {
RSGroupInfo info = rsGroupMap.get(RSGroupInfo.DEFAULT_GROUP);
- RSGroupInfo newInfo = new RSGroupInfo(info.getName(), hostPort, info.getTables());
+ RSGroupInfo newInfo = new RSGroupInfo(info.getName(), servers, info.getTables());
HashMap<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
newGroupMap.put(newInfo.getName(), newInfo);
flushConfig(newGroupMap);
@@ -503,44 +523,40 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
@Override
public void serverAdded(ServerName serverName) {
+ // #serverChanged is internally synchronized
defaultServerUpdater.serverChanged();
}
@Override
public void serverRemoved(ServerName serverName) {
+ // #serverChanged is internally synchronized
defaultServerUpdater.serverChanged();
}
private static class DefaultServerUpdater extends Thread {
private static final Log LOG = LogFactory.getLog(DefaultServerUpdater.class);
- private RSGroupInfoManagerImpl mgr;
+ private final RSGroupInfoManagerImpl mgr;
private boolean hasChanged = false;
public DefaultServerUpdater(RSGroupInfoManagerImpl mgr) {
+ super("RSGroup.ServerUpdater");
+ setDaemon(true);
this.mgr = mgr;
}
@Override
public void run() {
- List<HostAndPort> prevDefaultServers = new LinkedList<HostAndPort>();
- while(!mgr.master.isAborted() || !mgr.master.isStopped()) {
+ SortedSet<Address> prevDefaultServers = new TreeSet<>();
+ while(isMasterRunning(this.mgr.master)) {
try {
LOG.info("Updating default servers.");
- List<HostAndPort> servers = mgr.getDefaultServers();
- Collections.sort(servers, new Comparator<HostAndPort>() {
- @Override
- public int compare(HostAndPort o1, HostAndPort o2) {
- int diff = o1.getHostText().compareTo(o2.getHostText());
- if (diff != 0) {
- return diff;
- }
- return o1.getPort() - o2.getPort();
+ synchronized (this.mgr) {
+ SortedSet<Address> servers = mgr.getDefaultServers();
+ if (!servers.equals(prevDefaultServers)) {
+ mgr.updateDefaultServers(servers);
+ prevDefaultServers = servers;
+ LOG.info("Updated with servers: "+servers.size());
}
- });
- if(!servers.equals(prevDefaultServers)) {
- mgr.updateDefaultServers(Sets.<HostAndPort>newHashSet(servers));
- prevDefaultServers = servers;
- LOG.info("Updated with servers: "+servers.size());
}
try {
synchronized (this) {
@@ -550,6 +566,7 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
hasChanged = false;
}
} catch (InterruptedException e) {
+ LOG.warn("Interrupted", e);
}
} catch (IOException e) {
LOG.warn("Failed to update default servers", e);
@@ -565,20 +582,16 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
}
}
-
private static class RSGroupStartupWorker extends Thread {
private static final Log LOG = LogFactory.getLog(RSGroupStartupWorker.class);
-
- private Configuration conf;
- private volatile boolean isOnline = false;
- private MasterServices masterServices;
- private RSGroupInfoManagerImpl groupInfoManager;
- private ClusterConnection conn;
+ private volatile boolean online = false;
+ private final MasterServices masterServices;
+ private final RSGroupInfoManagerImpl groupInfoManager;
+ private final ClusterConnection conn;
public RSGroupStartupWorker(RSGroupInfoManagerImpl groupInfoManager,
MasterServices masterServices,
ClusterConnection conn) {
- this.conf = masterServices.getConfiguration();
this.masterServices = masterServices;
this.groupInfoManager = groupInfoManager;
this.conn = conn;
@@ -588,7 +601,7 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
@Override
public void run() {
- if(waitForGroupTableOnline()) {
+ if (waitForGroupTableOnline()) {
LOG.info("GroupBasedLoadBalancer is now online");
}
}
@@ -599,25 +612,21 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
final AtomicBoolean found = new AtomicBoolean(false);
final TableStateManager tsm = masterServices.getTableStateManager();
boolean createSent = false;
- while (!found.get() && isMasterRunning()) {
+ while (!found.get() && isMasterRunning(this.masterServices)) {
foundRegions.clear();
assignedRegions.clear();
found.set(true);
try {
- final Table nsTable = conn.getTable(TableName.NAMESPACE_TABLE_NAME);
- final Table groupTable = conn.getTable(RSGROUP_TABLE_NAME);
+ conn.getTable(TableName.NAMESPACE_TABLE_NAME);
+ conn.getTable(RSGROUP_TABLE_NAME);
boolean rootMetaFound =
masterServices.getMetaTableLocator().verifyMetaRegionLocation(
- conn,
- masterServices.getZooKeeper(),
- 1);
+ conn, masterServices.getZooKeeper(), 1);
final AtomicBoolean nsFound = new AtomicBoolean(false);
if (rootMetaFound) {
-
MetaTableAccessor.Visitor visitor = new DefaultVisitorBase() {
@Override
public boolean visitInternal(Result row) throws IOException {
-
HRegionInfo info = MetaTableAccessor.getHRegionInfo(row);
if (info != null) {
Cell serverCell =
@@ -670,12 +679,12 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
MetaTableAccessor.fullScanRegions(conn, visitor);
// if no regions in meta then we have to create the table
if (foundRegions.size() < 1 && rootMetaFound && !createSent && nsFound.get()) {
- groupInfoManager.createGroupTable(masterServices);
+ groupInfoManager.createRSGroupTable(masterServices);
createSent = true;
}
- LOG.info("Group table: " + RSGROUP_TABLE_NAME + " isOnline: " + found.get()
- + ", regionCount: " + foundRegions.size() + ", assignCount: "
- + assignedRegions.size() + ", rootMetaFound: "+rootMetaFound);
+ LOG.info("RSGroup table=" + RSGROUP_TABLE_NAME + " isOnline=" + found.get()
+ + ", regionCount=" + foundRegions.size() + ", assignCount="
+ + assignedRegions.size() + ", rootMetaFound=" + rootMetaFound);
found.set(found.get() && assignedRegions.size() == foundRegions.size()
&& foundRegions.size() > 0);
} else {
@@ -685,9 +694,9 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
if (found.get()) {
LOG.debug("With group table online, refreshing cached information.");
groupInfoManager.refresh(true);
- isOnline = true;
+ online = true;
//flush any inconsistencies between ZK and HTable
- groupInfoManager.flushConfig(groupInfoManager.rsGroupMap);
+ groupInfoManager.flushConfig();
}
} catch (RuntimeException e) {
throw e;
@@ -705,15 +714,15 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
}
public boolean isOnline() {
- return isOnline;
+ return online;
}
+ }
- private boolean isMasterRunning() {
- return !masterServices.isAborted() && !masterServices.isStopped();
- }
+ private static boolean isMasterRunning(MasterServices masterServices) {
+ return !masterServices.isAborted() && !masterServices.isStopped();
}
- private void createGroupTable(MasterServices masterServices) throws IOException {
+ private void createRSGroupTable(MasterServices masterServices) throws IOException {
Long procId = masterServices.createSystemTable(RSGROUP_TABLE_DESC);
// wait for region to be online
int tries = 600;
@@ -772,4 +781,4 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene
throw new ConstraintException("Group name should only contain alphanumeric characters");
}
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupSerDe.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupSerDe.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupSerDe.java
index dfc566e..3cea939 100644
--- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupSerDe.java
+++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupSerDe.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -21,7 +19,6 @@
package org.apache.hadoop.hbase.rsgroup;
import com.google.common.collect.Lists;
-import com.google.common.net.HostAndPort;
import java.io.ByteArrayInputStream;
import java.io.IOException;
@@ -30,15 +27,16 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.generated.RSGroupProtos;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
import org.apache.zookeeper.KeeperException;
@@ -94,7 +92,7 @@ public class RSGroupSerDe {
public static RSGroupInfo toGroupInfo(RSGroupProtos.RSGroupInfo proto) {
RSGroupInfo RSGroupInfo = new RSGroupInfo(proto.getName());
for(HBaseProtos.ServerName el: proto.getServersList()) {
- RSGroupInfo.addServer(HostAndPort.fromParts(el.getHostName(), el.getPort()));
+ RSGroupInfo.addServer(Address.fromParts(el.getHostName(), el.getPort()));
}
for(HBaseProtos.TableName pTableName: proto.getTablesList()) {
RSGroupInfo.addTable(ProtobufUtil.toTableName(pTableName));
@@ -110,9 +108,9 @@ public class RSGroupSerDe {
}
List<HBaseProtos.ServerName> hostports =
new ArrayList<HBaseProtos.ServerName>(pojo.getServers().size());
- for(HostAndPort el: pojo.getServers()) {
+ for(Address el: pojo.getServers()) {
hostports.add(HBaseProtos.ServerName.newBuilder()
- .setHostName(el.getHostText())
+ .setHostName(el.getHostname())
.setPort(el.getPort())
.build());
}
@@ -120,4 +118,4 @@ public class RSGroupSerDe {
.addAllServers(hostports)
.addAllTables(tables).build();
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupableBalancer.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupableBalancer.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupableBalancer.java
index ec86dda..bff392b 100644
--- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupableBalancer.java
+++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupableBalancer.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -23,7 +21,9 @@ package org.apache.hadoop.hbase.rsgroup;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.master.LoadBalancer;
-
+/**
+ * Marker Interface. RSGroups feature will check for a LoadBalancer
+ * marked with this Interface before it runs.
+ */
@InterfaceAudience.Private
-public interface RSGroupableBalancer extends LoadBalancer {
-}
+public interface RSGroupableBalancer extends LoadBalancer {}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/Utility.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/Utility.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/Utility.java
index d9c7532..95ad05e 100644
--- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/Utility.java
+++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/Utility.java
@@ -25,8 +25,7 @@ import java.util.Set;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.master.MasterServices;
-
-import com.google.common.net.HostAndPort;
+import org.apache.hadoop.hbase.util.Address;
/**
* Utility for this RSGroup package in hbase-rsgroup.
@@ -37,11 +36,11 @@ class Utility {
* @param master
* @return Set of online Servers named for their hostname and port (not ServerName).
*/
- static Set<HostAndPort> getOnlineServers(final MasterServices master) {
- Set<HostAndPort> onlineServers = new HashSet<HostAndPort>();
+ static Set<Address> getOnlineServers(final MasterServices master) {
+ Set<Address> onlineServers = new HashSet<Address>();
if (master == null) return onlineServers;
for(ServerName server: master.getServerManager().getOnlineServers().keySet()) {
- onlineServers.add(server.getHostPort());
+ onlineServers.add(server.getAddress());
}
return onlineServers;
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/master/balancer/TestRSGroupBasedLoadBalancer.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/master/balancer/TestRSGroupBasedLoadBalancer.java b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/master/balancer/TestRSGroupBasedLoadBalancer.java
index 1539f73..73cf85a 100644
--- a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/master/balancer/TestRSGroupBasedLoadBalancer.java
+++ b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/master/balancer/TestRSGroupBasedLoadBalancer.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -21,7 +19,6 @@ package org.apache.hadoop.hbase.master.balancer;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
-import com.google.common.net.HostAndPort;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -40,6 +37,7 @@ import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.master.MasterServices;
import org.apache.hadoop.hbase.master.RegionPlan;
import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -176,12 +174,11 @@ public class TestRSGroupBasedLoadBalancer {
ServerName server = assignments.get(region);
TableName tableName = region.getTable();
- String groupName =
- getMockedGroupInfoManager().getRSGroupOfTable(tableName);
+ String groupName = getMockedGroupInfoManager().getRSGroupOfTable(tableName);
assertTrue(StringUtils.isNotEmpty(groupName));
RSGroupInfo gInfo = getMockedGroupInfoManager().getRSGroup(groupName);
assertTrue("Region is not correctly assigned to group servers.",
- gInfo.containsServer(server.getHostPort()));
+ gInfo.containsServer(server.getAddress()));
}
}
@@ -216,7 +213,7 @@ public class TestRSGroupBasedLoadBalancer {
groupName);
assertTrue(
"Region is not correctly assigned to group servers.",
- gInfo.containsServer(sn.getHostPort()));
+ gInfo.containsServer(sn.getAddress()));
}
}
ArrayListMultimap<String, ServerAndLoad> loadMap = convertToGroupBasedMap(assignments);
@@ -295,15 +292,15 @@ public class TestRSGroupBasedLoadBalancer {
groupName);
assertTrue(
"Region is not correctly assigned to group servers.",
- gInfo.containsServer(currentServer.getHostPort()));
+ gInfo.containsServer(currentServer.getAddress()));
if (oldAssignedServer != null
&& onlineHostNames.contains(oldAssignedServer
.getHostname())) {
// this region was previously assigned somewhere, and that
// host is still around, then the host must have been is a
// different group.
- if (!oldAssignedServer.getHostPort().equals(currentServer.getHostPort())) {
- assertFalse(gInfo.containsServer(oldAssignedServer.getHostPort()));
+ if (!oldAssignedServer.getAddress().equals(currentServer.getAddress())) {
+ assertFalse(gInfo.containsServer(oldAssignedServer.getAddress()));
}
}
}
@@ -346,11 +343,11 @@ public class TestRSGroupBasedLoadBalancer {
ArrayListMultimap<String, ServerAndLoad> loadMap = ArrayListMultimap
.create();
for (RSGroupInfo gInfo : getMockedGroupInfoManager().listRSGroups()) {
- Set<HostAndPort> groupServers = gInfo.getServers();
- for (HostAndPort hostPort : groupServers) {
+ Set<Address> groupServers = gInfo.getServers();
+ for (Address hostPort : groupServers) {
ServerName actual = null;
for(ServerName entry: servers) {
- if(entry.getHostPort().equals(hostPort)) {
+ if(entry.getAddress().equals(hostPort)) {
actual = entry;
break;
}
@@ -486,14 +483,14 @@ public class TestRSGroupBasedLoadBalancer {
Map<String, RSGroupInfo> groupMap = new HashMap<String, RSGroupInfo>();
for (String grpName : groups) {
RSGroupInfo RSGroupInfo = new RSGroupInfo(grpName);
- RSGroupInfo.addServer(servers.get(index).getHostPort());
+ RSGroupInfo.addServer(servers.get(index).getAddress());
groupMap.put(grpName, RSGroupInfo);
index++;
}
while (index < servers.size()) {
int grpIndex = rand.nextInt(groups.length);
groupMap.get(groups[grpIndex]).addServer(
- servers.get(index).getHostPort());
+ servers.get(index).getAddress());
index++;
}
return groupMap;
@@ -558,7 +555,7 @@ public class TestRSGroupBasedLoadBalancer {
RSGroupInfoManager gm = getMockedGroupInfoManager();
RSGroupInfo groupOfServer = null;
for(RSGroupInfo gInfo : gm.listRSGroups()){
- if(gInfo.containsServer(sn.getHostPort())){
+ if(gInfo.containsServer(sn.getAddress())){
groupOfServer = gInfo;
break;
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroups.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroups.java b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroups.java
index 957994f..94a76a8 100644
--- a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroups.java
+++ b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroups.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -19,9 +17,13 @@
*/
package org.apache.hadoop.hbase.rsgroup;
-import com.google.common.collect.Sets;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.Iterator;
-import com.google.common.net.HostAndPort;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HBaseTestingUtility;
@@ -41,6 +43,7 @@ import org.apache.hadoop.hbase.master.ServerManager;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.AfterClass;
@@ -50,18 +53,13 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import java.io.IOException;
-import java.util.Iterator;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.google.common.collect.Sets;
@Category({MediumTests.class})
public class TestRSGroups extends TestRSGroupsBase {
protected static final Log LOG = LogFactory.getLog(TestRSGroups.class);
private static HMaster master;
- private static boolean init = false;
+ private static boolean INIT = false;
private static RSGroupAdminEndpoint RSGroupAdminEndpoint;
@@ -93,8 +91,8 @@ public class TestRSGroups extends TestRSGroupsBase {
}
});
admin.setBalancerRunning(false,true);
- rsGroupAdmin = new VerifyingRSGroupAdminClient(rsGroupAdmin.newClient(TEST_UTIL.getConnection()),
- TEST_UTIL.getConfiguration());
+ rsGroupAdmin = new VerifyingRSGroupAdminClient(
+ new RSGroupAdminClient(TEST_UTIL.getConnection()), TEST_UTIL.getConfiguration());
RSGroupAdminEndpoint =
master.getMasterCoprocessorHost().findCoprocessors(RSGroupAdminEndpoint.class).get(0);
}
@@ -106,8 +104,8 @@ public class TestRSGroups extends TestRSGroupsBase {
@Before
public void beforeMethod() throws Exception {
- if(!init) {
- init = true;
+ if (!INIT) {
+ INIT = true;
afterMethod();
}
@@ -130,11 +128,9 @@ public class TestRSGroups extends TestRSGroupsBase {
((MiniHBaseCluster)cluster).getMaster().getServerName();
try {
- rsGroupAdmin.moveServers(
- Sets.newHashSet(masterServerName.getHostPort()),
- "master");
+ rsGroupAdmin.moveServers(Sets.newHashSet(masterServerName.getAddress()), "master");
} catch (Exception ex) {
- // ignore
+ LOG.warn("Got this on setup, FYI", ex);
}
TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
@Override
@@ -241,10 +237,13 @@ public class TestRSGroups extends TestRSGroupsBase {
@Test
public void testGroupInfoMultiAccessing() throws Exception {
RSGroupInfoManager manager = RSGroupAdminEndpoint.getGroupInfoManager();
- final RSGroupInfo defaultGroup = manager.getRSGroup("default");
+ RSGroupInfo defaultGroup = null;
+ synchronized (manager) {
+ defaultGroup = manager.getRSGroup("default");
+ }
// getRSGroup updates default group's server list
// this process must not affect other threads iterating the list
- Iterator<HostAndPort> it = defaultGroup.getServers().iterator();
+ Iterator<Address> it = defaultGroup.getServers().iterator();
manager.getRSGroup("default");
it.next();
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsBase.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsBase.java b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsBase.java
index 991f6e1..9ad53e4 100644
--- a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsBase.java
+++ b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsBase.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -19,9 +17,21 @@
*/
package org.apache.hadoop.hbase.rsgroup;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.google.common.net.HostAndPort;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.security.SecureRandom;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.ClusterStatus;
@@ -40,23 +50,13 @@ import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.constraint.ConstraintException;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Assert;
import org.junit.Test;
-import java.io.IOException;
-import java.security.SecureRandom;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
public abstract class TestRSGroupsBase {
protected static final Log LOG = LogFactory.getLog(TestRSGroupsBase.class);
@@ -85,8 +85,8 @@ public abstract class TestRSGroupsBase {
assertTrue(defaultInfo.getServers().size() >= serverCount);
gAdmin.addRSGroup(groupName);
- Set<HostAndPort> set = new HashSet<HostAndPort>();
- for(HostAndPort server: defaultInfo.getServers()) {
+ Set<Address> set = new HashSet<Address>();
+ for(Address server: defaultInfo.getServers()) {
if(set.size() == serverCount) {
break;
}
@@ -120,7 +120,8 @@ public abstract class TestRSGroupsBase {
}
protected void deleteGroups() throws IOException {
- RSGroupAdmin groupAdmin = rsGroupAdmin.newClient(TEST_UTIL.getConnection());
+ RSGroupAdmin groupAdmin =
+ new RSGroupAdminClient(TEST_UTIL.getConnection());
for(RSGroupInfo group: groupAdmin.listRSGroups()) {
if(!group.getName().equals(RSGroupInfo.DEFAULT_GROUP)) {
groupAdmin.moveTables(group.getTables(), RSGroupInfo.DEFAULT_GROUP);
@@ -167,7 +168,7 @@ public abstract class TestRSGroupsBase {
@Test
public void testBogusArgs() throws Exception {
assertNull(rsGroupAdmin.getRSGroupInfoOfTable(TableName.valueOf("nonexistent")));
- assertNull(rsGroupAdmin.getRSGroupOfServer(HostAndPort.fromParts("bogus",123)));
+ assertNull(rsGroupAdmin.getRSGroupOfServer(Address.fromParts("bogus",123)));
assertNull(rsGroupAdmin.getRSGroupInfo("bogus"));
try {
@@ -185,7 +186,7 @@ public abstract class TestRSGroupsBase {
}
try {
- rsGroupAdmin.moveServers(Sets.newHashSet(HostAndPort.fromParts("bogus",123)), "bogus");
+ rsGroupAdmin.moveServers(Sets.newHashSet(Address.fromParts("bogus",123)), "bogus");
fail("Expected move with bogus group to fail");
} catch(ConstraintException ex) {
//expected
@@ -276,10 +277,10 @@ public abstract class TestRSGroupsBase {
//test fail bogus server move
try {
- rsGroupAdmin.moveServers(Sets.newHashSet(HostAndPort.fromString("foo:9999")),"foo");
+ rsGroupAdmin.moveServers(Sets.newHashSet(Address.fromString("foo:9999")),"foo");
fail("Bogus servers shouldn't have been successfully moved.");
} catch(IOException ex) {
- String exp = "Server foo:9999 does not have a rsgroup";
+ String exp = "Source RSGroup for server foo:9999 does not exist.";
String msg = "Expected '"+exp+"' in exception message: ";
assertTrue(msg+" "+ex.getMessage(), ex.getMessage().contains(exp));
}
@@ -354,7 +355,7 @@ public abstract class TestRSGroupsBase {
int count = 0;
if (serverMap != null) {
for (ServerName rs : serverMap.keySet()) {
- if (newGroup.containsServer(rs.getHostPort())) {
+ if (newGroup.containsServer(rs.getAddress())) {
count += serverMap.get(rs).size();
}
}
@@ -476,7 +477,7 @@ public abstract class TestRSGroupsBase {
//get server which is not a member of new group
ServerName targetServer = null;
for(ServerName server : admin.getClusterStatus().getServers()) {
- if(!newGroup.containsServer(server.getHostPort())) {
+ if(!newGroup.containsServer(server.getAddress())) {
targetServer = server;
break;
}
@@ -486,7 +487,7 @@ public abstract class TestRSGroupsBase {
((ClusterConnection) admin.getConnection()).getAdmin(targetServer);
//move target server to group
- rsGroupAdmin.moveServers(Sets.newHashSet(targetServer.getHostPort()),
+ rsGroupAdmin.moveServers(Sets.newHashSet(targetServer.getAddress()),
newGroup.getName());
TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
@Override
@@ -592,7 +593,7 @@ public abstract class TestRSGroupsBase {
return cluster.getClusterStatus().getRegionsInTransition().isEmpty();
}
});
- Set<HostAndPort> newServers = Sets.newHashSet();
+ Set<Address> newServers = Sets.newHashSet();
newServers.add(
rsGroupAdmin.getRSGroupInfo(RSGroupInfo.DEFAULT_GROUP).getServers().iterator().next());
rsGroupAdmin.moveServers(newServers, appInfo.getName());
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsOfflineMode.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsOfflineMode.java b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsOfflineMode.java
index 4a8656c8..a9157b6 100644
--- a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsOfflineMode.java
+++ b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsOfflineMode.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -105,50 +103,47 @@ public class TestRSGroupsOfflineMode {
//so it gets assigned later
final TableName failoverTable = TableName.valueOf(name.getMethodName());
TEST_UTIL.createTable(failoverTable, Bytes.toBytes("f"));
-
- RSGroupAdmin groupAdmin = RSGroupAdmin.newClient(TEST_UTIL.getConnection());
-
final HRegionServer killRS = ((MiniHBaseCluster)cluster).getRegionServer(0);
final HRegionServer groupRS = ((MiniHBaseCluster)cluster).getRegionServer(1);
final HRegionServer failoverRS = ((MiniHBaseCluster)cluster).getRegionServer(2);
-
String newGroup = "my_group";
+ RSGroupAdmin groupAdmin = new RSGroupAdminClient(TEST_UTIL.getConnection());
groupAdmin.addRSGroup(newGroup);
if(master.getAssignmentManager().getRegionStates().getRegionAssignments()
- .containsValue(failoverRS.getServerName())) {
- for(HRegionInfo regionInfo: hbaseAdmin.getOnlineRegions(failoverRS.getServerName())) {
- hbaseAdmin.move(regionInfo.getEncodedNameAsBytes(),
- Bytes.toBytes(failoverRS.getServerName().getServerName()));
- }
- LOG.info("Waiting for region unassignments on failover RS...");
+ .containsValue(failoverRS.getServerName())) {
+ for(HRegionInfo regionInfo: hbaseAdmin.getOnlineRegions(failoverRS.getServerName())) {
+ hbaseAdmin.move(regionInfo.getEncodedNameAsBytes(),
+ Bytes.toBytes(failoverRS.getServerName().getServerName()));
+ }
+ LOG.info("Waiting for region unassignments on failover RS...");
+ TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
+ @Override
+ public boolean evaluate() throws Exception {
+ return master.getServerManager().getLoad(failoverRS.getServerName())
+ .getRegionsLoad().size() > 0;
+ }
+ });
+
+ //move server to group and make sure all tables are assigned
+ groupAdmin.moveServers(Sets.newHashSet(groupRS.getServerName().getAddress()), newGroup);
TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
@Override
public boolean evaluate() throws Exception {
- return master.getServerManager().getLoad(failoverRS.getServerName())
- .getRegionsLoad().size() > 0;
+ return groupRS.getNumberOfOnlineRegions() < 1 &&
+ master.getAssignmentManager().getRegionStates().getRegionsInTransition().size() < 1;
+ }
+ });
+ //move table to group and wait
+ groupAdmin.moveTables(Sets.newHashSet(RSGroupInfoManager.RSGROUP_TABLE_NAME), newGroup);
+ LOG.info("Waiting for move table...");
+ TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
+ @Override
+ public boolean evaluate() throws Exception {
+ return groupRS.getNumberOfOnlineRegions() == 1;
}
});
}
- //move server to group and make sure all tables are assigned
- groupAdmin.moveServers(Sets.newHashSet(groupRS.getServerName().getHostPort()), newGroup);
- TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
- @Override
- public boolean evaluate() throws Exception {
- return groupRS.getNumberOfOnlineRegions() < 1 &&
- master.getAssignmentManager().getRegionStates().getRegionsInTransition().size() < 1;
- }
- });
- //move table to group and wait
- groupAdmin.moveTables(Sets.newHashSet(RSGroupInfoManager.RSGROUP_TABLE_NAME), newGroup);
- LOG.info("Waiting for move table...");
- TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
- @Override
- public boolean evaluate() throws Exception {
- return groupRS.getNumberOfOnlineRegions() == 1;
- }
- });
-
groupRS.stop("die");
//race condition here
TEST_UTIL.getHBaseCluster().getMaster().stopMaster();
@@ -167,12 +162,13 @@ public class TestRSGroupsOfflineMode {
RSGroupInfoManager groupMgr = RSGroupAdminEndpoint.getGroupInfoManager();
//make sure balancer is in offline mode, since this is what we're testing
- assertFalse(groupMgr.isOnline());
- //verify the group affiliation that's loaded from ZK instead of tables
- assertEquals(newGroup,
+ synchronized (groupMgr) {
+ assertFalse(groupMgr.isOnline());
+ //verify the group affiliation that's loaded from ZK instead of tables
+ assertEquals(newGroup,
groupMgr.getRSGroupOfTable(RSGroupInfoManager.RSGROUP_TABLE_NAME));
- assertEquals(RSGroupInfo.DEFAULT_GROUP, groupMgr.getRSGroupOfTable(failoverTable));
-
+ assertEquals(RSGroupInfo.DEFAULT_GROUP, groupMgr.getRSGroupOfTable(failoverTable));
+ }
//kill final regionserver to see the failover happens for all tables
//except GROUP table since it's group does not have any online RS
killRS.stop("die");
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdminClient.java
----------------------------------------------------------------------
diff --git a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdminClient.java b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdminClient.java
index 441cc39..0732d7c 100644
--- a/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdminClient.java
+++ b/hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdminClient.java
@@ -1,6 +1,4 @@
/**
- * Copyright The Apache Software Foundation
- *
* 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
@@ -21,7 +19,6 @@ package org.apache.hadoop.hbase.rsgroup;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.net.HostAndPort;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -32,6 +29,7 @@ import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.generated.RSGroupProtos;
+import org.apache.hadoop.hbase.util.Address;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
import org.apache.zookeeper.KeeperException;
@@ -44,10 +42,9 @@ import java.util.Map;
import java.util.Set;
@InterfaceAudience.Private
-public class VerifyingRSGroupAdminClient extends RSGroupAdmin {
+public class VerifyingRSGroupAdminClient implements RSGroupAdmin {
private Table table;
private ZooKeeperWatcher zkw;
- private RSGroupSerDe serDe;
private RSGroupAdmin wrapped;
public VerifyingRSGroupAdminClient(RSGroupAdmin RSGroupAdmin, Configuration conf)
@@ -55,7 +52,6 @@ public class VerifyingRSGroupAdminClient extends RSGroupAdmin {
wrapped = RSGroupAdmin;
table = ConnectionFactory.createConnection(conf).getTable(RSGroupInfoManager.RSGROUP_TABLE_NAME);
zkw = new ZooKeeperWatcher(conf, this.getClass().getSimpleName(), null);
- serDe = new RSGroupSerDe();
}
@Override
@@ -75,7 +71,7 @@ public class VerifyingRSGroupAdminClient extends RSGroupAdmin {
}
@Override
- public void moveServers(Set<HostAndPort> servers, String targetGroup) throws IOException {
+ public void moveServers(Set<Address> servers, String targetGroup) throws IOException {
wrapped.moveServers(servers, targetGroup);
verify();
}
@@ -103,7 +99,7 @@ public class VerifyingRSGroupAdminClient extends RSGroupAdmin {
}
@Override
- public RSGroupInfo getRSGroupOfServer(HostAndPort hostPort) throws IOException {
+ public RSGroupInfo getRSGroupOfServer(Address hostPort) throws IOException {
return wrapped.getRSGroupOfServer(hostPort);
}
@@ -144,8 +140,4 @@ public class VerifyingRSGroupAdminClient extends RSGroupAdmin {
throw new IOException("ZK verification failed", e);
}
}
-
- @Override
- public void close() throws IOException {
- }
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterAndRegionObserver.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterAndRegionObserver.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterAndRegionObserver.java
index a0a1d49..8cb41b6 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterAndRegionObserver.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterAndRegionObserver.java
@@ -19,7 +19,6 @@
package org.apache.hadoop.hbase.coprocessor;
-import com.google.common.net.HostAndPort;
import java.io.IOException;
import java.util.List;
@@ -44,6 +43,7 @@ import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.SnapshotDescription;
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
+import org.apache.hadoop.hbase.util.Address;
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
@InterfaceStability.Evolving
@@ -782,7 +782,7 @@ public class BaseMasterAndRegionObserver extends BaseRegionObserver
}
@Override
- public void postMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx, Set<HostAndPort>
+ public void postMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx, Set<Address>
servers, String targetGroup) throws IOException {
}
@@ -808,7 +808,7 @@ public class BaseMasterAndRegionObserver extends BaseRegionObserver
@Override
public void preMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx,
- Set<HostAndPort> servers, String targetGroup) throws IOException {
+ Set<Address> servers, String targetGroup) throws IOException {
}
@Override
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterObserver.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterObserver.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterObserver.java
index 98f21b2..6e9a2ad 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterObserver.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseMasterObserver.java
@@ -19,7 +19,6 @@
package org.apache.hadoop.hbase.coprocessor;
-import com.google.common.net.HostAndPort;
import java.io.IOException;
import java.util.List;
@@ -44,6 +43,7 @@ import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.SnapshotDescription;
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
+import org.apache.hadoop.hbase.util.Address;
@InterfaceAudience.LimitedPrivate({HBaseInterfaceAudience.COPROC, HBaseInterfaceAudience.CONFIG})
@InterfaceStability.Evolving
@@ -1113,12 +1113,12 @@ public class BaseMasterObserver implements MasterObserver {
}
@Override
- public void preMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx, Set<HostAndPort>
+ public void preMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx, Set<Address>
servers, String targetGroup) throws IOException {
}
@Override
- public void postMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx, Set<HostAndPort>
+ public void postMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx, Set<Address>
servers, String targetGroup) throws IOException {
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java
index bb0e732..1d0c467d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java
@@ -44,8 +44,8 @@ import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.SnapshotDescription;
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
+import org.apache.hadoop.hbase.util.Address;
-import com.google.common.net.HostAndPort;
/**
* Defines coprocessor hooks for interacting with operations on the
@@ -1750,7 +1750,7 @@ public interface MasterObserver extends Coprocessor {
* @throws IOException on failure
*/
void preMoveServers(final ObserverContext<MasterCoprocessorEnvironment> ctx,
- Set<HostAndPort> servers, String targetGroup) throws IOException;
+ Set<Address> servers, String targetGroup) throws IOException;
/**
* Called after servers are moved to target region server group
@@ -1760,7 +1760,7 @@ public interface MasterObserver extends Coprocessor {
* @throws IOException on failure
*/
void postMoveServers(final ObserverContext<MasterCoprocessorEnvironment> ctx,
- Set<HostAndPort> servers, String targetGroup) throws IOException;
+ Set<Address> servers, String targetGroup) throws IOException;
/**
* Called before tables are moved to target region server group
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java
index 1472a91..277dcc8 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java
@@ -21,9 +21,6 @@ package org.apache.hadoop.hbase.master;
import java.util.List;
import java.util.Map;
-import edu.umd.cs.findbugs.annotations.Nullable;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.conf.ConfigurationObserver;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClusterStatus;
@@ -32,6 +29,10 @@ import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.conf.ConfigurationObserver;
+
+import edu.umd.cs.findbugs.annotations.Nullable;
/**
* Makes decisions about the placement and movement of Regions across
@@ -48,8 +49,8 @@ import org.apache.hadoop.hbase.TableName;
@InterfaceAudience.Private
public interface LoadBalancer extends Configurable, Stoppable, ConfigurationObserver {
- //used to signal to the caller that the region(s) cannot be assigned
- ServerName BOGUS_SERVER_NAME = ServerName.parseServerName("localhost,1,1");
+ // Used to signal to the caller that the region(s) cannot be assigned
+ static final ServerName BOGUS_SERVER_NAME = ServerName.valueOf("bogus.example.com,1,1");
/**
* Set the current cluster status. This allows a LoadBalancer to map host name to a server
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java
index 28a8550..70d749e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterCoprocessorHost.java
@@ -54,8 +54,7 @@ import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.SnapshotDescription;
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
-
-import com.google.common.net.HostAndPort;
+import org.apache.hadoop.hbase.util.Address;
/**
* Provides the coprocessor framework and environment for master oriented
@@ -1512,7 +1511,7 @@ public class MasterCoprocessorHost
return bypass;
}
- public void preMoveServers(final Set<HostAndPort> servers, final String targetGroup)
+ public void preMoveServers(final Set<Address> servers, final String targetGroup)
throws IOException {
execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
@Override
@@ -1525,7 +1524,7 @@ public class MasterCoprocessorHost
});
}
- public void postMoveServers(final Set<HostAndPort> servers, final String targetGroup)
+ public void postMoveServers(final Set<Address> servers, final String targetGroup)
throws IOException {
execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
@Override
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java
index d4a390a..5893323 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java
@@ -17,11 +17,6 @@
*/
package org.apache.hadoop.hbase.master.balancer;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Joiner;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -39,6 +34,7 @@ import java.util.Random;
import java.util.Set;
import java.util.TreeMap;
import java.util.function.Predicate;
+
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -47,7 +43,6 @@ import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.HBaseIOException;
import org.apache.hadoop.hbase.HDFSBlocksDistribution;
import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.RegionLoad;
import org.apache.hadoop.hbase.ServerLoad;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
@@ -60,6 +55,12 @@ import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer.Cluster.Action.T
import org.apache.hadoop.hbase.security.access.AccessControlLists;
import org.apache.hadoop.util.StringUtils;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Joiner;
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
/**
* The base class for load balancers. It provides the the functions used to by
* {@link org.apache.hadoop.hbase.master.AssignmentManager} to assign regions
@@ -1348,7 +1349,7 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
int numServers = servers == null ? 0 : servers.size();
if (numServers == 0) {
- LOG.warn("Wanted to do retain assignment but no servers to assign to");
+ LOG.warn("Wanted to retain assignment but no servers to assign to");
return null;
}
if (numServers == 1) { // Only one server, nothing fancy we can do here
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
index f68ae94..aeffe1f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java
@@ -120,6 +120,7 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.SimpleMutableByteRange;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.hadoop.hbase.util.Address;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableSet;
@@ -128,7 +129,6 @@ import com.google.common.collect.Lists;
import com.google.common.collect.MapMaker;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.net.HostAndPort;
import com.google.protobuf.Message;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
@@ -2671,7 +2671,7 @@ public class AccessController extends BaseMasterAndRegionObserver
@Override
public void preMoveServers(ObserverContext<MasterCoprocessorEnvironment> ctx,
- Set<HostAndPort> servers, String targetGroup) throws IOException {
+ Set<Address> servers, String targetGroup) throws IOException {
requirePermission(getActiveUser(ctx), "moveServers", Action.ADMIN);
}
@@ -2773,4 +2773,4 @@ public class AccessController extends BaseMasterAndRegionObserver
throw new DoNotRetryIOException("Invalid lock level when requesting permissions.");
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
index ea7c07b..35a1176 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
@@ -255,7 +255,8 @@ public abstract class AbstractFSWALProvider<T extends AbstractFSWAL<?>> implemen
}
/**
- * Construct the directory name for all WALs on a given server.
+ * Construct the directory name for all WALs on a given server. Dir names currently look like
+ * this for WALs: <code>hbase//WALs/kalashnikov.att.net,61634,1486865297088</code>.
* @param serverName Server name formatted as described in {@link ServerName}
* @return the relative WAL directory name, e.g. <code>.logs/1.example.org,60030,12345</code> if
* <code>serverName</code> passed is <code>1.example.org,60030,12345</code>
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
index 19ab955..2cf2c6b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
@@ -309,6 +309,8 @@ public class WALSplitter {
outputSinkStarted = true;
Entry entry;
Long lastFlushedSequenceId = -1L;
+ // THIS IS BROKEN!!!! GETTING SERVERNAME FROM PATH IS NOT GOING TO WORK IF LAYOUT CHANGES!!!
+ // TODO: Fix.
ServerName serverName = AbstractFSWALProvider.getServerNameFromWALDirectoryName(logPath);
failedServerName = (serverName == null) ? "" : serverName.getServerName();
while ((entry = getNextLogLine(in, logPath, skipErrors)) != null) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/test/java/org/apache/hadoop/hbase/TestServerName.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestServerName.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestServerName.java
index cde26e6..d1b2145 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestServerName.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestServerName.java
@@ -91,24 +91,15 @@ public class TestServerName {
assertEquals(sn.hashCode(), sn2.hashCode());
assertNotSame(sn.hashCode(), sn3.hashCode());
assertEquals(sn.toString(),
- ServerName.getServerName("www.example.org", 1234, 5678));
+ ServerName.valueOf("www.example.org", 1234, 5678).toString());
assertEquals(sn.toString(),
- ServerName.getServerName("www.example.org:1234", 5678));
+ ServerName.valueOf("www.example.org:1234", 5678).toString());
assertEquals(sn.toString(),
"www.example.org" + ServerName.SERVERNAME_SEPARATOR + "1234" +
ServerName.SERVERNAME_SEPARATOR + "5678");
}
@Test
- public void getServerStartcodeFromServerName() {
- ServerName sn = ServerName.valueOf("www.example.org", 1234, 5678);
- assertEquals(5678,
- ServerName.getServerStartcodeFromServerName(sn.toString()));
- assertNotSame(5677,
- ServerName.getServerStartcodeFromServerName(sn.toString()));
- }
-
- @Test
public void testHostNameCaseSensitivity() {
ServerName lower = ServerName.valueOf("www.example.org", 1234, 5678);
ServerName upper = ServerName.valueOf("www.EXAMPLE.org", 1234, 5678);
http://git-wip-us.apache.org/repos/asf/hbase/blob/e0199611/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java
index 7633fce..eb15d91 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java
@@ -767,16 +767,6 @@ public class TestAdmin2 {
}
/*
- * Test that invalid draining server names (invalid start code) don't get added to drain list.
- */
- @Test(timeout = 10000, expected = IllegalArgumentException.class)
- public void testCheckDrainServerName() throws Exception {
- List<ServerName> servers = new ArrayList<ServerName>();
- servers.add(ServerName.parseServerName("127.0.0.1:123"));
- admin.drainRegionServers(servers);
- }
-
- /*
* This test drains all regions so cannot be run in parallel with other tests.
*/
@Test(timeout = 30000)