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

HBASE-1157, HBASE-1156 If we do not take start code as a part of region server recovery, we could inadvertantly try to reassign regions assigned to a restarted server with a different start code; Improve lease handling 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@753483 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jim Kellerman 2009-03-14 01:38:45 +00:00
parent d67d131f10
commit e8520cf5d3
15 changed files with 326 additions and 284 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGES.txt
View File

@ -42,6 +42,10 @@ Release 0.20.0 - Unreleased
HBASE-1169 When a shutdown is requested, stop scanning META regions immediately HBASE-1169 When a shutdown is requested, stop scanning META regions immediately
HBASE-1251 HConnectionManager.getConnection(HBaseConfiguration) returns same HBASE-1251 HConnectionManager.getConnection(HBaseConfiguration) returns same
HConnection for different HBaseConfigurations HConnection for different HBaseConfigurations
HBASE-1157, HBASE-1156 If we do not take start code as a part of region
server recovery, we could inadvertantly try to reassign regions
assigned to a restarted server with a different start code;
Improve lease handling
IMPROVEMENTS IMPROVEMENTS
HBASE-1089 Add count of regions on filesystem to master UI; add percentage HBASE-1089 Add count of regions on filesystem to master UI; add percentage

6
src/java/org/apache/hadoop/hbase/HRegionInfo.java
View File

@ -23,7 +23,6 @@ import java.io.DataInput;
import java.io.DataOutput; import java.io.DataOutput;
import java.io.IOException; import java.io.IOException;
import org.apache.hadoop.hbase.HStoreKey;
import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.JenkinsHash; import org.apache.hadoop.hbase.util.JenkinsHash;
import org.apache.hadoop.io.VersionedWritable; import org.apache.hadoop.io.VersionedWritable;
@ -59,7 +58,7 @@ public class HRegionInfo extends VersionedWritable implements WritableComparable
private byte [] endKey = HConstants.EMPTY_BYTE_ARRAY; private byte [] endKey = HConstants.EMPTY_BYTE_ARRAY;
private boolean offLine = false; private boolean offLine = false;
private long regionId = -1; private long regionId = -1;
private byte [] regionName = HConstants.EMPTY_BYTE_ARRAY; private transient byte [] regionName = HConstants.EMPTY_BYTE_ARRAY;
private String regionNameStr = ""; private String regionNameStr = "";
private boolean split = false; private boolean split = false;
private byte [] startKey = HConstants.EMPTY_BYTE_ARRAY; private byte [] startKey = HConstants.EMPTY_BYTE_ARRAY;
@ -221,6 +220,7 @@ public class HRegionInfo extends VersionedWritable implements WritableComparable
* Separate elements of a regionName. * Separate elements of a regionName.
* @param regionName * @param regionName
* @return Array of byte[] containing tableName, startKey and id * @return Array of byte[] containing tableName, startKey and id
* @throws IOException
*/ */
public static byte [][] parseRegionName(final byte [] regionName) public static byte [][] parseRegionName(final byte [] regionName)
throws IOException { throws IOException {
@ -438,6 +438,8 @@ public class HRegionInfo extends VersionedWritable implements WritableComparable
/** /**
* For internal use in forcing splits ahead of file size limit. * For internal use in forcing splits ahead of file size limit.
* @param b
* @return previous value
*/ */
public boolean shouldSplit(boolean b) { public boolean shouldSplit(boolean b) {
boolean old = this.splitRequest; boolean old = this.splitRequest;

71
src/java/org/apache/hadoop/hbase/HServerInfo.java
View File

@ -38,6 +38,7 @@ public class HServerInfo implements WritableComparable<HServerInfo> {
private long startCode; private long startCode;
private HServerLoad load; private HServerLoad load;
private int infoPort; private int infoPort;
private transient volatile String serverName = null;
/** default constructor - used by Writable */ /** default constructor - used by Writable */
public HServerInfo() { public HServerInfo() {
@ -85,15 +86,16 @@ public class HServerInfo implements WritableComparable<HServerInfo> {
/** @return the server address */ /** @return the server address */
public HServerAddress getServerAddress() { public HServerAddress getServerAddress() {
return serverAddress; return new HServerAddress(serverAddress);
} }
/** /**
* Change the server address. * Change the server address.
* @param serverAddress New server address * @param serverAddress New server address
*/ */
public void setServerAddress(HServerAddress serverAddress) { public synchronized void setServerAddress(HServerAddress serverAddress) {
this.serverAddress = serverAddress; this.serverAddress = serverAddress;
this.serverName = null;
} }
/** @return the server start code */ /** @return the server start code */
@ -111,8 +113,19 @@ public class HServerInfo implements WritableComparable<HServerInfo> {
/** /**
* @param startCode the startCode to set * @param startCode the startCode to set
*/ */
public void setStartCode(long startCode) { public synchronized void setStartCode(long startCode) {
this.startCode = startCode; this.startCode = startCode;
this.serverName = null;
}
/**
* @return the server name in the form hostname_startcode_port
*/
public synchronized String getServerName() {
if (this.serverName == null) {
this.serverName = getServerName(this.serverAddress, this.startCode);
}
return this.serverName;
} }
@Override @Override
@ -128,10 +141,7 @@ public class HServerInfo implements WritableComparable<HServerInfo> {
@Override @Override
public int hashCode() { public int hashCode() {
int result = this.serverAddress.hashCode(); return this.getServerName().hashCode();
result ^= this.infoPort;
result ^= this.startCode;
return result;
} }
@ -152,17 +162,46 @@ public class HServerInfo implements WritableComparable<HServerInfo> {
} }
public int compareTo(HServerInfo o) { public int compareTo(HServerInfo o) {
int result = getServerAddress().compareTo(o.getServerAddress()); return this.getServerName().compareTo(o.getServerName());
if (result != 0) {
return result;
} }
if (this.infoPort != o.infoPort) {
return this.infoPort - o.infoPort; /**
* @param info
* @return the server name in the form hostname_startcode_port
*/
public static String getServerName(HServerInfo info) {
return getServerName(info.getServerAddress(), info.getStartCode());
} }
if (getStartCode() == o.getStartCode()) {
return 0; /**
* @param serverAddress in the form hostname:port
* @param startCode
* @return the server name in the form hostname_startcode_port
*/
public static String getServerName(String serverAddress, long startCode) {
String name = null;
if (serverAddress != null) {
HServerAddress address = new HServerAddress(serverAddress);
name = getServerName(address.getHostname(), address.getPort(), startCode);
} }
// Startcodes are timestamps. return name;
return (int)(getStartCode() - o.getStartCode()); }
/**
* @param address
* @param startCode
* @return the server name in the form hostname_startcode_port
*/
public static String getServerName(HServerAddress address, long startCode) {
return getServerName(address.getHostname(), address.getPort(), startCode);
}
private static String getServerName(String hostName, int port, long startCode) {
StringBuilder name = new StringBuilder(hostName);
name.append("_");
name.append(startCode);
name.append("_");
name.append(port);
return name.toString();
} }
} }

35
src/java/org/apache/hadoop/hbase/master/BaseScanner.java
View File

@ -327,40 +327,40 @@ abstract class BaseScanner extends Chore implements HConstants {
} }
protected void checkAssigned(final HRegionInfo info, protected void checkAssigned(final HRegionInfo info,
final String serverName, final long startCode) final String serverAddress, final long startCode)
throws IOException { throws IOException {
String serverName = null;
if (serverAddress != null && serverAddress.length() > 0) {
serverName = HServerInfo.getServerName(serverAddress, startCode);
}
HServerInfo storedInfo = null;
synchronized (this.master.regionManager) { synchronized (this.master.regionManager) {
if (serverName != null) {
/* /*
* We don't assign regions that are offline, in transition or were on * We don't assign regions that are offline, in transition or were on
* a dead server. Regions that were on a dead server will get reassigned * a dead server. Regions that were on a dead server will get reassigned
* by ProcessServerShutdown * by ProcessServerShutdown
*/ */
if(info.isOffline() || if(info.isOffline() ||
this.master.regionManager.regionIsInTransition(info.getRegionName()) || this.master.regionManager.regionIsInTransition(
info.getRegionNameAsString()) ||
this.master.serverManager.isDead(serverName)) { this.master.serverManager.isDead(serverName)) {
return; return;
} }
HServerInfo storedInfo = null;
if (serverName.length() != 0) {
storedInfo = this.master.serverManager.getServerInfo(serverName); storedInfo = this.master.serverManager.getServerInfo(serverName);
} }
/* // If we can't find the HServerInfo, then add it to the list of
* If the startcode is off -- either null or doesn't match the start code // unassigned regions.
* for the address -- then add it to the list of unassigned regions.
*/
if (storedInfo == null || storedInfo.getStartCode() != startCode) {
if (storedInfo == null) {
// The current assignment is invalid // The current assignment is invalid
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Current assignment of " + info.getRegionNameAsString() + LOG.debug("Current assignment of " + info.getRegionNameAsString() +
" is not valid; " + " is not valid; " + " Server '" + serverAddress + "' startCode: " +
(storedInfo == null ? " Server '" + serverName + "' unknown." : startCode + " unknown.");
" serverInfo: " + storedInfo + ", passed startCode: " +
startCode + ", storedInfo.startCode: " +
storedInfo.getStartCode()));
} }
// Recover the region server's log if there is one. // Recover the region server's log if there is one.
@ -368,10 +368,9 @@ abstract class BaseScanner extends Chore implements HConstants {
// data in the meta region. Once we are on-line, dead server log // data in the meta region. Once we are on-line, dead server log
// recovery is handled by lease expiration and ProcessServerShutdown // recovery is handled by lease expiration and ProcessServerShutdown
if (!this.master.regionManager.isInitialMetaScanComplete() && if (!this.master.regionManager.isInitialMetaScanComplete() &&
serverName.length() != 0) { serverName != null) {
StringBuilder dirName = new StringBuilder("log_"); Path logDir =
dirName.append(serverName.replace(":", "_")); new Path(this.master.rootdir, HLog.getHLogDirectoryName(serverName));
Path logDir = new Path(this.master.rootdir, dirName.toString());
try { try {
if (master.fs.exists(logDir)) { if (master.fs.exists(logDir)) {
this.master.regionManager.splitLogLock.lock(); this.master.regionManager.splitLogLock.lock();

7
src/java/org/apache/hadoop/hbase/master/ChangeTableState.java
View File

@ -49,10 +49,9 @@ class ChangeTableState extends TableOperation {
} }
@Override @Override
protected void processScanItem(String serverName, long startCode, protected void processScanItem(String serverName, HRegionInfo info) {
HRegionInfo info) {
if (isBeingServed(serverName, startCode)) { if (isBeingServed(serverName)) {
HashSet<HRegionInfo> regions = servedRegions.get(serverName); HashSet<HRegionInfo> regions = servedRegions.get(serverName);
if (regions == null) { if (regions == null) {
regions = new HashSet<HRegionInfo>(); regions = new HashSet<HRegionInfo>();
@ -91,7 +90,7 @@ class ChangeTableState extends TableOperation {
synchronized (master.regionManager) { synchronized (master.regionManager) {
if (online) { if (online) {
// Bring offline regions on-line // Bring offline regions on-line
if (!master.regionManager.regionIsOpening(i.getRegionName())) { if (!master.regionManager.regionIsOpening(i.getRegionNameAsString())) {
master.regionManager.setUnassigned(i, false); master.regionManager.setUnassigned(i, false);
} }
} else { } else {

5
src/java/org/apache/hadoop/hbase/master/ColumnOperation.java
View File

@ -38,8 +38,9 @@ abstract class ColumnOperation extends TableOperation {
} }
@Override @Override
protected void processScanItem(String serverName, long startCode, protected void processScanItem(@SuppressWarnings("unused") String serverName,
final HRegionInfo info) throws IOException { final HRegionInfo info)
throws IOException {
if (isEnabled(info)) { if (isEnabled(info)) {
throw new TableNotDisabledException(tableName); throw new TableNotDisabledException(tableName);
} }

3
src/java/org/apache/hadoop/hbase/master/HMaster.java
View File

@ -548,7 +548,8 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
/* /*
* HMasterRegionInterface * HMasterRegionInterface
*/ */
public MapWritable regionServerStartup(final HServerInfo serverInfo) { public MapWritable regionServerStartup(final HServerInfo serverInfo)
throws IOException {
// Set the address for now even tho it will not be persisted on HRS side. // Set the address for now even tho it will not be persisted on HRS side.
String rsAddress = HBaseServer.getRemoteAddress(); String rsAddress = HBaseServer.getRemoteAddress();
serverInfo.setServerAddress(new HServerAddress(rsAddress, serverInfo.setServerAddress(new HServerAddress(rsAddress,

2
src/java/org/apache/hadoop/hbase/master/ModifyTableMeta.java
View File

@ -57,7 +57,7 @@ class ModifyTableMeta extends TableOperation {
} }
@Override @Override
protected void processScanItem(String serverName, long startCode, protected void processScanItem(@SuppressWarnings("unused") String serverName,
final HRegionInfo info) throws IOException { final HRegionInfo info) throws IOException {
if (isEnabled(info)) { if (isEnabled(info)) {
throw new TableNotDisabledException(tableName.toString()); throw new TableNotDisabledException(tableName.toString());

34
src/java/org/apache/hadoop/hbase/master/ProcessRegionOpen.java
View File

@ -34,8 +34,7 @@ import org.apache.hadoop.hbase.util.Bytes;
* root region which is handled specially. * root region which is handled specially.
*/ */
class ProcessRegionOpen extends ProcessRegionStatusChange { class ProcessRegionOpen extends ProcessRegionStatusChange {
protected final HServerAddress serverAddress; protected final HServerInfo serverInfo;
protected final byte [] startCode;
/** /**
* @param master * @param master
@ -48,17 +47,16 @@ class ProcessRegionOpen extends ProcessRegionStatusChange {
HRegionInfo regionInfo) HRegionInfo regionInfo)
throws IOException { throws IOException {
super(master, regionInfo); super(master, regionInfo);
this.serverAddress = info.getServerAddress(); if (info == null) {
if (this.serverAddress == null) { throw new NullPointerException("HServerInfo cannot be null; " +
throw new NullPointerException("Server address cannot be null; " +
"hbase-958 debugging"); "hbase-958 debugging");
} }
this.startCode = Bytes.toBytes(info.getStartCode()); this.serverInfo = info;
} }
@Override @Override
public String toString() { public String toString() {
return "PendingOpenOperation from " + serverAddress.toString(); return "PendingOpenOperation from " + HServerInfo.getServerName(serverInfo);
} }
@Override @Override
@ -69,7 +67,7 @@ class ProcessRegionOpen extends ProcessRegionStatusChange {
public Boolean call() throws IOException { public Boolean call() throws IOException {
LOG.info(regionInfo.getRegionNameAsString() + " open on " + LOG.info(regionInfo.getRegionNameAsString() + " open on " +
serverAddress.toString()); serverInfo.getServerAddress().toString());
if (!metaRegionAvailable()) { if (!metaRegionAvailable()) {
// We can't proceed unless the meta region we are going to update // We can't proceed unless the meta region we are going to update
// is online. metaRegionAvailable() has put this operation on the // is online. metaRegionAvailable() has put this operation on the
@ -80,12 +78,13 @@ class ProcessRegionOpen extends ProcessRegionStatusChange {
// Register the newly-available Region's location. // Register the newly-available Region's location.
LOG.info("updating row " + regionInfo.getRegionNameAsString() + LOG.info("updating row " + regionInfo.getRegionNameAsString() +
" in region " + Bytes.toString(metaRegionName) + " in region " + Bytes.toString(metaRegionName) + " with " +
" with startcode " + Bytes.toLong(startCode) + " and server " + " with startcode " + serverInfo.getStartCode() + " and server " +
serverAddress.toString()); serverInfo.getServerAddress());
BatchUpdate b = new BatchUpdate(regionInfo.getRegionName()); BatchUpdate b = new BatchUpdate(regionInfo.getRegionName());
b.put(COL_SERVER, Bytes.toBytes(serverAddress.toString())); b.put(COL_SERVER,
b.put(COL_STARTCODE, startCode); Bytes.toBytes(serverInfo.getServerAddress().toString()));
b.put(COL_STARTCODE, Bytes.toBytes(serverInfo.getStartCode()));
server.batchUpdate(metaRegionName, b, -1L); server.batchUpdate(metaRegionName, b, -1L);
if (!this.historian.isOnline()) { if (!this.historian.isOnline()) {
// This is safest place to do the onlining of the historian in // This is safest place to do the onlining of the historian in
@ -93,19 +92,24 @@ class ProcessRegionOpen extends ProcessRegionStatusChange {
// for the historian to go against. // for the historian to go against.
this.historian.online(this.master.getConfiguration()); this.historian.online(this.master.getConfiguration());
} }
this.historian.addRegionOpen(regionInfo, serverAddress); this.historian.addRegionOpen(regionInfo, serverInfo.getServerAddress());
synchronized (master.regionManager) { synchronized (master.regionManager) {
if (isMetaTable) { if (isMetaTable) {
// It's a meta region. // It's a meta region.
MetaRegion m = new MetaRegion(new HServerAddress(serverAddress), MetaRegion m =
new MetaRegion(new HServerAddress(serverInfo.getServerAddress()),
regionInfo.getRegionName(), regionInfo.getStartKey()); regionInfo.getRegionName(), regionInfo.getStartKey());
if (!master.regionManager.isInitialMetaScanComplete()) { if (!master.regionManager.isInitialMetaScanComplete()) {
// Put it on the queue to be scanned for the first time. // Put it on the queue to be scanned for the first time.
if (LOG.isDebugEnabled()) {
LOG.debug("Adding " + m.toString() + " to regions to scan"); LOG.debug("Adding " + m.toString() + " to regions to scan");
}
master.regionManager.addMetaRegionToScan(m); master.regionManager.addMetaRegionToScan(m);
} else { } else {
// Add it to the online meta regions // Add it to the online meta regions
if (LOG.isDebugEnabled()) {
LOG.debug("Adding to onlineMetaRegions: " + m.toString()); LOG.debug("Adding to onlineMetaRegions: " + m.toString());
}
master.regionManager.putMetaRegionOnline(m); master.regionManager.putMetaRegionOnline(m);
// Interrupting the Meta Scanner sleep so that it can // Interrupting the Meta Scanner sleep so that it can
// process regions right away // process regions right away

29
src/java/org/apache/hadoop/hbase/master/ProcessServerShutdown.java
View File

@ -28,7 +28,6 @@ import java.util.Set;
import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HServerAddress;
import org.apache.hadoop.hbase.HServerInfo; import org.apache.hadoop.hbase.HServerInfo;
import org.apache.hadoop.hbase.RemoteExceptionHandler; import org.apache.hadoop.hbase.RemoteExceptionHandler;
import org.apache.hadoop.hbase.ipc.HRegionInterface; import org.apache.hadoop.hbase.ipc.HRegionInterface;
@ -44,11 +43,7 @@ import org.apache.hadoop.hbase.io.RowResult;
* serving, and the regions need to get reassigned. * serving, and the regions need to get reassigned.
*/ */
class ProcessServerShutdown extends RegionServerOperation { class ProcessServerShutdown extends RegionServerOperation {
private final HServerAddress deadServer; private final String deadServer;
/*
* Cache of the server name.
*/
private final String deadServerStr;
private final boolean rootRegionServer; private final boolean rootRegionServer;
private boolean rootRegionReassigned = false; private boolean rootRegionReassigned = false;
private Path oldLogDir; private Path oldLogDir;
@ -76,8 +71,7 @@ class ProcessServerShutdown extends RegionServerOperation {
public ProcessServerShutdown(HMaster master, HServerInfo serverInfo, public ProcessServerShutdown(HMaster master, HServerInfo serverInfo,
boolean rootRegionServer) { boolean rootRegionServer) {
super(master); super(master);
this.deadServer = serverInfo.getServerAddress(); this.deadServer = HServerInfo.getServerName(serverInfo);
this.deadServerStr = this.deadServer.toString();
this.rootRegionServer = rootRegionServer; this.rootRegionServer = rootRegionServer;
this.logSplit = false; this.logSplit = false;
this.rootRescanned = false; this.rootRescanned = false;
@ -87,7 +81,7 @@ class ProcessServerShutdown extends RegionServerOperation {
@Override @Override
public String toString() { public String toString() {
return "ProcessServerShutdown of " + this.deadServerStr; return "ProcessServerShutdown of " + this.deadServer;
} }
/** Finds regions that the dead region server was serving /** Finds regions that the dead region server was serving
@ -116,8 +110,13 @@ class ProcessServerShutdown extends RegionServerOperation {
// shutdown server but that would mean that we'd reassign regions that // shutdown server but that would mean that we'd reassign regions that
// were already out being assigned, ones that were product of a split // were already out being assigned, ones that were product of a split
// that happened while the shutdown was being processed. // that happened while the shutdown was being processed.
String serverName = Writables.cellToString(values.get(COL_SERVER)); String serverAddress = Writables.cellToString(values.get(COL_SERVER));
if (serverName == null || !deadServerStr.equals(serverName)) { long startCode = Writables.cellToLong(values.get(COL_STARTCODE));
String serverName = null;
if (serverAddress != null && serverAddress.length() > 0) {
serverName = HServerInfo.getServerName(serverAddress, startCode);
}
if (serverName == null || !deadServer.equals(serverName)) {
// This isn't the server you're looking for - move along // This isn't the server you're looking for - move along
continue; continue;
} }
@ -146,7 +145,7 @@ class ProcessServerShutdown extends RegionServerOperation {
ToDoEntry todo = new ToDoEntry(row, info); ToDoEntry todo = new ToDoEntry(row, info);
toDoList.add(todo); toDoList.add(todo);
if (master.regionManager.isOfflined(info.getRegionName()) || if (master.regionManager.isOfflined(info.getRegionNameAsString()) ||
info.isOffline()) { info.isOffline()) {
master.regionManager.removeRegion(info); master.regionManager.removeRegion(info);
// Mark region offline // Mark region offline
@ -232,7 +231,7 @@ class ProcessServerShutdown extends RegionServerOperation {
@Override @Override
protected boolean process() throws IOException { protected boolean process() throws IOException {
LOG.info("process shutdown of server " + this.deadServerStr + LOG.info("process shutdown of server " + this.deadServer +
": logSplit: " + ": logSplit: " +
logSplit + ", rootRescanned: " + rootRescanned + logSplit + ", rootRescanned: " + rootRescanned +
", numberOfMetaRegions: " + ", numberOfMetaRegions: " +
@ -310,9 +309,9 @@ class ProcessServerShutdown extends RegionServerOperation {
} }
} }
// Remove this server from dead servers list. Finished splitting logs. // Remove this server from dead servers list. Finished splitting logs.
this.master.serverManager.removeDeadServer(deadServerStr); this.master.serverManager.removeDeadServer(deadServer);
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Removed " + deadServerStr + " from deadservers Map"); LOG.debug("Removed " + deadServer + " from deadservers Map");
} }
return true; return true;
} }

101
src/java/org/apache/hadoop/hbase/master/RegionManager.java
View File

@ -96,9 +96,8 @@ class RegionManager implements HConstants {
* *
* @see RegionState inner-class below * @see RegionState inner-class below
*/ */
private final SortedMap<byte[], RegionState> regionsInTransition = private final SortedMap<String, RegionState> regionsInTransition =
Collections.synchronizedSortedMap( Collections.synchronizedSortedMap(new TreeMap<String, RegionState>());
new TreeMap<byte[], RegionState>(Bytes.BYTES_COMPARATOR));
// How many regions to assign a server at a time. // How many regions to assign a server at a time.
private final int maxAssignInOneGo; private final int maxAssignInOneGo;
@ -108,23 +107,23 @@ class RegionManager implements HConstants {
private final float slop; private final float slop;
/** Set of regions to split. */ /** Set of regions to split. */
private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>> regionsToSplit = private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>>
Collections.synchronizedSortedMap( regionsToSplit = Collections.synchronizedSortedMap(
new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>> new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>>
(Bytes.BYTES_COMPARATOR)); (Bytes.BYTES_COMPARATOR));
/** Set of regions to compact. */ /** Set of regions to compact. */
private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>> regionsToCompact = private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>>
Collections.synchronizedSortedMap( regionsToCompact = Collections.synchronizedSortedMap(
new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>> new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>>
(Bytes.BYTES_COMPARATOR)); (Bytes.BYTES_COMPARATOR));
/** Set of regions to major compact. */ /** Set of regions to major compact. */
private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>> regionsToMajorCompact = private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>>
Collections.synchronizedSortedMap( regionsToMajorCompact = Collections.synchronizedSortedMap(
new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>> new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>>
(Bytes.BYTES_COMPARATOR)); (Bytes.BYTES_COMPARATOR));
/** Set of regions to flush. */ /** Set of regions to flush. */
private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>> regionsToFlush = private final SortedMap<byte[], Pair<HRegionInfo,HServerAddress>>
Collections.synchronizedSortedMap( regionsToFlush = Collections.synchronizedSortedMap(
new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>> new TreeMap<byte[],Pair<HRegionInfo,HServerAddress>>
(Bytes.BYTES_COMPARATOR)); (Bytes.BYTES_COMPARATOR));
@ -163,7 +162,8 @@ class RegionManager implements HConstants {
void unsetRootRegion() { void unsetRootRegion() {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
rootRegionLocation.set(null); rootRegionLocation.set(null);
regionsInTransition.remove(HRegionInfo.ROOT_REGIONINFO.getRegionName()); regionsInTransition.remove(
HRegionInfo.ROOT_REGIONINFO.getRegionNameAsString());
} }
} }
@ -173,7 +173,8 @@ class RegionManager implements HConstants {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = new RegionState(HRegionInfo.ROOT_REGIONINFO); RegionState s = new RegionState(HRegionInfo.ROOT_REGIONINFO);
s.setUnassigned(); s.setUnassigned();
regionsInTransition.put(HRegionInfo.ROOT_REGIONINFO.getRegionName(), s); regionsInTransition.put(
HRegionInfo.ROOT_REGIONINFO.getRegionNameAsString(), s);
} }
} }
} }
@ -184,11 +185,11 @@ class RegionManager implements HConstants {
* (ServerManager.processMsgs) already owns the monitor for the RegionManager. * (ServerManager.processMsgs) already owns the monitor for the RegionManager.
* *
* @param info * @param info
* @param serverName * @param mostLoadedRegions
* @param returnMsgs * @param returnMsgs
*/ */
void assignRegions(HServerInfo info, String serverName, void assignRegions(HServerInfo info, HRegionInfo[] mostLoadedRegions,
HRegionInfo[] mostLoadedRegions, ArrayList<HMsg> returnMsgs) { ArrayList<HMsg> returnMsgs) {
HServerLoad thisServersLoad = info.getLoad(); HServerLoad thisServersLoad = info.getLoad();
// figure out what regions need to be assigned and aren't currently being // figure out what regions need to be assigned and aren't currently being
// worked on elsewhere. // worked on elsewhere.
@ -204,24 +205,24 @@ class RegionManager implements HConstants {
if (avgLoad > 2.0 && if (avgLoad > 2.0 &&
thisServersLoad.getNumberOfRegions() > avgLoadWithSlop) { thisServersLoad.getNumberOfRegions() > avgLoadWithSlop) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Server " + serverName + LOG.debug("Server " + info.getServerName() +
" is overloaded. Server load: " + " is overloaded. Server load: " +
thisServersLoad.getNumberOfRegions() + " avg: " + avgLoad + thisServersLoad.getNumberOfRegions() + " avg: " + avgLoad +
", slop: " + this.slop); ", slop: " + this.slop);
} }
unassignSomeRegions(serverName, thisServersLoad, unassignSomeRegions(info, thisServersLoad,
avgLoad, mostLoadedRegions, returnMsgs); avgLoad, mostLoadedRegions, returnMsgs);
} }
} }
} else { } else {
// if there's only one server, just give it all the regions // if there's only one server, just give it all the regions
if (master.serverManager.numServers() == 1) { if (master.serverManager.numServers() == 1) {
assignRegionsToOneServer(regionsToAssign, serverName, returnMsgs); assignRegionsToOneServer(regionsToAssign, info, returnMsgs);
} else { } else {
// otherwise, give this server a few regions taking into account the // otherwise, give this server a few regions taking into account the
// load of all the other servers. // load of all the other servers.
assignRegionsToMultipleServers(thisServersLoad, regionsToAssign, assignRegionsToMultipleServers(thisServersLoad, regionsToAssign,
serverName, returnMsgs); info, returnMsgs);
} }
} }
} }
@ -234,7 +235,7 @@ class RegionManager implements HConstants {
* whose caller owns the monitor for RegionManager * whose caller owns the monitor for RegionManager
*/ */
private void assignRegionsToMultipleServers(final HServerLoad thisServersLoad, private void assignRegionsToMultipleServers(final HServerLoad thisServersLoad,
final Set<RegionState> regionsToAssign, final String serverName, final Set<RegionState> regionsToAssign, final HServerInfo info,
final ArrayList<HMsg> returnMsgs) { final ArrayList<HMsg> returnMsgs) {
int nRegionsToAssign = regionsToAssign.size(); int nRegionsToAssign = regionsToAssign.size();
@ -281,9 +282,10 @@ class RegionManager implements HConstants {
for (RegionState s: regionsToAssign) { for (RegionState s: regionsToAssign) {
LOG.info("assigning region " + Bytes.toString(s.getRegionName())+ LOG.info("assigning region " + Bytes.toString(s.getRegionName())+
" to server " + serverName); " to server " + info.getServerName());
s.setPendingOpen(serverName); s.setPendingOpen(info.getServerName());
this.historian.addRegionAssignment(s.getRegionInfo(), serverName); this.historian.addRegionAssignment(s.getRegionInfo(),
info.getServerName());
returnMsgs.add( returnMsgs.add(
new HMsg(HMsg.Type.MSG_REGION_OPEN, s.getRegionInfo(), inSafeMode())); new HMsg(HMsg.Type.MSG_REGION_OPEN, s.getRegionInfo(), inSafeMode()));
if (--nregions <= 0) { if (--nregions <= 0) {
@ -406,12 +408,12 @@ class RegionManager implements HConstants {
* @param returnMsgs * @param returnMsgs
*/ */
private void assignRegionsToOneServer(final Set<RegionState> regionsToAssign, private void assignRegionsToOneServer(final Set<RegionState> regionsToAssign,
final String serverName, final ArrayList<HMsg> returnMsgs) { final HServerInfo info, final ArrayList<HMsg> returnMsgs) {
for (RegionState s: regionsToAssign) { for (RegionState s: regionsToAssign) {
LOG.info("assigning region " + Bytes.toString(s.getRegionName()) + LOG.info("assigning region " + Bytes.toString(s.getRegionName()) +
" to the only server " + serverName); " to the only server " + info.getServerName());
s.setPendingOpen(serverName); s.setPendingOpen(info.getServerName());
this.historian.addRegionAssignment(s.getRegionInfo(), serverName); this.historian.addRegionAssignment(s.getRegionInfo(), info.getServerName());
returnMsgs.add( returnMsgs.add(
new HMsg(HMsg.Type.MSG_REGION_OPEN, s.getRegionInfo(), inSafeMode())); new HMsg(HMsg.Type.MSG_REGION_OPEN, s.getRegionInfo(), inSafeMode()));
} }
@ -425,7 +427,7 @@ class RegionManager implements HConstants {
* Note that no synchronization is needed because the only caller * Note that no synchronization is needed because the only caller
* (assignRegions) whose caller owns the monitor for RegionManager * (assignRegions) whose caller owns the monitor for RegionManager
*/ */
private void unassignSomeRegions(final String serverName, private void unassignSomeRegions(final HServerInfo info,
final HServerLoad load, final double avgLoad, final HServerLoad load, final double avgLoad,
final HRegionInfo[] mostLoadedRegions, ArrayList<HMsg> returnMsgs) { final HRegionInfo[] mostLoadedRegions, ArrayList<HMsg> returnMsgs) {
int numRegionsToClose = load.getNumberOfRegions() - (int)Math.ceil(avgLoad); int numRegionsToClose = load.getNumberOfRegions() - (int)Math.ceil(avgLoad);
@ -443,19 +445,20 @@ class RegionManager implements HConstants {
if (currentRegion.isRootRegion() || currentRegion.isMetaTable()) { if (currentRegion.isRootRegion() || currentRegion.isMetaTable()) {
continue; continue;
} }
byte[] regionName = currentRegion.getRegionName(); String regionName = currentRegion.getRegionNameAsString();
if (regionIsInTransition(regionName)) { if (regionIsInTransition(regionName)) {
skipped++; skipped++;
continue; continue;
} }
LOG.debug("Going to close region " + if (LOG.isDebugEnabled()) {
currentRegion.getRegionNameAsString()); LOG.debug("Going to close region " + regionName);
}
// make a message to close the region // make a message to close the region
returnMsgs.add(new HMsg(HMsg.Type.MSG_REGION_CLOSE, currentRegion, returnMsgs.add(new HMsg(HMsg.Type.MSG_REGION_CLOSE, currentRegion,
OVERLOADED, inSafeMode())); OVERLOADED, inSafeMode()));
// mark the region as closing // mark the region as closing
setClosing(serverName, currentRegion, false); setClosing(info.getServerName(), currentRegion, false);
setPendingClose(currentRegion.getRegionName()); setPendingClose(regionName);
// increment the count of regions we've marked // increment the count of regions we've marked
regionsClosed++; regionsClosed++;
} }
@ -716,14 +719,14 @@ class RegionManager implements HConstants {
* @param info * @param info
*/ */
public void removeRegion(HRegionInfo info) { public void removeRegion(HRegionInfo info) {
regionsInTransition.remove(info.getRegionName()); regionsInTransition.remove(info.getRegionNameAsString());
} }
/** /**
* @param regionName * @param regionName
* @return true if the named region is in a transition state * @return true if the named region is in a transition state
*/ */
public boolean regionIsInTransition(byte[] regionName) { public boolean regionIsInTransition(String regionName) {
return regionsInTransition.containsKey(regionName); return regionsInTransition.containsKey(regionName);
} }
@ -731,7 +734,7 @@ class RegionManager implements HConstants {
* @param regionName * @param regionName
* @return true if the region is unassigned, pendingOpen or open * @return true if the region is unassigned, pendingOpen or open
*/ */
public boolean regionIsOpening(byte[] regionName) { public boolean regionIsOpening(String regionName) {
RegionState state = regionsInTransition.get(regionName); RegionState state = regionsInTransition.get(regionName);
if (state != null) { if (state != null) {
return state.isOpening(); return state.isOpening();
@ -746,10 +749,10 @@ class RegionManager implements HConstants {
*/ */
public void setUnassigned(HRegionInfo info, boolean force) { public void setUnassigned(HRegionInfo info, boolean force) {
synchronized(this.regionsInTransition) { synchronized(this.regionsInTransition) {
RegionState s = regionsInTransition.get(info.getRegionName()); RegionState s = regionsInTransition.get(info.getRegionNameAsString());
if (s == null) { if (s == null) {
s = new RegionState(info); s = new RegionState(info);
regionsInTransition.put(info.getRegionName(), s); regionsInTransition.put(info.getRegionNameAsString(), s);
} }
if (force || (!s.isPendingOpen() && !s.isOpen())) { if (force || (!s.isPendingOpen() && !s.isOpen())) {
s.setUnassigned(); s.setUnassigned();
@ -766,7 +769,7 @@ class RegionManager implements HConstants {
*/ */
public boolean isUnassigned(HRegionInfo info) { public boolean isUnassigned(HRegionInfo info) {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = regionsInTransition.get(info.getRegionName()); RegionState s = regionsInTransition.get(info.getRegionNameAsString());
if (s != null) { if (s != null) {
return s.isUnassigned(); return s.isUnassigned();
} }
@ -781,7 +784,7 @@ class RegionManager implements HConstants {
* @param regionName name of the region * @param regionName name of the region
* @return true if open, false otherwise * @return true if open, false otherwise
*/ */
public boolean isPendingOpen(byte [] regionName) { public boolean isPendingOpen(String regionName) {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = regionsInTransition.get(regionName); RegionState s = regionsInTransition.get(regionName);
if (s != null) { if (s != null) {
@ -795,7 +798,7 @@ class RegionManager implements HConstants {
* Region has been assigned to a server and the server has told us it is open * Region has been assigned to a server and the server has told us it is open
* @param regionName * @param regionName
*/ */
public void setOpen(byte [] regionName) { public void setOpen(String regionName) {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = regionsInTransition.get(regionName); RegionState s = regionsInTransition.get(regionName);
if (s != null) { if (s != null) {
@ -808,7 +811,7 @@ class RegionManager implements HConstants {
* @param regionName * @param regionName
* @return true if region is marked to be offlined. * @return true if region is marked to be offlined.
*/ */
public boolean isOfflined(byte[] regionName) { public boolean isOfflined(String regionName) {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = regionsInTransition.get(regionName); RegionState s = regionsInTransition.get(regionName);
if (s != null) { if (s != null) {
@ -827,12 +830,13 @@ class RegionManager implements HConstants {
public void setClosing(final String serverName, final HRegionInfo regionInfo, public void setClosing(final String serverName, final HRegionInfo regionInfo,
final boolean setOffline) { final boolean setOffline) {
synchronized (this.regionsInTransition) { synchronized (this.regionsInTransition) {
RegionState s = this.regionsInTransition.get(regionInfo.getRegionName()); RegionState s =
this.regionsInTransition.get(regionInfo.getRegionNameAsString());
if (s == null) { if (s == null) {
s = new RegionState(regionInfo); s = new RegionState(regionInfo);
} }
s.setClosing(serverName, setOffline); s.setClosing(serverName, setOffline);
this.regionsInTransition.put(regionInfo.getRegionName(), s); this.regionsInTransition.put(regionInfo.getRegionNameAsString(), s);
} }
} }
@ -861,7 +865,7 @@ class RegionManager implements HConstants {
* *
* @param regionName * @param regionName
*/ */
public void setPendingClose(byte[] regionName) { public void setPendingClose(String regionName) {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = regionsInTransition.get(regionName); RegionState s = regionsInTransition.get(regionName);
if (s != null) { if (s != null) {
@ -873,7 +877,7 @@ class RegionManager implements HConstants {
/** /**
* @param regionName * @param regionName
*/ */
public void setClosed(byte[] regionName) { public void setClosed(String regionName) {
synchronized (regionsInTransition) { synchronized (regionsInTransition) {
RegionState s = regionsInTransition.get(regionName); RegionState s = regionsInTransition.get(regionName);
if (s != null) { if (s != null) {
@ -1117,8 +1121,7 @@ class RegionManager implements HConstants {
SortedMap<byte[], Pair<HRegionInfo,HServerAddress>> map, SortedMap<byte[], Pair<HRegionInfo,HServerAddress>> map,
final HMsg.Type msg) { final HMsg.Type msg) {
HServerAddress addr = serverInfo.getServerAddress(); HServerAddress addr = serverInfo.getServerAddress();
Iterator<Pair<HRegionInfo, HServerAddress>> i = Iterator<Pair<HRegionInfo, HServerAddress>> i = map.values().iterator();
map.values().iterator();
synchronized (map) { synchronized (map) {
while (i.hasNext()) { while (i.hasNext()) {
Pair<HRegionInfo,HServerAddress> pair = i.next(); Pair<HRegionInfo,HServerAddress> pair = i.next();

196
src/java/org/apache/hadoop/hbase/master/ServerManager.java
View File

@ -20,7 +20,6 @@
package org.apache.hadoop.hbase.master; package org.apache.hadoop.hbase.master;
import java.io.IOException; import java.io.IOException;
import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.SortedMap; import java.util.SortedMap;
import java.util.TreeMap; import java.util.TreeMap;
@ -41,6 +40,7 @@ import org.apache.hadoop.hbase.HMsg;
import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.Leases;
import org.apache.hadoop.hbase.HMsg.Type; import org.apache.hadoop.hbase.HMsg.Type;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWrapper; import org.apache.hadoop.hbase.zookeeper.ZooKeeperWrapper;
import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.WatchedEvent;
@ -71,11 +71,13 @@ class ServerManager implements HConstants {
/** /**
* Set of known dead servers. On znode expiration, servers are added here. * Set of known dead servers. On znode expiration, servers are added here.
* Boolean holds whether its logs have been split or not. Initially set to * This is needed in case of a network partitioning where the server's lease
* false. * expires, but the server is still running. After the network is healed,
* and it's server logs are recovered, it will be told to call server startup
* because by then, its regions have probably been reassigned.
*/ */
private final Map<String, Boolean> deadServers = private final Set<String> deadServers =
new ConcurrentHashMap<String, Boolean>(); Collections.synchronizedSet(new HashSet<String>());
/** SortedMap server load -> Set of server names */ /** SortedMap server load -> Set of server names */
final SortedMap<HServerLoad, Set<String>> loadToServers = final SortedMap<HServerLoad, Set<String>> loadToServers =
@ -108,64 +110,37 @@ class ServerManager implements HConstants {
getInt("hbase.regions.nobalancing.count", 4); getInt("hbase.regions.nobalancing.count", 4);
} }
/**
* Look to see if we have ghost references to this regionserver such as
* if regionserver is on the dead servers list getting its logs processed.
* @param serverInfo
* @return True if still ghost references and we have not been able to clear
* them or the server is shutting down.
*/
private boolean checkForGhostReferences(final HServerInfo serverInfo) {
boolean result = false;
for (long sleepTime = -1; !master.closed.get() && !result;) {
if (sleepTime != -1) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
// Continue
}
}
// May be on list of dead servers. If so, wait till we've cleared it.
String addr = serverInfo.getServerAddress().toString();
if (isDead(addr)) {
LOG.debug("Waiting on " + addr + " removal from dead list before " +
"processing report-for-duty request");
sleepTime = this.master.threadWakeFrequency;
continue;
}
result = true;
}
return result;
}
/** /**
* Let the server manager know a new regionserver has come online * Let the server manager know a new regionserver has come online
* @param serverInfo * @param serverInfo
* @throws Leases.LeaseStillHeldException
*/ */
public void regionServerStartup(final HServerInfo serverInfo) { public void regionServerStartup(final HServerInfo serverInfo)
String s = serverInfo.getServerAddress().toString().trim(); throws Leases.LeaseStillHeldException {
Watcher watcher = new ServerExpirer(serverInfo.getServerAddress() HServerInfo info = new HServerInfo(serverInfo);
.toString().trim()); String serverName = HServerInfo.getServerName(info);
zooKeeperWrapper.updateRSLocationGetWatch(serverInfo, watcher); if (serversToServerInfo.containsKey(serverName) ||
deadServers.contains(serverName)) {
LOG.info("Received start message from: " + s); throw new Leases.LeaseStillHeldException(serverName);
if (!checkForGhostReferences(serverInfo)) {
return;
} }
Watcher watcher = new ServerExpirer(serverName);
zooKeeperWrapper.updateRSLocationGetWatch(info, watcher);
LOG.info("Received start message from: " + serverName);
// Go on to process the regionserver registration. // Go on to process the regionserver registration.
HServerLoad load = serversToLoad.remove(s); HServerLoad load = serversToLoad.remove(serverName);
if (load != null) { if (load != null) {
// The startup message was from a known server. // The startup message was from a known server.
// Remove stale information about the server's load. // Remove stale information about the server's load.
synchronized (loadToServers) { synchronized (loadToServers) {
Set<String> servers = loadToServers.get(load); Set<String> servers = loadToServers.get(load);
if (servers != null) { if (servers != null) {
servers.remove(s); servers.remove(serverName);
loadToServers.put(load, servers); loadToServers.put(load, servers);
} }
} }
} }
HServerInfo storedInfo = serversToServerInfo.remove(s); HServerInfo storedInfo = serversToServerInfo.remove(serverName);
if (storedInfo != null && !master.closed.get()) { if (storedInfo != null && !master.closed.get()) {
// The startup message was from a known server with the same name. // The startup message was from a known server with the same name.
// Timeout the old one right away. // Timeout the old one right away.
@ -184,15 +159,15 @@ class ServerManager implements HConstants {
} }
// record new server // record new server
load = new HServerLoad(); load = new HServerLoad();
serverInfo.setLoad(load); info.setLoad(load);
serversToServerInfo.put(s, serverInfo); serversToServerInfo.put(serverName, info);
serversToLoad.put(s, load); serversToLoad.put(serverName, load);
synchronized (loadToServers) { synchronized (loadToServers) {
Set<String> servers = loadToServers.get(load); Set<String> servers = loadToServers.get(load);
if (servers == null) { if (servers == null) {
servers = new HashSet<String>(); servers = new HashSet<String>();
} }
servers.add(s); servers.add(serverName);
loadToServers.put(load, servers); loadToServers.put(load, servers);
} }
} }
@ -213,13 +188,16 @@ class ServerManager implements HConstants {
public HMsg [] regionServerReport(final HServerInfo serverInfo, public HMsg [] regionServerReport(final HServerInfo serverInfo,
final HMsg msgs[], final HRegionInfo[] mostLoadedRegions) final HMsg msgs[], final HRegionInfo[] mostLoadedRegions)
throws IOException { throws IOException {
String serverName = serverInfo.getServerAddress().toString().trim(); HServerInfo info = new HServerInfo(serverInfo);
if (isDead(info.getServerName())) {
throw new Leases.LeaseStillHeldException(info.getServerName());
}
if (msgs.length > 0) { if (msgs.length > 0) {
if (msgs[0].isType(HMsg.Type.MSG_REPORT_EXITING)) { if (msgs[0].isType(HMsg.Type.MSG_REPORT_EXITING)) {
processRegionServerExit(serverName, msgs); processRegionServerExit(info, msgs);
return EMPTY_HMSG_ARRAY; return EMPTY_HMSG_ARRAY;
} else if (msgs[0].isType(HMsg.Type.MSG_REPORT_QUIESCED)) { } else if (msgs[0].isType(HMsg.Type.MSG_REPORT_QUIESCED)) {
LOG.info("Region server " + serverName + " quiesced"); LOG.info("Region server " + info.getServerName() + " quiesced");
quiescedServers.incrementAndGet(); quiescedServers.incrementAndGet();
} }
} }
@ -251,16 +229,17 @@ class ServerManager implements HConstants {
return new HMsg [] {REGIONSERVER_STOP}; return new HMsg [] {REGIONSERVER_STOP};
} }
HServerInfo storedInfo = serversToServerInfo.get(serverName); HServerInfo storedInfo = serversToServerInfo.get(info.getServerName());
if (storedInfo == null) { if (storedInfo == null) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("received server report from unknown server: " + serverName); LOG.debug("received server report from unknown server: " +
info.getServerName());
} }
// The HBaseMaster may have been restarted. // The HBaseMaster may have been restarted.
// Tell the RegionServer to start over and call regionServerStartup() // Tell the RegionServer to start over and call regionServerStartup()
return new HMsg[]{CALL_SERVER_STARTUP}; return new HMsg[]{CALL_SERVER_STARTUP};
} else if (storedInfo.getStartCode() != serverInfo.getStartCode()) { } else if (storedInfo.getStartCode() != info.getStartCode()) {
// This state is reachable if: // This state is reachable if:
// //
// 1) RegionServer A started // 1) RegionServer A started
@ -271,35 +250,37 @@ class ServerManager implements HConstants {
// The answer is to ask A to shut down for good. // The answer is to ask A to shut down for good.
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("region server race condition detected: " + serverName); LOG.debug("region server race condition detected: " +
info.getServerName());
} }
synchronized (serversToServerInfo) { synchronized (serversToServerInfo) {
removeServerInfo(serverName); removeServerInfo(info.getServerName());
serversToServerInfo.notifyAll(); serversToServerInfo.notifyAll();
} }
return new HMsg[]{REGIONSERVER_STOP}; return new HMsg[]{REGIONSERVER_STOP};
} else { } else {
return processRegionServerAllsWell(serverName, serverInfo, return processRegionServerAllsWell(info, mostLoadedRegions, msgs);
mostLoadedRegions, msgs);
} }
} }
/** Region server is exiting */ /** Region server is exiting */
private void processRegionServerExit(String serverName, HMsg[] msgs) { private void processRegionServerExit(HServerInfo serverInfo, HMsg[] msgs) {
synchronized (serversToServerInfo) { synchronized (serversToServerInfo) {
try { try {
// HRegionServer is shutting down. // HRegionServer is shutting down.
if (removeServerInfo(serverName)) { if (removeServerInfo(serverInfo.getServerName())) {
// Only process the exit message if the server still has registered info. // Only process the exit message if the server still has registered info.
// Otherwise we could end up processing the server exit twice. // Otherwise we could end up processing the server exit twice.
LOG.info("Region server " + serverName + ": MSG_REPORT_EXITING"); LOG.info("Region server " + serverInfo.getServerName() +
": MSG_REPORT_EXITING");
// Get all the regions the server was serving reassigned // Get all the regions the server was serving reassigned
// (if we are not shutting down). // (if we are not shutting down).
if (!master.closed.get()) { if (!master.closed.get()) {
for (int i = 1; i < msgs.length; i++) { for (int i = 1; i < msgs.length; i++) {
LOG.info("Processing " + msgs[i] + " from " + serverName); LOG.info("Processing " + msgs[i] + " from " +
serverInfo.getServerName());
HRegionInfo info = msgs[i].getRegionInfo(); HRegionInfo info = msgs[i].getRegionInfo();
synchronized (master.regionManager) { synchronized (master.regionManager) {
if (info.isRootRegion()) { if (info.isRootRegion()) {
@ -308,7 +289,8 @@ class ServerManager implements HConstants {
if (info.isMetaTable()) { if (info.isMetaTable()) {
master.regionManager.offlineMetaRegion(info.getStartKey()); master.regionManager.offlineMetaRegion(info.getStartKey());
} }
if (!master.regionManager.isOfflined(info.getRegionName())) { if (!master.regionManager.isOfflined(
info.getRegionNameAsString())) {
master.regionManager.setUnassigned(info, true); master.regionManager.setUnassigned(info, true);
} else { } else {
master.regionManager.removeRegion(info); master.regionManager.removeRegion(info);
@ -328,21 +310,19 @@ class ServerManager implements HConstants {
/** /**
* RegionServer is checking in, no exceptional circumstances * RegionServer is checking in, no exceptional circumstances
* @param serverName
* @param serverInfo * @param serverInfo
* @param mostLoadedRegions * @param mostLoadedRegions
* @param msgs * @param msgs
* @return * @return
* @throws IOException * @throws IOException
*/ */
private HMsg[] processRegionServerAllsWell(String serverName, private HMsg[] processRegionServerAllsWell(HServerInfo serverInfo, HRegionInfo[] mostLoadedRegions, HMsg[] msgs)
HServerInfo serverInfo, HRegionInfo[] mostLoadedRegions, HMsg[] msgs)
throws IOException { throws IOException {
// Refresh the info object and the load information // Refresh the info object and the load information
serversToServerInfo.put(serverName, serverInfo); serversToServerInfo.put(serverInfo.getServerName(), serverInfo);
HServerLoad load = serversToLoad.get(serverName); HServerLoad load = serversToLoad.get(serverInfo.getServerName());
if (load != null) { if (load != null) {
this.master.getMetrics().incrementRequests(load.getNumberOfRequests()); this.master.getMetrics().incrementRequests(load.getNumberOfRequests());
if (!load.equals(serverInfo.getLoad())) { if (!load.equals(serverInfo.getLoad())) {
@ -352,7 +332,7 @@ class ServerManager implements HConstants {
Set<String> servers = loadToServers.get(load); Set<String> servers = loadToServers.get(load);
// Note that servers should never be null because loadToServers // Note that servers should never be null because loadToServers
// and serversToLoad are manipulated in pairs // and serversToLoad are manipulated in pairs
servers.remove(serverName); servers.remove(serverInfo.getServerName());
loadToServers.put(load, servers); loadToServers.put(load, servers);
} }
} }
@ -360,18 +340,18 @@ class ServerManager implements HConstants {
// Set the current load information // Set the current load information
load = serverInfo.getLoad(); load = serverInfo.getLoad();
serversToLoad.put(serverName, load); serversToLoad.put(serverInfo.getServerName(), load);
synchronized (loadToServers) { synchronized (loadToServers) {
Set<String> servers = loadToServers.get(load); Set<String> servers = loadToServers.get(load);
if (servers == null) { if (servers == null) {
servers = new HashSet<String>(); servers = new HashSet<String>();
} }
servers.add(serverName); servers.add(serverInfo.getServerName());
loadToServers.put(load, servers); loadToServers.put(load, servers);
} }
// Next, process messages for this server // Next, process messages for this server
return processMsgs(serverName, serverInfo, mostLoadedRegions, msgs); return processMsgs(serverInfo, mostLoadedRegions, msgs);
} }
/** /**
@ -380,7 +360,7 @@ class ServerManager implements HConstants {
* Note that we never need to update the server's load information because * Note that we never need to update the server's load information because
* that has already been done in regionServerReport. * that has already been done in regionServerReport.
*/ */
private HMsg[] processMsgs(String serverName, HServerInfo serverInfo, private HMsg[] processMsgs(HServerInfo serverInfo,
HRegionInfo[] mostLoadedRegions, HMsg incomingMsgs[]) HRegionInfo[] mostLoadedRegions, HMsg incomingMsgs[])
throws IOException { throws IOException {
ArrayList<HMsg> returnMsgs = new ArrayList<HMsg>(); ArrayList<HMsg> returnMsgs = new ArrayList<HMsg>();
@ -392,14 +372,15 @@ class ServerManager implements HConstants {
int openingCount = 0; int openingCount = 0;
for (int i = 0; i < incomingMsgs.length; i++) { for (int i = 0; i < incomingMsgs.length; i++) {
HRegionInfo region = incomingMsgs[i].getRegionInfo(); HRegionInfo region = incomingMsgs[i].getRegionInfo();
LOG.info("Received " + incomingMsgs[i] + " from " + serverName); LOG.info("Received " + incomingMsgs[i] + " from " +
serverInfo.getServerName());
switch (incomingMsgs[i].getType()) { switch (incomingMsgs[i].getType()) {
case MSG_REPORT_PROCESS_OPEN: case MSG_REPORT_PROCESS_OPEN:
openingCount++; openingCount++;
break; break;
case MSG_REPORT_OPEN: case MSG_REPORT_OPEN:
processRegionOpen(serverName, serverInfo, region, returnMsgs); processRegionOpen(serverInfo, region, returnMsgs);
break; break;
case MSG_REPORT_CLOSE: case MSG_REPORT_CLOSE:
@ -407,8 +388,8 @@ class ServerManager implements HConstants {
break; break;
case MSG_REPORT_SPLIT: case MSG_REPORT_SPLIT:
processSplitRegion(serverName, serverInfo, region, incomingMsgs[++i], processSplitRegion(region, incomingMsgs[++i], incomingMsgs[++i],
incomingMsgs[++i], returnMsgs); returnMsgs);
break; break;
default: default:
@ -420,11 +401,12 @@ class ServerManager implements HConstants {
synchronized (master.regionManager) { synchronized (master.regionManager) {
// Tell the region server to close regions that we have marked for closing. // Tell the region server to close regions that we have marked for closing.
for (HRegionInfo i: master.regionManager.getMarkedToClose(serverName)) { for (HRegionInfo i:
master.regionManager.getMarkedToClose(serverInfo.getServerName())) {
returnMsgs.add(new HMsg(HMsg.Type.MSG_REGION_CLOSE, i, returnMsgs.add(new HMsg(HMsg.Type.MSG_REGION_CLOSE, i,
master.regionManager.inSafeMode())); master.regionManager.inSafeMode()));
// Transition the region from toClose to closing state // Transition the region from toClose to closing state
master.regionManager.setPendingClose(i.getRegionName()); master.regionManager.setPendingClose(i.getRegionNameAsString());
} }
// Figure out what the RegionServer ought to do, and write back. // Figure out what the RegionServer ought to do, and write back.
@ -432,8 +414,8 @@ class ServerManager implements HConstants {
// Should we tell it close regions because its overloaded? If its // Should we tell it close regions because its overloaded? If its
// currently opening regions, leave it alone till all are open. // currently opening regions, leave it alone till all are open.
if (openingCount < this.nobalancingCount) { if (openingCount < this.nobalancingCount) {
this.master.regionManager.assignRegions(serverInfo, serverName, this.master.regionManager.assignRegions(serverInfo, mostLoadedRegions,
mostLoadedRegions, returnMsgs); returnMsgs);
} }
// Send any pending table actions. // Send any pending table actions.
this.master.regionManager.applyActions(serverInfo, returnMsgs); this.master.regionManager.applyActions(serverInfo, returnMsgs);
@ -444,15 +426,13 @@ class ServerManager implements HConstants {
/** /**
* A region has split. * A region has split.
* *
* @param serverName
* @param serverInfo
* @param region * @param region
* @param splitA * @param splitA
* @param splitB * @param splitB
* @param returnMsgs * @param returnMsgs
*/ */
private void processSplitRegion(String serverName, HServerInfo serverInfo, private void processSplitRegion(HRegionInfo region, HMsg splitA, HMsg splitB,
HRegionInfo region, HMsg splitA, HMsg splitB, ArrayList<HMsg> returnMsgs) { ArrayList<HMsg> returnMsgs) {
synchronized (master.regionManager) { synchronized (master.regionManager) {
// Cancel any actions pending for the affected region. // Cancel any actions pending for the affected region.
@ -475,18 +455,18 @@ class ServerManager implements HConstants {
} }
/** Region server is reporting that a region is now opened */ /** Region server is reporting that a region is now opened */
private void processRegionOpen(String serverName, HServerInfo serverInfo, private void processRegionOpen(HServerInfo serverInfo,
HRegionInfo region, ArrayList<HMsg> returnMsgs) HRegionInfo region, ArrayList<HMsg> returnMsgs)
throws IOException { throws IOException {
boolean duplicateAssignment = false; boolean duplicateAssignment = false;
synchronized (master.regionManager) { synchronized (master.regionManager) {
if (!master.regionManager.isUnassigned(region) && if (!master.regionManager.isUnassigned(region) &&
!master.regionManager.isPendingOpen(region.getRegionName())) { !master.regionManager.isPendingOpen(region.getRegionNameAsString())) {
if (region.isRootRegion()) { if (region.isRootRegion()) {
// Root region // Root region
HServerAddress rootServer = master.getRootRegionLocation(); HServerAddress rootServer = master.getRootRegionLocation();
if (rootServer != null) { if (rootServer != null) {
if (rootServer.toString().compareTo(serverName) == 0) { if (rootServer.compareTo(serverInfo.getServerAddress()) == 0) {
// A duplicate open report from the correct server // A duplicate open report from the correct server
return; return;
} }
@ -498,7 +478,8 @@ class ServerManager implements HConstants {
// Not root region. If it is not a pending region, then we are // Not root region. If it is not a pending region, then we are
// going to treat it as a duplicate assignment, although we can't // going to treat it as a duplicate assignment, although we can't
// tell for certain that's the case. // tell for certain that's the case.
if (master.regionManager.isPendingOpen(region.getRegionName())) { if (master.regionManager.isPendingOpen(
region.getRegionNameAsString())) {
// A duplicate report from the correct server // A duplicate report from the correct server
return; return;
} }
@ -535,7 +516,7 @@ class ServerManager implements HConstants {
} else { } else {
// Note that the table has been assigned and is waiting for the // Note that the table has been assigned and is waiting for the
// meta table to be updated. // meta table to be updated.
master.regionManager.setOpen(region.getRegionName()); master.regionManager.setOpen(region.getRegionNameAsString());
// Queue up an update to note the region location. // Queue up an update to note the region location.
try { try {
master.toDoQueue.put( master.toDoQueue.put(
@ -567,7 +548,7 @@ class ServerManager implements HConstants {
} }
boolean offlineRegion = boolean offlineRegion =
master.regionManager.isOfflined(region.getRegionName()); master.regionManager.isOfflined(region.getRegionNameAsString());
boolean reassignRegion = !region.isOffline() && !offlineRegion; boolean reassignRegion = !region.isOffline() && !offlineRegion;
// NOTE: If the region was just being closed and not offlined, we cannot // NOTE: If the region was just being closed and not offlined, we cannot
@ -575,7 +556,7 @@ class ServerManager implements HConstants {
// the messages we've received. In this case, a close could be // the messages we've received. In this case, a close could be
// processed before an open resulting in the master not agreeing on // processed before an open resulting in the master not agreeing on
// the region's state. // the region's state.
master.regionManager.setClosed(region.getRegionName()); master.regionManager.setClosed(region.getRegionNameAsString());
try { try {
master.toDoQueue.put(new ProcessRegionClose(master, region, master.toDoQueue.put(new ProcessRegionClose(master, region,
offlineRegion, reassignRegion)); offlineRegion, reassignRegion));
@ -648,11 +629,11 @@ class ServerManager implements HConstants {
} }
/** /**
* @param address server address * @param name server name
* @return HServerInfo for the given server address * @return HServerInfo for the given server address
*/ */
public HServerInfo getServerInfo(String address) { public HServerInfo getServerInfo(String name) {
return serversToServerInfo.get(address); return serversToServerInfo.get(name);
} }
/** /**
@ -660,7 +641,7 @@ class ServerManager implements HConstants {
*/ */
public Map<String, HServerInfo> getServersToServerInfo() { public Map<String, HServerInfo> getServersToServerInfo() {
synchronized (serversToServerInfo) { synchronized (serversToServerInfo) {
return new HashMap<String, HServerInfo>(serversToServerInfo); return Collections.unmodifiableMap(serversToServerInfo);
} }
} }
@ -669,16 +650,7 @@ class ServerManager implements HConstants {
*/ */
public Map<String, HServerLoad> getServersToLoad() { public Map<String, HServerLoad> getServersToLoad() {
synchronized (serversToLoad) { synchronized (serversToLoad) {
return new HashMap<String, HServerLoad>(serversToLoad); return Collections.unmodifiableMap(serversToLoad);
}
}
/**
* @return Read-only map of load to servers.
*/
public Map<HServerLoad, Set<String>> getLoadToServers() {
synchronized (loadToServers) {
return new HashMap<HServerLoad, Set<String>>(loadToServers);
} }
} }
@ -740,7 +712,7 @@ class ServerManager implements HConstants {
master.regionManager.unsetRootRegion(); master.regionManager.unsetRootRegion();
rootServer = true; rootServer = true;
} }
String serverName = info.getServerAddress().toString(); String serverName = HServerInfo.getServerName(info);
HServerLoad load = serversToLoad.remove(serverName); HServerLoad load = serversToLoad.remove(serverName);
if (load != null) { if (load != null) {
synchronized (loadToServers) { synchronized (loadToServers) {
@ -751,7 +723,7 @@ class ServerManager implements HConstants {
} }
} }
} }
deadServers.put(server, Boolean.FALSE); deadServers.add(server);
try { try {
master.toDoQueue.put(new ProcessServerShutdown(master, info, master.toDoQueue.put(new ProcessServerShutdown(master, info,
rootServer)); rootServer));
@ -778,6 +750,6 @@ class ServerManager implements HConstants {
* @return true if server is dead * @return true if server is dead
*/ */
public boolean isDead(String serverName) { public boolean isDead(String serverName) {
return deadServers.containsKey(serverName); return deadServers.contains(serverName);
} }
} }

4
src/java/org/apache/hadoop/hbase/master/TableDelete.java
View File

@ -41,8 +41,8 @@ class TableDelete extends TableOperation {
} }
@Override @Override
protected void processScanItem(String serverName, protected void processScanItem(@SuppressWarnings("unused") String serverName,
long startCode, final HRegionInfo info) throws IOException { final HRegionInfo info) throws IOException {
if (isEnabled(info)) { if (isEnabled(info)) {
throw new TableNotDisabledException(tableName); throw new TableNotDisabledException(tableName);

20
src/java/org/apache/hadoop/hbase/master/TableOperation.java
View File

@ -92,17 +92,21 @@ abstract class TableOperation implements HConstants {
LOG.error(COL_REGIONINFO + " not found on " + values.getRow()); LOG.error(COL_REGIONINFO + " not found on " + values.getRow());
continue; continue;
} }
String serverName = Writables.cellToString(values.get(COL_SERVER)); String serverAddress = Writables.cellToString(values.get(COL_SERVER));
long startCode = Writables.cellToLong(values.get(COL_STARTCODE)); long startCode = Writables.cellToLong(values.get(COL_STARTCODE));
String serverName = null;
if (serverAddress != null && serverAddress.length() > 0) {
serverName = HServerInfo.getServerName(serverAddress, startCode);
}
if (Bytes.compareTo(info.getTableDesc().getName(), tableName) > 0) { if (Bytes.compareTo(info.getTableDesc().getName(), tableName) > 0) {
break; // Beyond any more entries for this table break; // Beyond any more entries for this table
} }
tableExists = true; tableExists = true;
if (!isBeingServed(serverName, startCode) || !isEnabled(info)) { if (!isBeingServed(serverName) || !isEnabled(info)) {
unservedRegions.add(info); unservedRegions.add(info);
} }
processScanItem(serverName, startCode, info); processScanItem(serverName, info);
} }
} finally { } finally {
if (scannerId != -1L) { if (scannerId != -1L) {
@ -145,11 +149,11 @@ abstract class TableOperation implements HConstants {
} }
} }
protected boolean isBeingServed(String serverName, long startCode) { protected boolean isBeingServed(String serverName) {
boolean result = false; boolean result = false;
if (serverName != null && serverName.length() > 0 && startCode != -1L) { if (serverName != null && serverName.length() > 0) {
HServerInfo s = master.serverManager.getServerInfo(serverName); HServerInfo s = master.serverManager.getServerInfo(serverName);
result = s != null && s.getStartCode() == startCode; result = s != null;
} }
return result; return result;
} }
@ -158,8 +162,8 @@ abstract class TableOperation implements HConstants {
return !info.isOffline(); return !info.isOffline();
} }
protected abstract void processScanItem(String serverName, long startCode, protected abstract void processScanItem(String serverName, HRegionInfo info)
HRegionInfo info) throws IOException; throws IOException;
protected abstract void postProcessMeta(MetaRegion m, protected abstract void postProcessMeta(MetaRegion m,
HRegionInterface server) throws IOException; HRegionInterface server) throws IOException;

39
src/java/org/apache/hadoop/hbase/regionserver/HLog.java
View File

@ -22,8 +22,6 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.EOFException; import java.io.EOFException;
import java.io.FileNotFoundException; import java.io.FileNotFoundException;
import java.io.IOException; import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.Collections; import java.util.Collections;
import java.util.Map; import java.util.Map;
import java.util.SortedMap; import java.util.SortedMap;
@ -843,18 +841,35 @@ public class HLog implements HConstants, Syncable {
* @return the HLog directory name * @return the HLog directory name
*/ */
public static String getHLogDirectoryName(HServerInfo info) { public static String getHLogDirectoryName(HServerInfo info) {
StringBuilder dirName = new StringBuilder("log_"); return getHLogDirectoryName(HServerInfo.getServerName(info));
try {
dirName.append(URLEncoder.encode(
info.getServerAddress().getBindAddress(), UTF8_ENCODING));
} catch (UnsupportedEncodingException e) {
LOG.error("Error encoding '" + info.getServerAddress().getBindAddress()
+ "'", e);
} }
/**
* Construct the HLog directory name
*
* @param serverAddress
* @param startCode
* @return the HLog directory name
*/
public static String getHLogDirectoryName(String serverAddress,
long startCode) {
if (serverAddress == null || serverAddress.length() == 0) {
return null;
}
return getHLogDirectoryName(
HServerInfo.getServerName(serverAddress, startCode));
}
/**
* Construct the HLog directory name
*
* @param serverName
* @return the HLog directory name
*/
public static String getHLogDirectoryName(String serverName) {
StringBuilder dirName = new StringBuilder(HConstants.HREGION_LOGDIR_NAME);
dirName.append("_"); dirName.append("_");
dirName.append(info.getStartCode()); dirName.append(serverName);
dirName.append("_");
dirName.append(info.getServerAddress().getPort());
return dirName.toString(); return dirName.toString();
} }