public void initialize(SnapshotOfRegionAssignmentFromMeta snapshotOfRegionAssignment) throws HBaseIOException { globalFavoredNodesAssignmentPlan = snapshotOfRegionAssignment.getExistingAssignmentPlan(); primaryRSToRegionMap = snapshotOfRegionAssignment.getPrimaryToRegionInfoMap(); secondaryRSToRegionMap = snapshotOfRegionAssignment.getSecondaryToRegionInfoMap(); teritiaryRSToRegionMap = snapshotOfRegionAssignment.getTertiaryToRegionInfoMap(); datanodeDataTransferPort = getDataNodePort(); }
@Override public List<RegionPlan> balanceCluster(Map<ServerName, List<HRegionInfo>> clusterState) { //TODO. Look at is whether Stochastic loadbalancer can be integrated with this List<RegionPlan> plans = new ArrayList<RegionPlan>(); //perform a scan of the meta to get the latest updates (if any) SnapshotOfRegionAssignmentFromMeta snaphotOfRegionAssignment = new SnapshotOfRegionAssignmentFromMeta(super.services.getConnection()); try { snaphotOfRegionAssignment.initialize(); } catch (IOException ie) { LOG.warn("Not running balancer since exception was thrown " + ie); return plans; } globalFavoredNodesAssignmentPlan = snaphotOfRegionAssignment.getExistingAssignmentPlan(); Map<ServerName, ServerName> serverNameToServerNameWithoutCode = new HashMap<ServerName, ServerName>(); Map<ServerName, ServerName> serverNameWithoutCodeToServerName = new HashMap<ServerName, ServerName>(); ServerManager serverMgr = super.services.getServerManager(); for (ServerName sn: serverMgr.getOnlineServersList()) { ServerName s = ServerName.valueOf(sn.getHostname(), sn.getPort(), ServerName.NON_STARTCODE); serverNameToServerNameWithoutCode.put(sn, s); serverNameWithoutCodeToServerName.put(s, sn); } for (Map.Entry<ServerName, List<HRegionInfo>> entry : clusterState.entrySet()) { ServerName currentServer = entry.getKey(); //get a server without the startcode for the currentServer ServerName currentServerWithoutStartCode = ServerName.valueOf(currentServer.getHostname(), currentServer.getPort(), ServerName.NON_STARTCODE); List<HRegionInfo> list = entry.getValue(); for (HRegionInfo region : list) { if(region.getTable().getNamespaceAsString() .equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) { continue; } List<ServerName> favoredNodes = globalFavoredNodesAssignmentPlan.getFavoredNodes(region); if (favoredNodes == null || favoredNodes.get(0).equals(currentServerWithoutStartCode)) { continue; //either favorednodes does not exist or we are already on the primary node } ServerName destination = null; //check whether the primary is available destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(0)); if (destination == null) { //check whether the region is on secondary/tertiary if (currentServerWithoutStartCode.equals(favoredNodes.get(1)) || currentServerWithoutStartCode.equals(favoredNodes.get(2))) { continue; } //the region is currently on none of the favored nodes //get it on one of them if possible ServerLoad l1 = super.services.getServerManager().getLoad( serverNameWithoutCodeToServerName.get(favoredNodes.get(1))); ServerLoad l2 = super.services.getServerManager().getLoad( serverNameWithoutCodeToServerName.get(favoredNodes.get(2))); if (l1 != null && l2 != null) { if (l1.getLoad() > l2.getLoad()) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(2)); } else { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(1)); } } else if (l1 != null) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(1)); } else if (l2 != null) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(2)); } } if (destination != null) { RegionPlan plan = new RegionPlan(region, currentServer, destination); plans.add(plan); } } } return plans; }
@Override public List<RegionPlan> balanceCluster(Map<ServerName, List<HRegionInfo>> clusterState) { //TODO. Look at is whether Stochastic loadbalancer can be integrated with this List<RegionPlan> plans = new ArrayList<RegionPlan>(); //perform a scan of the meta to get the latest updates (if any) SnapshotOfRegionAssignmentFromMeta snaphotOfRegionAssignment = new SnapshotOfRegionAssignmentFromMeta(super.services.getCatalogTracker()); try { snaphotOfRegionAssignment.initialize(); } catch (IOException ie) { LOG.warn("Not running balancer since exception was thrown " + ie); return plans; } globalFavoredNodesAssignmentPlan = snaphotOfRegionAssignment.getExistingAssignmentPlan(); Map<ServerName, ServerName> serverNameToServerNameWithoutCode = new HashMap<ServerName, ServerName>(); Map<ServerName, ServerName> serverNameWithoutCodeToServerName = new HashMap<ServerName, ServerName>(); ServerManager serverMgr = super.services.getServerManager(); for (ServerName sn: serverMgr.getOnlineServersList()) { ServerName s = ServerName.valueOf(sn.getHostname(), sn.getPort(), ServerName.NON_STARTCODE); serverNameToServerNameWithoutCode.put(sn, s); serverNameWithoutCodeToServerName.put(s, sn); } for (Map.Entry<ServerName, List<HRegionInfo>> entry : clusterState.entrySet()) { ServerName currentServer = entry.getKey(); //get a server without the startcode for the currentServer ServerName currentServerWithoutStartCode = ServerName.valueOf(currentServer.getHostname(), currentServer.getPort(), ServerName.NON_STARTCODE); List<HRegionInfo> list = entry.getValue(); for (HRegionInfo region : list) { if(region.getTable().getNamespaceAsString() .equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) { continue; } List<ServerName> favoredNodes = globalFavoredNodesAssignmentPlan.getFavoredNodes(region); if (favoredNodes == null || favoredNodes.get(0).equals(currentServerWithoutStartCode)) { continue; //either favorednodes does not exist or we are already on the primary node } ServerName destination = null; //check whether the primary is available destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(0)); if (destination == null) { //check whether the region is on secondary/tertiary if (currentServerWithoutStartCode.equals(favoredNodes.get(1)) || currentServerWithoutStartCode.equals(favoredNodes.get(2))) { continue; } //the region is currently on none of the favored nodes //get it on one of them if possible ServerLoad l1 = super.services.getServerManager().getLoad( serverNameWithoutCodeToServerName.get(favoredNodes.get(1))); ServerLoad l2 = super.services.getServerManager().getLoad( serverNameWithoutCodeToServerName.get(favoredNodes.get(2))); if (l1 != null && l2 != null) { if (l1.getLoad() > l2.getLoad()) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(2)); } else { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(1)); } } else if (l1 != null) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(1)); } else if (l2 != null) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(2)); } } if (destination != null) { RegionPlan plan = new RegionPlan(region, currentServer, destination); plans.add(plan); } } } return plans; }
@Override public List<RegionPlan> balanceCluster(Map<ServerName, List<RegionInfo>> clusterState) { //TODO. Look at is whether Stochastic loadbalancer can be integrated with this List<RegionPlan> plans = new ArrayList<>(); //perform a scan of the meta to get the latest updates (if any) SnapshotOfRegionAssignmentFromMeta snaphotOfRegionAssignment = new SnapshotOfRegionAssignmentFromMeta(super.services.getConnection()); try { snaphotOfRegionAssignment.initialize(); } catch (IOException ie) { LOG.warn("Not running balancer since exception was thrown " + ie); return plans; } // This is not used? Findbugs says so: Map<ServerName, ServerName> serverNameToServerNameWithoutCode = new HashMap<>(); Map<ServerName, ServerName> serverNameWithoutCodeToServerName = new HashMap<>(); ServerManager serverMgr = super.services.getServerManager(); for (ServerName sn: serverMgr.getOnlineServersList()) { ServerName s = ServerName.valueOf(sn.getHostname(), sn.getPort(), ServerName.NON_STARTCODE); // FindBugs complains about useless store! serverNameToServerNameWithoutCode.put(sn, s); serverNameWithoutCodeToServerName.put(s, sn); } for (Map.Entry<ServerName, List<RegionInfo>> entry : clusterState.entrySet()) { ServerName currentServer = entry.getKey(); //get a server without the startcode for the currentServer ServerName currentServerWithoutStartCode = ServerName.valueOf(currentServer.getHostname(), currentServer.getPort(), ServerName.NON_STARTCODE); List<RegionInfo> list = entry.getValue(); for (RegionInfo region : list) { if(!FavoredNodesManager.isFavoredNodeApplicable(region)) { continue; } List<ServerName> favoredNodes = fnm.getFavoredNodes(region); if (favoredNodes == null || favoredNodes.get(0).equals(currentServerWithoutStartCode)) { continue; //either favorednodes does not exist or we are already on the primary node } ServerName destination = null; //check whether the primary is available destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(0)); if (destination == null) { //check whether the region is on secondary/tertiary if (currentServerWithoutStartCode.equals(favoredNodes.get(1)) || currentServerWithoutStartCode.equals(favoredNodes.get(2))) { continue; } //the region is currently on none of the favored nodes //get it on one of them if possible ServerLoad l1 = super.services.getServerManager().getLoad( serverNameWithoutCodeToServerName.get(favoredNodes.get(1))); ServerLoad l2 = super.services.getServerManager().getLoad( serverNameWithoutCodeToServerName.get(favoredNodes.get(2))); if (l1 != null && l2 != null) { if (l1.getLoad() > l2.getLoad()) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(2)); } else { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(1)); } } else if (l1 != null) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(1)); } else if (l2 != null) { destination = serverNameWithoutCodeToServerName.get(favoredNodes.get(2)); } } if (destination != null) { RegionPlan plan = new RegionPlan(region, currentServer, destination); plans.add(plan); } } } return plans; }