diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java index 7172f06..49b02be 100644 --- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java +++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java @@ -93,7 +93,7 @@ public class RSGroupAdminEndpoint extends RSGroupAdminService public void start(CoprocessorEnvironment env) throws IOException { MasterCoprocessorEnvironment menv = (MasterCoprocessorEnvironment)env; master = menv.getMasterServices(); - groupInfoManager = new RSGroupInfoManagerImpl(master); + setGroupInfoManager(new RSGroupInfoManagerImpl(master)); groupAdminServer = new RSGroupAdminServer(master, groupInfoManager); Class clazz = master.getConfiguration().getClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, null); @@ -111,6 +111,20 @@ public class RSGroupAdminEndpoint extends RSGroupAdminService return this; } + private static void setStaticGroupInfoManager(RSGroupInfoManagerImpl groupInfoManager) { + RSGroupAdminEndpoint.groupInfoManager = groupInfoManager; + } + + private void setGroupInfoManager(RSGroupInfoManagerImpl groupInfoManager) throws IOException { + if (groupInfoManager == null) { + groupInfoManager = new RSGroupInfoManagerImpl(master); + groupInfoManager.init(); + } else if (!groupInfoManager.isInit()) { + groupInfoManager.init(); + } + setStaticGroupInfoManager(groupInfoManager); + } + public RSGroupInfoManager getGroupInfoManager() { return groupInfoManager; } diff --git a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminServer.java b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminServer.java index 7aea464..dc28f7d 100644 --- a/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminServer.java +++ b/hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminServer.java @@ -216,7 +216,7 @@ public class RSGroupAdminServer extends RSGroupAdmin { } } try { - Thread.sleep(1000); + manager.wait(1000); } catch (InterruptedException e) { LOG.warn("Sleep interrupted", e); Thread.currentThread().interrupt(); 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 0e7f267..b5dd743 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 @@ -316,18 +316,19 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc private Set getMisplacedRegions( Map regions) throws IOException { Set misplacedRegions = new HashSet(); - for (HRegionInfo region : regions.keySet()) { - ServerName assignedServer = regions.get(region); + for(Map.Entry region : regions.entrySet()) { + HRegionInfo regionInfo = region.getKey(); + ServerName assignedServer = region.getValue(); RSGroupInfo info = - RSGroupInfoManager.getRSGroup(RSGroupInfoManager.getRSGroupOfTable(region.getTable())); + RSGroupInfoManager.getRSGroup(RSGroupInfoManager.getRSGroupOfTable(regionInfo.getTable())); if (assignedServer != null && (info == null || !info.containsServer(assignedServer.getHostPort()))) { - LOG.debug("Found misplaced region: " + region.getRegionNameAsString() + + LOG.debug("Found misplaced region: " + regionInfo.getRegionNameAsString() + " on server: " + assignedServer + " found in group: " + RSGroupInfoManager.getRSGroupOfServer(assignedServer.getHostPort()) + " outside of group: " + (info == null ? "UNKNOWN" : info.getName())); - misplacedRegions.add(region); + misplacedRegions.add(regionInfo); } } return misplacedRegions; @@ -339,9 +340,10 @@ public class RSGroupBasedLoadBalancer implements RSGroupableBalancer, LoadBalanc new TreeMap>(); List misplacedRegions = new LinkedList(); correctAssignments.put(LoadBalancer.BOGUS_SERVER_NAME, new LinkedList()); - for (ServerName sName : existingAssignments.keySet()) { + for (Map.Entry> assignments : existingAssignments.entrySet()){ + ServerName sName = assignments.getKey(); correctAssignments.put(sName, new LinkedList()); - List regions = existingAssignments.get(sName); + List regions = assignments.getValue(); for (HRegionInfo region : regions) { RSGroupInfo info = null; try { 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 542ad9a..10e7eaf 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 @@ -122,6 +122,7 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene private volatile Set prevRSGroups; private RSGroupSerDe rsGroupSerDe; private DefaultServerUpdater defaultServerUpdater; + private boolean isInit = false; public RSGroupInfoManagerImpl(MasterServices master) throws IOException { @@ -131,13 +132,21 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene this.master = master; this.watcher = master.getZooKeeper(); this.conn = master.getClusterConnection(); - rsGroupStartupWorker = new RSGroupStartupWorker(this, master, conn); prevRSGroups = new HashSet(); + } + + public 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; + } + + boolean isInit() { + return isInit; } /** @@ -669,6 +678,8 @@ public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListene //flush any inconsistencies between ZK and HTable groupInfoManager.flushConfig(groupInfoManager.rsGroupMap); } + } catch (RuntimeException e) { + throw e; } catch(Exception e) { found.set(false); LOG.warn("Failed to perform check", e);