You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ji...@apache.org on 2007/06/11 18:46:31 UTC
svn commit: r546192 [1/3] - in /lucene/hadoop/trunk/src/contrib/hbase: ./
conf/ src/java/org/apache/hadoop/hbase/ src/test/org/apache/hadoop/hbase/
Author: jimk
Date: Mon Jun 11 09:46:27 2007
New Revision: 546192
URL: http://svn.apache.org/viewvc?view=rev&rev=546192
Log:
HADOOP-1421 HADOOP-1466 When a region server dies, its log file must be split up on a per region basis
so that region servers are assigned the regions have a log to apply edits from. Enhance fail over
capabilities.
For all the files modified, clean up javadoc, class method and field visibility.
Added:
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/KeyedData.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/RegionServerRunningException.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestCleanRegionServerExit.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestRegionServerAbort.java
Removed:
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/LabelledData.java
Modified:
lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt
lucene/hadoop/trunk/src/contrib/hbase/conf/hbase-default.xml
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HAbstractScanner.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HClient.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLogKey.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMerge.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMsg.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInterface.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegiondirReader.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStoreKey.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HTableDescriptor.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/LeaseListener.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/Leases.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/AbstractMergeTestBase.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/HBaseClusterTestCase.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MiniHBaseCluster.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestGet.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestHLog.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestHMemcache.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestHRegion.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestScanner.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestScanner2.java
Modified: lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt Mon Jun 11 09:46:27 2007
@@ -25,3 +25,6 @@
13. HADOOP-1445 Support updates across region splits and compactions
14. HADOOP-1460 On shutdown IOException with complaint 'Cannot cancel lease
that is not held'
+ 15. HADOOP-1421 Failover detection, split log files.
+ For the files modified, also clean up javadoc, class, field and method
+ visibility (HADOOP-1466)
Modified: lucene/hadoop/trunk/src/contrib/hbase/conf/hbase-default.xml
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/conf/hbase-default.xml?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/conf/hbase-default.xml (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/conf/hbase-default.xml Mon Jun 11 09:46:27 2007
@@ -15,7 +15,7 @@
</description>
</property>
<property>
- <name>hbase.regiondir</name>
+ <name>hbase.rootdir</name>
<value>${hadoop.tmp.dir}/hbase</value>
<description>The directory shared by region servers.
</description>
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HAbstractScanner.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HAbstractScanner.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HAbstractScanner.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HAbstractScanner.java Mon Jun 11 09:46:27 2007
@@ -42,9 +42,12 @@
// The kind of match we are doing on a column:
private static enum MATCH_TYPE {
- FAMILY_ONLY, // Just check the column family name
- REGEX, // Column family + matches regex
- SIMPLE // Literal matching
+ /** Just check the column family name */
+ FAMILY_ONLY,
+ /** Column family + matches regex */
+ REGEX,
+ /** Literal matching */
+ SIMPLE
}
// This class provides column matching functions that are more sophisticated
@@ -63,12 +66,12 @@
ColumnMatcher(Text col) throws IOException {
String column = col.toString();
try {
- int colpos = column.indexOf(":") + 1;
- if(colpos == 0) {
+ int colpos = column.indexOf(":");
+ if(colpos == -1) {
throw new InvalidColumnNameException("Column name has no family indicator.");
}
- String columnkey = column.substring(colpos);
+ String columnkey = column.substring(colpos + 1);
if(columnkey == null || columnkey.length() == 0) {
this.matchType = MATCH_TYPE.FAMILY_ONLY;
@@ -97,7 +100,7 @@
return c.equals(this.col);
} else if(this.matchType == MATCH_TYPE.FAMILY_ONLY) {
- return c.toString().startsWith(this.family);
+ return HStoreKey.extractFamily(c).toString().equals(this.family);
} else if(this.matchType == MATCH_TYPE.REGEX) {
return this.columnMatcher.matcher(c.toString()).matches();
@@ -211,6 +214,7 @@
* @param key The key that matched
* @param results All the results for <code>key</code>
* @return true if a match was found
+ * @throws IOException
*
* @see org.apache.hadoop.hbase.HScannerInterface#next(org.apache.hadoop.hbase.HStoreKey, java.util.TreeMap)
*/
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HClient.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HClient.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HClient.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HClient.java Mon Jun 11 09:46:27 2007
@@ -53,7 +53,7 @@
COL_REGIONINFO
};
- private static final Text EMPTY_START_ROW = new Text();
+ static final Text EMPTY_START_ROW = new Text();
long pause;
int numRetries;
@@ -64,8 +64,8 @@
* Data structure that holds current location for a region and its info.
*/
static class RegionLocation {
- public HRegionInfo regionInfo;
- public HServerAddress serverAddress;
+ HRegionInfo regionInfo;
+ HServerAddress serverAddress;
RegionLocation(HRegionInfo regionInfo, HServerAddress serverAddress) {
this.regionInfo = regionInfo;
@@ -83,7 +83,7 @@
private TreeMap<Text, SortedMap<Text, RegionLocation>> tablesToServers;
// For the "current" table: Map startRow -> (HRegionInfo, HServerAddress)
- private SortedMap<Text, RegionLocation> tableServers;
+ SortedMap<Text, RegionLocation> tableServers;
// Known region HServerAddress.toString() -> HRegionInterface
private TreeMap<String, HRegionInterface> servers;
@@ -95,7 +95,10 @@
Random rand;
long clientid;
- /** Creates a new HClient */
+ /**
+ * Creates a new HClient
+ * @param conf - Configuration object
+ */
public HClient(Configuration conf) {
this.conf = conf;
@@ -239,6 +242,12 @@
}
}
+ /**
+ * Deletes a table
+ *
+ * @param tableName - name of table to delete
+ * @throws IOException
+ */
public synchronized void deleteTable(Text tableName) throws IOException {
checkReservedTableName(tableName);
checkMaster();
@@ -254,23 +263,21 @@
HRegionInterface server =
getHRegionConnection(firstMetaServer.serverAddress);
DataInputBuffer inbuf = new DataInputBuffer();
- HStoreKey key = new HStoreKey();
HRegionInfo info = new HRegionInfo();
for (int tries = 0; tries < numRetries; tries++) {
long scannerId = -1L;
try {
scannerId = server.openScanner(firstMetaServer.regionInfo.regionName,
REGIONINFO, tableName);
- LabelledData[] values = server.next(scannerId, key);
+ KeyedData[] values = server.next(scannerId);
if(values == null || values.length == 0) {
break;
}
boolean found = false;
for(int j = 0; j < values.length; j++) {
- if(values[j].getLabel().equals(COL_REGIONINFO)) {
+ if(values[j].getKey().getColumn().equals(COL_REGIONINFO)) {
byte[] bytes = new byte[values[j].getData().getSize()];
- System.arraycopy(values[j].getData().get(), 0, bytes, 0,
- bytes.length);
+ System.arraycopy(values[j].getData().get(), 0, bytes, 0, bytes.length);
inbuf.reset(bytes, bytes.length);
info.readFields(inbuf);
if(info.tableDesc.getName().equals(tableName)) {
@@ -301,7 +308,15 @@
LOG.info("table " + tableName + " deleted");
}
- public synchronized void addColumn(Text tableName, HColumnDescriptor column) throws IOException {
+ /**
+ * Add a column to an existing table
+ *
+ * @param tableName - name of the table to add column to
+ * @param column - column descriptor of column to be added
+ * @throws IOException
+ */
+ public synchronized void addColumn(Text tableName, HColumnDescriptor column)
+ throws IOException {
checkReservedTableName(tableName);
checkMaster();
try {
@@ -312,7 +327,15 @@
}
}
- public synchronized void deleteColumn(Text tableName, Text columnName) throws IOException {
+ /**
+ * Delete a column from a table
+ *
+ * @param tableName - name of table
+ * @param columnName - name of column to be deleted
+ * @throws IOException
+ */
+ public synchronized void deleteColumn(Text tableName, Text columnName)
+ throws IOException {
checkReservedTableName(tableName);
checkMaster();
try {
@@ -323,6 +346,12 @@
}
}
+ /**
+ * Brings a table on-line (enables it)
+ *
+ * @param tableName - name of the table
+ * @throws IOException
+ */
public synchronized void enableTable(Text tableName) throws IOException {
checkReservedTableName(tableName);
checkMaster();
@@ -340,7 +369,6 @@
HRegionInterface server = getHRegionConnection(firstMetaServer.serverAddress);
DataInputBuffer inbuf = new DataInputBuffer();
- HStoreKey key = new HStoreKey();
HRegionInfo info = new HRegionInfo();
for(int tries = 0; tries < numRetries; tries++) {
int valuesfound = 0;
@@ -348,21 +376,28 @@
try {
scannerId = server.openScanner(firstMetaServer.regionInfo.regionName,
REGIONINFO, tableName);
- LabelledData[] values = server.next(scannerId, key);
- if(values == null || values.length == 0) {
- if(valuesfound == 0) {
- throw new NoSuchElementException("table " + tableName + " not found");
- }
- }
- valuesfound += 1;
boolean isenabled = false;
- for(int j = 0; j < values.length; j++) {
- if(values[j].getLabel().equals(COL_REGIONINFO)) {
- byte[] bytes = new byte[values[j].getData().getSize()];
- System.arraycopy(values[j].getData().get(), 0, bytes, 0, bytes.length);
- inbuf.reset(bytes, bytes.length);
- info.readFields(inbuf);
- isenabled = !info.offLine;
+ while(true) {
+ KeyedData[] values = server.next(scannerId);
+ if(values == null || values.length == 0) {
+ if(valuesfound == 0) {
+ throw new NoSuchElementException("table " + tableName + " not found");
+ }
+ break;
+ }
+ valuesfound += 1;
+ for(int j = 0; j < values.length; j++) {
+ if(values[j].getKey().getColumn().equals(COL_REGIONINFO)) {
+ byte[] bytes = new byte[values[j].getData().getSize()];
+ System.arraycopy(values[j].getData().get(), 0, bytes, 0, bytes.length);
+ inbuf.reset(bytes, bytes.length);
+ info.readFields(inbuf);
+ isenabled = !info.offLine;
+ break;
+ }
+ }
+ if(isenabled) {
+ break;
}
}
if(isenabled) {
@@ -395,6 +430,13 @@
LOG.info("Enabled table " + tableName);
}
+ /**
+ * Disables a table (takes it off-line) If it is being served, the master
+ * will tell the servers to stop serving it.
+ *
+ * @param tableName - name of table
+ * @throws IOException
+ */
public synchronized void disableTable(Text tableName) throws IOException {
checkReservedTableName(tableName);
checkMaster();
@@ -412,7 +454,6 @@
HRegionInterface server = getHRegionConnection(firstMetaServer.serverAddress);
DataInputBuffer inbuf = new DataInputBuffer();
- HStoreKey key = new HStoreKey();
HRegionInfo info = new HRegionInfo();
for(int tries = 0; tries < numRetries; tries++) {
int valuesfound = 0;
@@ -420,21 +461,28 @@
try {
scannerId = server.openScanner(firstMetaServer.regionInfo.regionName,
REGIONINFO, tableName);
- LabelledData[] values = server.next(scannerId, key);
- if(values == null || values.length == 0) {
- if(valuesfound == 0) {
- throw new NoSuchElementException("table " + tableName + " not found");
- }
- }
- valuesfound += 1;
boolean disabled = false;
- for(int j = 0; j < values.length; j++) {
- if(values[j].getLabel().equals(COL_REGIONINFO)) {
- byte[] bytes = new byte[values[j].getData().getSize()];
- System.arraycopy(values[j].getData().get(), 0, bytes, 0, bytes.length);
- inbuf.reset(bytes, bytes.length);
- info.readFields(inbuf);
- disabled = info.offLine;
+ while(true) {
+ KeyedData[] values = server.next(scannerId);
+ if(values == null || values.length == 0) {
+ if(valuesfound == 0) {
+ throw new NoSuchElementException("table " + tableName + " not found");
+ }
+ break;
+ }
+ valuesfound += 1;
+ for(int j = 0; j < values.length; j++) {
+ if(values[j].getKey().getColumn().equals(COL_REGIONINFO)) {
+ byte[] bytes = new byte[values[j].getData().getSize()];
+ System.arraycopy(values[j].getData().get(), 0, bytes, 0, bytes.length);
+ inbuf.reset(bytes, bytes.length);
+ info.readFields(inbuf);
+ disabled = info.offLine;
+ break;
+ }
+ }
+ if(disabled) {
+ break;
}
}
if(disabled) {
@@ -466,6 +514,10 @@
LOG.info("Disabled table " + tableName);
}
+ /**
+ * Shuts down the HBase instance
+ * @throws IOException
+ */
public synchronized void shutdown() throws IOException {
checkMaster();
this.master.shutdown();
@@ -675,8 +727,8 @@
* @throws IOException
*/
private TreeMap<Text, RegionLocation> scanOneMetaRegion(final RegionLocation t,
- final Text tableName)
- throws IOException {
+ final Text tableName) throws IOException {
+
HRegionInterface server = getHRegionConnection(t.serverAddress);
TreeMap<Text, RegionLocation> servers = new TreeMap<Text, RegionLocation>();
for(int tries = 0; servers.size() == 0 && tries < this.numRetries;
@@ -691,8 +743,7 @@
while(true) {
HRegionInfo regionInfo = null;
String serverAddress = null;
- HStoreKey key = new HStoreKey();
- LabelledData[] values = server.next(scannerId, key);
+ KeyedData[] values = server.next(scannerId);
if(values.length == 0) {
if(servers.size() == 0) {
// If we didn't find any servers then the table does not exist
@@ -713,7 +764,7 @@
for(int i = 0; i < values.length; i++) {
bytes = new byte[values[i].getData().getSize()];
System.arraycopy(values[i].getData().get(), 0, bytes, 0, bytes.length);
- results.put(values[i].getLabel(), bytes);
+ results.put(values[i].getKey().getColumn(), bytes);
}
regionInfo = new HRegionInfo();
bytes = results.get(COL_REGIONINFO);
@@ -808,6 +859,9 @@
* If we wanted this to be really fast, we could implement a special
* catalog table that just contains table names and their descriptors.
* Right now, it only exists as part of the META table's region info.
+ *
+ * @return - returns an array of HTableDescriptors
+ * @throws IOException
*/
public synchronized HTableDescriptor[] listTables()
throws IOException {
@@ -828,15 +882,14 @@
scannerId = server.openScanner(t.regionInfo.regionName,
META_COLUMNS, EMPTY_START_ROW);
- HStoreKey key = new HStoreKey();
DataInputBuffer inbuf = new DataInputBuffer();
while(true) {
- LabelledData[] values = server.next(scannerId, key);
+ KeyedData[] values = server.next(scannerId);
if(values.length == 0) {
break;
}
for(int i = 0; i < values.length; i++) {
- if(values[i].getLabel().equals(COL_REGIONINFO)) {
+ if(values[i].getKey().getColumn().equals(COL_REGIONINFO)) {
byte[] bytes = values[i].getData().get();
inbuf.reset(bytes, bytes.length);
HRegionInfo info = new HRegionInfo();
@@ -901,7 +954,14 @@
}
}
- /** Get a single value for the specified row and column */
+ /**
+ * Get a single value for the specified row and column
+ *
+ * @param row - row key
+ * @param column - column name
+ * @return - value for specified row/column
+ * @throws IOException
+ */
public byte[] get(Text row, Text column) throws IOException {
RegionLocation info = null;
BytesWritable value = null;
@@ -931,7 +991,15 @@
return null;
}
- /** Get the specified number of versions of the specified row and column */
+ /**
+ * Get the specified number of versions of the specified row and column
+ *
+ * @param row - row key
+ * @param column - column name
+ * @param numVersions - number of versions to retrieve
+ * @return - array byte values
+ * @throws IOException
+ */
public byte[][] get(Text row, Text column, int numVersions) throws IOException {
RegionLocation info = null;
BytesWritable[] values = null;
@@ -968,8 +1036,16 @@
/**
* Get the specified number of versions of the specified row and column with
* the specified timestamp.
+ *
+ * @param row - row key
+ * @param column - column name
+ * @param timestamp - timestamp
+ * @param numVersions - number of versions to retrieve
+ * @return - array of values that match the above criteria
+ * @throws IOException
*/
- public byte[][] get(Text row, Text column, long timestamp, int numVersions) throws IOException {
+ public byte[][] get(Text row, Text column, long timestamp, int numVersions)
+ throws IOException {
RegionLocation info = null;
BytesWritable[] values = null;
@@ -1002,10 +1078,16 @@
return null;
}
- /** Get all the data for the specified row */
- public LabelledData[] getRow(Text row) throws IOException {
+ /**
+ * Get all the data for the specified row
+ *
+ * @param row - row key
+ * @return - map of colums to values
+ * @throws IOException
+ */
+ public SortedMap<Text, byte[]> getRow(Text row) throws IOException {
RegionLocation info = null;
- LabelledData[] value = null;
+ KeyedData[] value = null;
for(int tries = 0; tries < numRetries && info == null; tries++) {
info = getRegionLocation(row);
@@ -1023,15 +1105,29 @@
info = null;
}
}
-
- return value;
+ TreeMap<Text, byte[]> results = new TreeMap<Text, byte[]>();
+ if(value != null && value.length != 0) {
+ for(int i = 0; i < value.length; i++) {
+ byte[] bytes = new byte[value[i].getData().getSize()];
+ System.arraycopy(value[i].getData().get(), 0, bytes, 0, bytes.length);
+ results.put(value[i].getKey().getColumn(), bytes);
+ }
+ }
+ return results;
}
/**
* Get a scanner on the current table starting at the specified row.
* Return the specified columns.
+ *
+ * @param columns - array of columns to return
+ * @param startRow - starting row in table to scan
+ * @return - scanner
+ * @throws IOException
*/
- public synchronized HScannerInterface obtainScanner(Text[] columns, Text startRow) throws IOException {
+ public synchronized HScannerInterface obtainScanner(Text[] columns,
+ Text startRow) throws IOException {
+
if(this.tableServers == null) {
throw new IllegalStateException("Must open table first");
}
@@ -1069,9 +1165,20 @@
long startUpdate() throws IOException;
}
- /* Start an atomic row insertion or update
+ /**
+ * Start an atomic row insertion/update. No changes are committed until the
+ * call to commit() returns. A call to abort() will abandon any updates in progress.
+ *
+ * Callers to this method are given a lease for each unique lockid; before the
+ * lease expires, either abort() or commit() must be called. If it is not
+ * called, the system will automatically call abort() on the client's behalf.
+ *
+ * The client can gain extra time with a call to renewLease().
+ * Start an atomic row insertion or update
+ *
* @param row Name of row to start update against.
* @return Row lockid.
+ * @throws IOException
*/
public long startUpdate(final Text row) throws IOException {
// Implemention of the StartUpdate interface.
@@ -1114,7 +1221,14 @@
return retryProxy.startUpdate();
}
- /** Change a value for the specified column */
+ /**
+ * Change a value for the specified column
+ *
+ * @param lockid - lock id returned from startUpdate
+ * @param column - column whose value is being set
+ * @param val - new value for column
+ * @throws IOException
+ */
public void put(long lockid, Text column, byte val[]) throws IOException {
try {
this.currentServer.put(this.currentRegion, this.clientid, lockid, column,
@@ -1131,7 +1245,13 @@
}
}
- /** Delete the value for a column */
+ /**
+ * Delete the value for a column
+ *
+ * @param lockid - lock id returned from startUpdate
+ * @param column - name of column whose value is to be deleted
+ * @throws IOException
+ */
public void delete(long lockid, Text column) throws IOException {
try {
this.currentServer.delete(this.currentRegion, this.clientid, lockid,
@@ -1148,7 +1268,12 @@
}
}
- /** Abort a row mutation */
+ /**
+ * Abort a row mutation
+ *
+ * @param lockid - lock id returned from startUpdate
+ * @throws IOException
+ */
public void abort(long lockid) throws IOException {
try {
this.currentServer.abort(this.currentRegion, this.clientid, lockid);
@@ -1159,7 +1284,12 @@
}
}
- /** Finalize a row mutation */
+ /**
+ * Finalize a row mutation
+ *
+ * @param lockid - lock id returned from startUpdate
+ * @throws IOException
+ */
public void commit(long lockid) throws IOException {
try {
this.currentServer.commit(this.currentRegion, this.clientid, lockid);
@@ -1170,11 +1300,33 @@
}
/**
+ * Renew lease on update
+ *
+ * @param lockid - lock id returned from startUpdate
+ * @throws IOException
+ */
+ public void renewLease(long lockid) throws IOException {
+ try {
+ this.currentServer.renewLease(lockid, this.clientid);
+ } catch(IOException e) {
+ try {
+ this.currentServer.abort(this.currentRegion, this.clientid, lockid);
+ } catch(IOException e2) {
+ LOG.warn(e2);
+ }
+ this.currentServer = null;
+ this.currentRegion = null;
+ throw e;
+ }
+ }
+
+ /**
* Implements the scanner interface for the HBase client.
* If there are multiple regions in a table, this scanner will iterate
* through them all.
*/
private class ClientScanner implements HScannerInterface {
+ private final Text EMPTY_COLUMN = new Text();
private Text[] columns;
private Text startRow;
private boolean closed;
@@ -1198,7 +1350,7 @@
this.regions = info.toArray(new RegionLocation[info.size()]);
}
- public ClientScanner(Text[] columns, Text startRow) throws IOException {
+ ClientScanner(Text[] columns, Text startRow) throws IOException {
this.columns = columns;
this.startRow = startRow;
this.closed = false;
@@ -1260,17 +1412,22 @@
if(this.closed) {
return false;
}
- LabelledData[] values = null;
+ KeyedData[] values = null;
do {
- values = this.server.next(this.scannerId, key);
- } while(values.length == 0 && nextScanner());
+ values = this.server.next(this.scannerId);
+ } while(values != null && values.length == 0 && nextScanner());
- for(int i = 0; i < values.length; i++) {
- byte[] bytes = new byte[values[i].getData().getSize()];
- System.arraycopy(values[i].getData().get(), 0, bytes, 0, bytes.length);
- results.put(values[i].getLabel(), bytes);
+ if(values != null && values.length != 0) {
+ for(int i = 0; i < values.length; i++) {
+ key.setRow(values[i].getKey().getRow());
+ key.setVersion(values[i].getKey().getTimestamp());
+ key.setColumn(EMPTY_COLUMN);
+ byte[] bytes = new byte[values[i].getData().getSize()];
+ System.arraycopy(values[i].getData().get(), 0, bytes, 0, bytes.length);
+ results.put(values[i].getKey().getColumn(), bytes);
+ }
}
- return values.length != 0;
+ return values == null ? false : values.length != 0;
}
/* (non-Javadoc)
@@ -1333,8 +1490,13 @@
" deleteTable testtable");
}
+ /**
+ * Process command-line args.
+ * @param args - command arguments
+ * @return 0 if successful -1 otherwise
+ */
public int doCommandLine(final String args[]) {
- // Process command-line args. TODO: Better cmd-line processing
+ // TODO: Better cmd-line processing
// (but hopefully something not as painful as cli options).
int errCode = -1;
if (args.length < 1) {
@@ -1416,6 +1578,10 @@
return errCode;
}
+ /**
+ * Main program
+ * @param args
+ */
public static void main(final String args[]) {
Configuration c = new HBaseConfiguration();
int errCode = (new HClient(c)).doCommandLine(args);
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java Mon Jun 11 09:46:27 2007
@@ -81,6 +81,12 @@
this.versionNumber = COLUMN_DESCRIPTOR_VERSION;
}
+ /**
+ * Construct a column descriptor specifying only the family name
+ * The other attributes are defaulted.
+ *
+ * @param columnName - column family name
+ */
public HColumnDescriptor(String columnName) {
this();
this.name.set(columnName);
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java Mon Jun 11 09:46:27 2007
@@ -34,22 +34,41 @@
// TODO: Support 'local': i.e. default of all running in single
// process. Same for regionserver. TODO: Is having HBase homed
// on port 60k OK?
+
+ /** Parameter name for master address */
static final String MASTER_ADDRESS = "hbase.master";
+
+ /** Default master address */
static final String DEFAULT_MASTER_ADDRESS = "localhost:60000";
- // Key for hbase.regionserver address.
+ /** Parameter name for hbase.regionserver address. */
static final String REGIONSERVER_ADDRESS = "hbase.regionserver";
+
+ /** Default region server address */
static final String DEFAULT_REGIONSERVER_ADDRESS = "localhost:60010";
+ /** Parameter name for how often threads should wake up */
static final String THREAD_WAKE_FREQUENCY = "hbase.server.thread.wakefrequency";
- static final String HREGION_DIR = "hbase.regiondir";
- static final String DEFAULT_HREGION_DIR = "/hbase";
+
+ /** Parameter name for HBase instance root directory */
+ static final String HBASE_DIR = "hbase.rootdir";
+
+ /** Default HBase instance root directory */
+ static final String DEFAULT_HBASE_DIR = "/hbase";
+
+ /** Used to construct the name of the directory in which a HRegion resides */
static final String HREGIONDIR_PREFIX = "hregion_";
// TODO: Someone may try to name a column family 'log'. If they
// do, it will clash with the HREGION log dir subdirectory. FIX.
+
+ /** Used to construct the name of the log directory for a region server */
static final String HREGION_LOGDIR_NAME = "log";
+
+ /** Name of old log file for reconstruction */
+ static final String HREGION_OLDLOGFILE_NAME = "oldlogfile.log";
+ /** Default maximum file size */
static final long DEFAULT_MAX_FILE_SIZE = 128 * 1024 * 1024; // 128MB
// Always store the location of the root table's HRegion.
@@ -59,26 +78,36 @@
// each row in the root and meta tables describes exactly 1 region
// Do we ever need to know all the information that we are storing?
- // The root tables' name.
+ /** The root table's name. */
static final Text ROOT_TABLE_NAME = new Text("--ROOT--");
- // The META tables' name.
+ /** The META table's name. */
static final Text META_TABLE_NAME = new Text("--META--");
- // Defines for the column names used in both ROOT and META HBase 'meta'
- // tables.
+ // Defines for the column names used in both ROOT and META HBase 'meta' tables.
+
+ /** The ROOT and META column family */
static final Text COLUMN_FAMILY = new Text("info:");
+
+ /** ROOT/META column family member - contains HRegionInfo */
static final Text COL_REGIONINFO = new Text(COLUMN_FAMILY + "regioninfo");
+
+ /** ROOT/META column family member - contains HServerAddress.toString() */
static final Text COL_SERVER = new Text(COLUMN_FAMILY + "server");
+
+ /** ROOT/META column family member - contains server start code (a long) */
static final Text COL_STARTCODE = new Text(COLUMN_FAMILY + "serverstartcode");
// Other constants
-
+
+ /** When we encode strings, we always specify UTF8 encoding */
static final String UTF8_ENCODING = "UTF-8";
-
+
+ /** Value stored for a deleted item */
static final BytesWritable DELETE_BYTES =
new BytesWritable("HBASE::DELETEVAL".getBytes());
-
+
+ /** Value written to HLog on a complete cache flush */
static final BytesWritable COMPLETE_CACHEFLUSH =
new BytesWritable("HBASE::CACHEFLUSH".getBytes());
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java Mon Jun 11 09:46:27 2007
@@ -81,26 +81,26 @@
Integer rollLock = 0;
/**
- * Bundle up a bunch of log files (which are no longer being written to),
- * into a new file. Delete the old log files when ready.
- * @param srcDir Directory of log files to bundle:
- * e.g. <code>${REGIONDIR}/log_HOST_PORT</code>
- * @param dstFile Destination file:
- * e.g. <code>${REGIONDIR}/oldlogfile_HOST_PORT</code>
+ * Split up a bunch of log files, that are no longer being written to,
+ * into new files, one per region. Delete the old log files when ready.
+ * @param rootDir Root directory of the HBase instance
+ * @param srcDir Directory of log files to split:
+ * e.g. <code>${ROOTDIR}/log_HOST_PORT</code>
* @param fs FileSystem
* @param conf HBaseConfiguration
* @throws IOException
*/
- public static void consolidateOldLog(Path srcDir, Path dstFile,
- FileSystem fs, Configuration conf)
- throws IOException {
+ static void splitLog(Path rootDir, Path srcDir, FileSystem fs,
+ Configuration conf) throws IOException {
+
if(LOG.isDebugEnabled()) {
- LOG.debug("consolidating log files");
+ LOG.debug("splitting log files");
}
Path logfiles[] = fs.listPaths(srcDir);
- SequenceFile.Writer newlog = SequenceFile.createWriter(fs, conf, dstFile,
- HLogKey.class, HLogEdit.class);
+ TreeMap<Text, SequenceFile.Writer> logWriters =
+ new TreeMap<Text, SequenceFile.Writer>();
+
try {
for(int i = 0; i < logfiles.length; i++) {
SequenceFile.Reader in =
@@ -109,7 +109,17 @@
HLogKey key = new HLogKey();
HLogEdit val = new HLogEdit();
while(in.next(key, val)) {
- newlog.append(key, val);
+ Text regionName = key.getRegionName();
+ SequenceFile.Writer w = logWriters.get(regionName);
+ if(w == null) {
+ Path logfile = new Path(HStoreFile.getHRegionDir(rootDir,
+ regionName), HREGION_OLDLOGFILE_NAME);
+
+ w = SequenceFile.createWriter(fs, conf, logfile, HLogKey.class,
+ HLogEdit.class);
+ logWriters.put(regionName, w);
+ }
+ w.append(key, val);
}
} finally {
@@ -118,7 +128,9 @@
}
} finally {
- newlog.close();
+ for(SequenceFile.Writer w: logWriters.values()) {
+ w.close();
+ }
}
if(fs.exists(srcDir)) {
@@ -132,7 +144,7 @@
}
}
if(LOG.isDebugEnabled()) {
- LOG.debug("log file consolidation completed");
+ LOG.debug("log file splitting completed");
}
}
@@ -142,8 +154,13 @@
* You should never have to load an existing log. If there is a log
* at startup, it should have already been processed and deleted by
* the time the HLog object is started up.
+ *
+ * @param fs
+ * @param dir
+ * @param conf
+ * @throws IOException
*/
- public HLog(FileSystem fs, Path dir, Configuration conf) throws IOException {
+ HLog(FileSystem fs, Path dir, Configuration conf) throws IOException {
this.fs = fs;
this.dir = dir;
this.conf = conf;
@@ -163,8 +180,10 @@
*
* The 'this' lock limits access to the current writer so
* we don't append multiple items simultaneously.
+ *
+ * @throws IOException
*/
- public void rollWriter() throws IOException {
+ void rollWriter() throws IOException {
synchronized(rollLock) {
// Try to roll the writer to a new file. We may have to
@@ -267,8 +286,21 @@
return new Path(dir, HLOG_DATFILE + String.format("%1$03d", filenum));
}
- /** Shut down the log. */
- public synchronized void close() throws IOException {
+ /**
+ * Shut down the log and delete the log directory
+ * @throws IOException
+ */
+ synchronized void closeAndDelete() throws IOException {
+ rollWriter();
+ close();
+ fs.delete(dir);
+ }
+
+ /**
+ * Shut down the log.
+ * @throws IOException
+ */
+ synchronized void close() throws IOException {
if(LOG.isDebugEnabled()) {
LOG.debug("closing log writer");
}
@@ -300,7 +332,7 @@
* @param timestamp
* @throws IOException
*/
- public synchronized void append(Text regionName, Text tableName, Text row,
+ synchronized void append(Text regionName, Text tableName, Text row,
TreeMap<Text, BytesWritable> columns, long timestamp)
throws IOException {
if(closed) {
@@ -327,8 +359,8 @@
}
}
- /** How many items have been added to the log? */
- public int getNumEntries() {
+ /** @return How many items have been added to the log */
+ int getNumEntries() {
return numEntries;
}
@@ -340,6 +372,12 @@
return logSeqNum++;
}
+ /**
+ * Obtain a specified number of sequence numbers
+ *
+ * @param num - number of sequence numbers to obtain
+ * @return - array of sequence numbers
+ */
synchronized long[] obtainSeqNum(int num) {
long[] results = new long[num];
for (int i = 0; i < num; i++) {
@@ -358,7 +396,7 @@
* @return sequence ID to pass {@link #completeCacheFlush(Text, Text, long)}
* @see #completeCacheFlush(Text, Text, long)
*/
- public synchronized long startCacheFlush() {
+ synchronized long startCacheFlush() {
while (insideCacheFlush) {
try {
wait();
@@ -370,8 +408,13 @@
return obtainSeqNum();
}
- /** Complete the cache flush */
- public synchronized void completeCacheFlush(final Text regionName,
+ /** Complete the cache flush
+ * @param regionName
+ * @param tableName
+ * @param logSeqId
+ * @throws IOException
+ */
+ synchronized void completeCacheFlush(final Text regionName,
final Text tableName, final long logSeqId)
throws IOException {
if(closed) {
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLogKey.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLogKey.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLogKey.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLogKey.java Mon Jun 11 09:46:27 2007
@@ -32,14 +32,20 @@
Text row = new Text();
long logSeqNum = 0L;
+ /** Create an empty key useful when deserializing */
+ public HLogKey() {
+ }
+
/**
* Create the log key!
* We maintain the tablename mainly for debugging purposes.
* A regionName is always a sub-table object.
+ *
+ * @param regionName - name of region
+ * @param tablename - name of table
+ * @param row - row key
+ * @param logSeqNum - log sequence number
*/
- public HLogKey() {
- }
-
public HLogKey(Text regionName, Text tablename, Text row, long logSeqNum) {
this.regionName.set(regionName);
this.tablename.set(tablename);
@@ -51,26 +57,25 @@
// A bunch of accessors
//////////////////////////////////////////////////////////////////////////////
- public Text getRegionName() {
+ Text getRegionName() {
return regionName;
}
- public Text getTablename() {
+ Text getTablename() {
return tablename;
}
- public Text getRow() {
+ Text getRow() {
return row;
}
- public long getLogSeqNum() {
+ long getLogSeqNum() {
return logSeqNum;
}
@Override
public String toString() {
- return getTablename().toString() + " " + getRegionName().toString() + " " +
- getRow().toString() + " " + getLogSeqNum();
+ return tablename + " " + regionName + " " + row + " " + logSeqNum;
}
@Override
@@ -90,10 +95,8 @@
// Comparable
//////////////////////////////////////////////////////////////////////////////
- /**
- * When sorting through log entries, we want to group items
- * first in the same table, then to the same row, then finally
- * ordered by write-order.
+ /* (non-Javadoc)
+ * @see java.lang.Comparable#compareTo(java.lang.Object)
*/
public int compareTo(Object o) {
HLogKey other = (HLogKey) o;
@@ -119,6 +122,9 @@
// Writable
//////////////////////////////////////////////////////////////////////////////
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.io.Writable#write(java.io.DataOutput)
+ */
public void write(DataOutput out) throws IOException {
this.regionName.write(out);
this.tablename.write(out);
@@ -126,6 +132,9 @@
out.writeLong(logSeqNum);
}
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.io.Writable#readFields(java.io.DataInput)
+ */
public void readFields(DataInput in) throws IOException {
this.regionName.readFields(in);
this.tablename.readFields(in);
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java?view=diff&rev=546192&r1=546191&r2=546192
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java Mon Jun 11 09:46:27 2007
@@ -21,6 +21,7 @@
import java.io.UnsupportedEncodingException;
import java.util.Collections;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.Map;
import java.util.Random;
import java.util.SortedMap;
@@ -50,6 +51,9 @@
public class HMaster implements HConstants, HMasterInterface,
HMasterRegionInterface, Runnable {
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.ipc.VersionedProtocol#getProtocolVersion(java.lang.String, long)
+ */
public long getProtocolVersion(String protocol,
@SuppressWarnings("unused") long clientVersion)
throws IOException {
@@ -67,14 +71,14 @@
volatile boolean closed;
Path dir;
- private Configuration conf;
+ Configuration conf;
FileSystem fs;
Random rand;
- private long threadWakeFrequency;
- private int numRetries;
- private long maxRegionOpenTime;
+ long threadWakeFrequency;
+ int numRetries;
+ long maxRegionOpenTime;
- Vector<PendingOperation> msgQueue;
+ LinkedList<PendingOperation> msgQueue;
private Leases serverLeases;
private Server server;
@@ -84,7 +88,7 @@
long metaRescanInterval;
- private HServerAddress rootRegionLocation;
+ HServerAddress rootRegionLocation;
/**
* Columns in the 'meta' ROOT and META tables.
@@ -93,7 +97,6 @@
COLUMN_FAMILY
};
- static final String MASTER_NOT_RUNNING = "Master not running";
boolean rootScanned;
int numMetaRegions;
@@ -166,20 +169,20 @@
try {
regionServer = client.getHRegionConnection(region.server);
scannerId = regionServer.openScanner(region.regionName, METACOLUMNS,
- FIRST_ROW);
+ FIRST_ROW);
+
while (true) {
TreeMap<Text, byte[]> results = new TreeMap<Text, byte[]>();
- HStoreKey key = new HStoreKey();
- LabelledData[] values = regionServer.next(scannerId, key);
+ KeyedData[] values = regionServer.next(scannerId);
if (values.length == 0) {
break;
}
-
+
for (int i = 0; i < values.length; i++) {
byte[] bytes = new byte[values[i].getData().getSize()];
System.arraycopy(values[i].getData().get(), 0, bytes, 0,
- bytes.length);
- results.put(values[i].getLabel(), bytes);
+ bytes.length);
+ results.put(values[i].getKey().getColumn(), bytes);
}
HRegionInfo info = HRegion.getRegionInfo(results);
@@ -188,19 +191,21 @@
if(LOG.isDebugEnabled()) {
LOG.debug(Thread.currentThread().getName() + " scanner: " +
- Long.valueOf(scannerId) + " regioninfo: {" + info.toString() +
- "}, server: " + serverName + ", startCode: " + startCode);
+ Long.valueOf(scannerId) + " regioninfo: {" + info.toString() +
+ "}, server: " + serverName + ", startCode: " + startCode);
}
// Note Region has been assigned.
checkAssigned(info, serverName, startCode);
scannedRegion = true;
}
+
} catch (UnknownScannerException e) {
// Reset scannerId so we do not try closing a scanner the other side
// has lost account of: prevents duplicated stack trace out of the
// below close in the finally.
scannerId = -1L;
+
} finally {
try {
if (scannerId != -1L) {
@@ -249,13 +254,25 @@
}
storedInfo = serversToServerInfo.get(serverName);
}
- if(storedInfo == null || storedInfo.getStartCode() != startCode) {
+ if( !(
+ unassignedRegions.containsKey(info.regionName)
+ || pendingRegions.contains(info.regionName)
+ )
+ && (storedInfo == null
+ || storedInfo.getStartCode() != startCode)) {
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("region unassigned: " + info.regionName
+ + " serverName: " + serverName
+ + (storedInfo == null ? " storedInfo == null"
+ : (" startCode=" + startCode + ", storedStartCode="
+ + storedInfo.getStartCode())));
+ }
+
// The current assignment is no good; load the region.
+
unassignedRegions.put(info.regionName, info);
assignAttempts.put(info.regionName, 0L);
- if(LOG.isDebugEnabled()) {
- LOG.debug("region unassigned: " + info.regionName);
- }
}
}
}
@@ -263,13 +280,14 @@
/**
* Scanner for the <code>ROOT</code> HRegion.
*/
- private class RootScanner extends BaseScanner {
+ class RootScanner extends BaseScanner {
public void run() {
if (LOG.isDebugEnabled()) {
LOG.debug("Running ROOT scanner");
}
- try {
- while(!closed) {
+ int tries = 0;
+ while(!closed && tries < numRetries) {
+ try {
// rootRegionLocation will be filled in when we get an 'open region'
// regionServerReport message from the HRegionServer that has been
// allocated the ROOT region below. If we get back false, then
@@ -289,16 +307,25 @@
}
rootScanned = true;
}
- try {
- Thread.sleep(metaRescanInterval);
- } catch(InterruptedException e) {
- // Catch and go around again. If interrupt, its spurious or we're
- // being shutdown. Go back up to the while test.
+ tries = 0;
+
+ } catch(IOException e) {
+ tries++;
+ if(tries < numRetries) {
+ LOG.warn("ROOT scanner", e);
+
+ } else {
+ LOG.error("ROOT scanner", e);
+ closed = true;
+ break;
}
}
- } catch(IOException e) {
- LOG.error("ROOT scanner", e);
- closed = true;
+ try {
+ Thread.sleep(metaRescanInterval);
+ } catch(InterruptedException e) {
+ // Catch and go around again. If interrupt, its spurious or we're
+ // being shutdown. Go back up to the while test.
+ }
}
LOG.info("ROOT scanner exiting");
}
@@ -306,12 +333,13 @@
private RootScanner rootScanner;
private Thread rootScannerThread;
- private Integer rootScannerLock = 0;
-
- private static class MetaRegion implements Comparable {
- public HServerAddress server;
- public Text regionName;
- public Text startKey;
+ Integer rootScannerLock = 0;
+
+ @SuppressWarnings("unchecked")
+ static class MetaRegion implements Comparable {
+ HServerAddress server;
+ Text regionName;
+ Text startKey;
@Override
public boolean equals(Object o) {
@@ -354,7 +382,7 @@
* It's important to do this work in a separate thread, or else the blocking
* action would prevent other work from getting done.
*/
- private class MetaScanner extends BaseScanner {
+ class MetaScanner extends BaseScanner {
@SuppressWarnings("null")
public void run() {
while (!closed) {
@@ -369,7 +397,7 @@
}
if (region == null) {
try {
- metaRegionsToScan.wait();
+ metaRegionsToScan.wait(threadWakeFrequency);
} catch (InterruptedException e) {
// Catch and go around again. We've been woken because there
// are new meta regions available or because we are being
@@ -394,6 +422,7 @@
}
}
+ int tries = 0;
do {
try {
Thread.sleep(metaRescanInterval);
@@ -405,13 +434,26 @@
break;
}
- // Rescan the known meta regions every so often
+ try {
+
+ // Rescan the known meta regions every so often
- synchronized(metaScannerLock) { // Don't interrupt us while we're working
- Vector<MetaRegion> v = new Vector<MetaRegion>();
- v.addAll(knownMetaRegions.values());
- for(Iterator<MetaRegion> i = v.iterator(); i.hasNext(); ) {
- scanRegion(i.next());
+ synchronized(metaScannerLock) { // Don't interrupt us while we're working
+ Vector<MetaRegion> v = new Vector<MetaRegion>();
+ v.addAll(knownMetaRegions.values());
+ for(Iterator<MetaRegion> i = v.iterator(); i.hasNext(); ) {
+ scanRegion(i.next());
+ }
+ }
+ tries = 0;
+
+ } catch (IOException e) {
+ tries++;
+ if(tries < numRetries) {
+ LOG.warn("META scanner", e);
+
+ } else {
+ throw e;
}
}
} while(true);
@@ -424,58 +466,82 @@
LOG.info("META scanner exiting");
}
+ /**
+ * Called by the meta scanner when it has completed scanning all meta
+ * regions. This wakes up any threads that were waiting for this to happen.
+ */
private synchronized void metaRegionsScanned() {
notifyAll();
}
- public synchronized void waitForMetaScan() {
+ /**
+ * Other threads call this method to wait until all the meta regions have
+ * been scanned.
+ */
+ synchronized boolean waitForMetaScanOrClose() {
while(!closed && !allMetaRegionsScanned) {
try {
- wait();
+ wait(threadWakeFrequency);
} catch(InterruptedException e) {
// continue
}
}
+ return closed;
}
}
MetaScanner metaScanner;
private Thread metaScannerThread;
Integer metaScannerLock = 0;
-
- // The 'unassignedRegions' table maps from a region name to a HRegionInfo record,
- // which includes the region's table, its id, and its start/end keys.
- //
- // We fill 'unassignedRecords' by scanning ROOT and META tables, learning the
- // set of all known valid regions.
+ /**
+ * The 'unassignedRegions' table maps from a region name to a HRegionInfo
+ * record, which includes the region's table, its id, and its start/end keys.
+ *
+ * We fill 'unassignedRecords' by scanning ROOT and META tables, learning the
+ * set of all known valid regions.
+ */
SortedMap<Text, HRegionInfo> unassignedRegions;
- // The 'assignAttempts' table maps from regions to a timestamp that indicates
- // the last time we *tried* to assign the region to a RegionServer. If the
- // timestamp is out of date, then we can try to reassign it.
-
+ /**
+ * The 'assignAttempts' table maps from regions to a timestamp that indicates
+ * the last time we *tried* to assign the region to a RegionServer. If the
+ * timestamp is out of date, then we can try to reassign it.
+ */
SortedMap<Text, Long> assignAttempts;
- SortedMap<String, TreeMap<Text, HRegionInfo>> killList;
+ /**
+ * Regions that have been assigned, and the server has reported that it has
+ * started serving it, but that we have not yet recorded in the meta table.
+ */
+ SortedSet<Text> pendingRegions;
- // 'killedRegions' contains regions that are in the process of being closed
+ /**
+ * The 'killList' is a list of regions that are going to be closed, but not
+ * reopened.
+ */
+ SortedMap<String, TreeMap<Text, HRegionInfo>> killList;
+ /** 'killedRegions' contains regions that are in the process of being closed */
SortedSet<Text> killedRegions;
-
- // 'regionsToDelete' contains regions that need to be deleted, but cannot be
- // until the region server closes it
-
+
+ /**
+ * 'regionsToDelete' contains regions that need to be deleted, but cannot be
+ * until the region server closes it
+ */
SortedSet<Text> regionsToDelete;
- // A map of known server names to server info
-
+ /** The map of known server names to server info */
SortedMap<String, HServerInfo> serversToServerInfo =
Collections.synchronizedSortedMap(new TreeMap<String, HServerInfo>());
- /** Build the HMaster out of a raw configuration item. */
+ /** Build the HMaster out of a raw configuration item.
+ *
+ * @param conf - Configuration object
+ * @throws IOException
+ */
public HMaster(Configuration conf) throws IOException {
- this(new Path(conf.get(HREGION_DIR, DEFAULT_HREGION_DIR)),
+ this(new Path(conf.get(HBASE_DIR, DEFAULT_HBASE_DIR)),
new HServerAddress(conf.get(MASTER_ADDRESS, DEFAULT_MASTER_ADDRESS)),
conf);
}
@@ -515,9 +581,9 @@
// Add first region from the META table to the ROOT region.
HRegion.addRegionToMETA(root, meta);
root.close();
- root.getLog().close();
+ root.getLog().closeAndDelete();
meta.close();
- meta.getLog().close();
+ meta.getLog().closeAndDelete();
} catch(IOException e) {
LOG.error(e);
}
@@ -526,7 +592,7 @@
this.threadWakeFrequency = conf.getLong(THREAD_WAKE_FREQUENCY, 10 * 1000);
this.numRetries = conf.getInt("hbase.client.retries.number", 2);
this.maxRegionOpenTime = conf.getLong("hbase.hbasemaster.maxregionopen", 30 * 1000);
- this.msgQueue = new Vector<PendingOperation>();
+ this.msgQueue = new LinkedList<PendingOperation>();
this.serverLeases = new Leases(
conf.getLong("hbase.master.lease.period", 30 * 1000),
conf.getLong("hbase.master.lease.thread.wakefrequency", 15 * 1000));
@@ -573,6 +639,9 @@
this.assignAttempts =
Collections.synchronizedSortedMap(new TreeMap<Text, Long>());
+ this.pendingRegions =
+ Collections.synchronizedSortedSet(new TreeSet<Text>());
+
this.assignAttempts.put(HGlobals.rootRegionInfo.regionName, 0L);
this.killList =
@@ -589,14 +658,17 @@
this.closed = false;
- LOG.info("HMaster initialized on " + address.toString());
+ LOG.info("HMaster initialized on " + this.address.toString());
}
- /** returns the HMaster server address */
- public HServerAddress getMasterAddress() {
+ /**
+ * @return HServerAddress of the master server
+ */
+ HServerAddress getMasterAddress() {
return address;
}
+ /** Main processing loop */
public void run() {
Thread.currentThread().setName("HMaster");
try {
@@ -626,7 +698,7 @@
if(closed) {
continue;
}
- op = msgQueue.remove(msgQueue.size()-1);
+ op = msgQueue.removeFirst();
}
try {
if (LOG.isDebugEnabled()) {
@@ -634,7 +706,10 @@
}
op.process();
} catch(Exception ex) {
- msgQueue.insertElementAt(op, 0);
+ LOG.warn(ex);
+ synchronized(msgQueue) {
+ msgQueue.addLast(op);
+ }
}
}
letRegionServersShutdown();
@@ -698,7 +773,8 @@
while (endTime > System.currentTimeMillis() &&
this.serversToServerInfo.size() > 0) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Waiting on regionservers: " + this.serversToServerInfo);
+ LOG.debug("Waiting on regionservers: "
+ + this.serversToServerInfo.values());
}
try {
Thread.sleep(threadWakeFrequency);
@@ -713,13 +789,13 @@
* <code>closed</code> flag has been set.
* @return True if <code>rootRegionLocation</code> was populated.
*/
- private synchronized boolean waitForRootRegionOrClose() {
+ synchronized boolean waitForRootRegionOrClose() {
while (!closed && rootRegionLocation == null) {
try {
if (LOG.isDebugEnabled()) {
LOG.debug("Wait for root region (or close)");
}
- wait();
+ wait(threadWakeFrequency);
if (LOG.isDebugEnabled()) {
LOG.debug("Wake from wait for root region (or close)");
}
@@ -736,7 +812,9 @@
// HMasterRegionInterface
//////////////////////////////////////////////////////////////////////////////
- /** HRegionServers call this method upon startup. */
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.hbase.HMasterRegionInterface#regionServerStartup(org.apache.hadoop.hbase.HServerInfo)
+ */
public void regionServerStartup(HServerInfo serverInfo) throws IOException {
String s = serverInfo.getServerAddress().toString().trim();
HServerInfo storedInfo = null;
@@ -752,7 +830,7 @@
if(storedInfo != null && !closed) {
synchronized(msgQueue) {
- msgQueue.add(new PendingServerShutdown(storedInfo));
+ msgQueue.addLast(new PendingServerShutdown(storedInfo));
msgQueue.notifyAll();
}
}
@@ -768,24 +846,47 @@
}
}
- /** HRegionServers call this method repeatedly. */
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.hbase.HMasterRegionInterface#regionServerReport(org.apache.hadoop.hbase.HServerInfo, org.apache.hadoop.hbase.HMsg[])
+ */
public HMsg[] regionServerReport(HServerInfo serverInfo, HMsg msgs[])
throws IOException {
String s = serverInfo.getServerAddress().toString().trim();
Text serverLabel = new Text(s);
- if (closed ||
- msgs.length == 1 && msgs[0].getMsg() == HMsg.MSG_REPORT_EXITING) {
- // HRegionServer is shutting down.
- if (serversToServerInfo.remove(s) != null) {
- // Only cancel lease once (This block can run a couple of times during
- // shutdown).
- LOG.debug("Cancelling lease for " + serverLabel);
- serverLeases.cancelLease(serverLabel, serverLabel);
- }
+ if (closed) {
+ // Cancel the server's lease
+ cancelLease(s, serverLabel);
+
+ // Tell server to shut down
HMsg returnMsgs[] = {new HMsg(HMsg.MSG_REGIONSERVER_STOP)};
return returnMsgs;
}
+
+ if(msgs.length > 0 && msgs[0].getMsg() == HMsg.MSG_REPORT_EXITING) {
+ // HRegionServer is shutting down.
+
+ // Cancel the server's lease
+ cancelLease(s, serverLabel);
+
+ // Get all the regions the server was serving reassigned
+
+ for(int i = 1; i < msgs.length; i++) {
+ HRegionInfo info = msgs[i].getRegionInfo();
+ if(info.tableDesc.getName().equals(ROOT_TABLE_NAME)) {
+ rootRegionLocation = null;
+
+ } else if(info.tableDesc.getName().equals(META_TABLE_NAME)) {
+ allMetaRegionsScanned = false;
+ }
+ unassignedRegions.put(info.regionName, info);
+ assignAttempts.put(info.regionName, 0L);
+ }
+
+ // We don't need to return anything to the server because it isn't
+ // going to do any more work.
+ return new HMsg[0];
+ }
HServerInfo storedInfo = serversToServerInfo.get(s);
@@ -838,8 +939,18 @@
}
}
+ /** cancel a server's lease */
+ private void cancelLease(String serverName, Text serverLabel) throws IOException {
+ if (serversToServerInfo.remove(serverName) != null) {
+ // Only cancel lease once.
+ // This method can be called a couple of times during shutdown.
+ LOG.debug("Cancelling lease for " + serverName);
+ serverLeases.cancelLease(serverLabel, serverLabel);
+ }
+ }
+
/** Process all the incoming messages from a server that's contacted us. */
- HMsg[] processMsgs(HServerInfo info, HMsg incomingMsgs[]) throws IOException {
+ private HMsg[] processMsgs(HServerInfo info, HMsg incomingMsgs[]) throws IOException {
Vector<HMsg> returnMsgs = new Vector<HMsg>();
TreeMap<Text, HRegionInfo> regionsToKill =
@@ -859,7 +970,7 @@
if(LOG.isDebugEnabled()) {
LOG.debug("region server " + info.getServerAddress().toString()
- + "should not have opened region " + region.regionName);
+ + " should not have opened region " + region.regionName);
}
// This Region should not have been opened.
@@ -876,6 +987,11 @@
+ region.regionName);
}
+ // Note that it has been assigned and is waiting for the meta table
+ // to be updated.
+
+ pendingRegions.add(region.regionName);
+
// Remove from unassigned list so we don't assign it to someone else
unassignedRegions.remove(region.regionName);
@@ -892,7 +1008,7 @@
rootRegionIsAvailable();
break;
- } else if(region.regionName.find(META_TABLE_NAME.toString()) == 0) {
+ } else if(region.tableDesc.getName().equals(META_TABLE_NAME)) {
// It's a meta region. Put it on the queue to be scanned.
@@ -910,7 +1026,7 @@
// Queue up an update to note the region location.
synchronized(msgQueue) {
- msgQueue.add(new PendingOpenReport(info, region.regionName));
+ msgQueue.addLast(new PendingOpenReport(info, region));
msgQueue.notifyAll();
}
}
@@ -943,7 +1059,7 @@
assignAttempts.remove(region.regionName);
synchronized(msgQueue) {
- msgQueue.add(new PendingCloseReport(region, reassignRegion, deleteRegion));
+ msgQueue.addLast(new PendingCloseReport(region, reassignRegion, deleteRegion));
msgQueue.notifyAll();
}
@@ -961,7 +1077,7 @@
// A region has split and the old server is serving the two new regions.
- if(region.regionName.find(META_TABLE_NAME.toString()) == 0) {
+ if(region.tableDesc.getName().equals(META_TABLE_NAME)) {
// A meta region has split.
allMetaRegionsScanned = false;
@@ -996,10 +1112,7 @@
int counter = 0;
long now = System.currentTimeMillis();
- for(Iterator<Text> it = unassignedRegions.keySet().iterator();
- it.hasNext(); ) {
-
- Text curRegionName = it.next();
+ for(Text curRegionName: unassignedRegions.keySet()) {
HRegionInfo regionInfo = unassignedRegions.get(curRegionName);
long assignedTime = assignAttempts.get(curRegionName);
@@ -1023,6 +1136,11 @@
return returnMsgs.toArray(new HMsg[returnMsgs.size()]);
}
+ /**
+ * Called when the master has received a report from a region server that it
+ * is now serving the root region. Causes any threads waiting for the root
+ * region to be available to be woken up.
+ */
private synchronized void rootRegionIsAvailable() {
notifyAll();
}
@@ -1038,37 +1156,45 @@
protected final Text startRow = new Text();
protected long clientId;
- public PendingOperation() {
+ PendingOperation() {
this.clientId = rand.nextLong();
}
- public abstract void process() throws IOException;
+ abstract void process() throws IOException;
}
-
+
+ /**
+ * Instantiated when a server's lease has expired, meaning it has crashed.
+ * The region server's log file needs to be split up for each region it was
+ * serving, and the regions need to get reassigned.
+ */
private class PendingServerShutdown extends PendingOperation {
- private String deadServer;
+ private HServerAddress deadServer;
+ private String deadServerName;
private long oldStartCode;
private class ToDoEntry {
boolean deleteRegion;
boolean regionOffline;
- HStoreKey key;
+ Text row;
HRegionInfo info;
- ToDoEntry(HStoreKey key, HRegionInfo info) {
+ ToDoEntry(Text row, HRegionInfo info) {
this.deleteRegion = false;
this.regionOffline = false;
- this.key = key;
+ this.row = row;
this.info = info;
}
}
- public PendingServerShutdown(HServerInfo serverInfo) {
+ PendingServerShutdown(HServerInfo serverInfo) {
super();
- this.deadServer = serverInfo.getServerAddress().toString();
+ this.deadServer = serverInfo.getServerAddress();
+ this.deadServerName = this.deadServer.toString();
this.oldStartCode = serverInfo.getStartCode();
}
+ /** Finds regions that the dead region server was serving */
private void scanMetaRegion(HRegionInterface server, long scannerId,
Text regionName) throws IOException {
@@ -1078,11 +1204,10 @@
DataInputBuffer inbuf = new DataInputBuffer();
try {
while(true) {
- LabelledData[] values = null;
+ KeyedData[] values = null;
- HStoreKey key = new HStoreKey();
try {
- values = server.next(scannerId, key);
+ values = server.next(scannerId);
} catch(NotServingRegionException e) {
throw e;
@@ -1097,13 +1222,24 @@
}
TreeMap<Text, byte[]> results = new TreeMap<Text, byte[]>();
+ Text row = null;
+ byte[] bytes = null;
for(int i = 0; i < values.length; i++) {
- byte[] bytes = new byte[values[i].getData().getSize()];
+ if(row == null) {
+ row = values[i].getKey().getRow();
+
+ } else {
+ if(!row.equals(values[i].getKey().getRow())) {
+ LOG.error("Multiple rows in same scanner result set. firstRow="
+ + row + ", currentRow=" + values[i].getKey().getRow());
+ }
+ }
+ bytes = new byte[values[i].getData().getSize()];
System.arraycopy(values[i].getData().get(), 0, bytes, 0, bytes.length);
- results.put(values[i].getLabel(), bytes);
+ results.put(values[i].getKey().getColumn(), bytes);
}
- byte[] bytes = results.get(COL_SERVER);
+ bytes = results.get(COL_SERVER);
String serverName = null;
if(bytes == null || bytes.length == 0) {
// No server
@@ -1117,7 +1253,7 @@
break;
}
- if(deadServer.compareTo(serverName) != 0) {
+ if(deadServerName.compareTo(serverName) != 0) {
// This isn't the server you're looking for - move along
continue;
}
@@ -1128,10 +1264,9 @@
continue;
}
long startCode = -1L;
-
try {
- startCode = Long.valueOf(new String(bytes, UTF8_ENCODING)).
- longValue();
+ startCode =
+ Long.valueOf(new String(bytes, UTF8_ENCODING)).longValue();
} catch(UnsupportedEncodingException e) {
LOG.error(e);
break;
@@ -1150,7 +1285,6 @@
}
inbuf.reset(bytes, bytes.length);
HRegionInfo info = new HRegionInfo();
-
try {
info.readFields(inbuf);
@@ -1160,17 +1294,21 @@
}
if(LOG.isDebugEnabled()) {
- LOG.debug(serverName + " was serving " + info.regionName);
+ LOG.debug(serverName + " was serving " + info.toString());
}
- ToDoEntry todo = new ToDoEntry(key, info);
+ if(info.tableDesc.getName().equals(META_TABLE_NAME)) {
+ allMetaRegionsScanned = false;
+ }
+
+ ToDoEntry todo = new ToDoEntry(row, info);
toDoList.add(todo);
- if(killList.containsKey(deadServer)) {
- TreeMap<Text, HRegionInfo> regionsToKill = killList.get(deadServer);
+ if(killList.containsKey(deadServerName)) {
+ TreeMap<Text, HRegionInfo> regionsToKill = killList.get(deadServerName);
if(regionsToKill.containsKey(info.regionName)) {
regionsToKill.remove(info.regionName);
- killList.put(deadServer, regionsToKill);
+ killList.put(deadServerName, regionsToKill);
unassignedRegions.remove(info.regionName);
assignAttempts.remove(info.regionName);
@@ -1209,7 +1347,7 @@
for(int i = 0; i < toDoList.size(); i++) {
ToDoEntry e = toDoList.get(i);
- long lockid = server.startUpdate(regionName, clientId, e.key.getRow());
+ long lockid = server.startUpdate(regionName, clientId, e.row);
if(e.deleteRegion) {
server.delete(regionName, clientId, lockid, COL_REGIONINFO);
@@ -1237,22 +1375,40 @@
assignAttempts.put(region, 0L);
}
}
-
- public void process() throws IOException {
+
+ @Override
+ void process() throws IOException {
if(LOG.isDebugEnabled()) {
- LOG.debug("server shutdown: " + deadServer);
+ LOG.debug("server shutdown: " + deadServerName);
}
+
+ // Process the old log file
+
+ HLog.splitLog(dir, new Path(dir, "log" + "_" + deadServer.getBindAddress()
+ + "_" + deadServer.getPort()), fs, conf);
+ if(rootRegionLocation != null
+ && deadServerName.equals(rootRegionLocation.toString())) {
+
+ rootRegionLocation = null;
+ unassignedRegions.put(HGlobals.rootRegionInfo.regionName,
+ HGlobals.rootRegionInfo);
+ assignAttempts.put(HGlobals.rootRegionInfo.regionName, 0L);
+ }
+
// Scan the ROOT region
HRegionInterface server = null;
long scannerId = -1L;
for(int tries = 0; tries < numRetries; tries ++) {
- waitForRootRegion(); // Wait until the root region is available
+ if(waitForRootRegionOrClose()) {// Wait until the root region is available
+ return; // We're shutting down. Forget it.
+ }
server = client.getHRegionConnection(rootRegionLocation);
scannerId = -1L;
try {
+ LOG.debug("scanning root region");
scannerId = server.openScanner(HGlobals.rootRegionInfo.regionName, columns, startRow);
scanMetaRegion(server, scannerId, HGlobals.rootRegionInfo.regionName);
break;
@@ -1269,7 +1425,9 @@
for(int tries = 0; tries < numRetries; tries ++) {
try {
- metaScanner.waitForMetaScan();
+ if(metaScanner.waitForMetaScanOrClose()) {
+ return; // We're shutting down. Forget it.
+ }
for(Iterator<MetaRegion> i = knownMetaRegions.values().iterator();
i.hasNext(); ) {
@@ -1295,14 +1453,17 @@
}
}
- /** PendingCloseReport is a close message that is saved in a different thread. */
+ /**
+ * PendingCloseReport is instantiated when a region server reports that it
+ * has closed a region.
+ */
private class PendingCloseReport extends PendingOperation {
private HRegionInfo regionInfo;
private boolean reassignRegion;
private boolean deleteRegion;
private boolean rootRegion;
- public PendingCloseReport(HRegionInfo regionInfo, boolean reassignRegion,
+ PendingCloseReport(HRegionInfo regionInfo, boolean reassignRegion,
boolean deleteRegion) {
super();
@@ -1314,7 +1475,7 @@
// If the region closing down is a meta region then we need to update
// the ROOT table
- if(this.regionInfo.regionName.find(HGlobals.metaTableDesc.getName().toString()) == 0) {
+ if(this.regionInfo.tableDesc.getName().equals(META_TABLE_NAME)) {
this.rootRegion = true;
} else {
@@ -1322,13 +1483,16 @@
}
}
- public void process() throws IOException {
+ @Override
+ void process() throws IOException {
for(int tries = 0; tries < numRetries; tries ++) {
// We can not access any meta region if they have not already been assigned
// and scanned.
- metaScanner.waitForMetaScan();
+ if(metaScanner.waitForMetaScanOrClose()) {
+ return; // We're shutting down. Forget it.
+ }
if(LOG.isDebugEnabled()) {
LOG.debug("region closed: " + regionInfo.regionName);
@@ -1340,7 +1504,9 @@
HRegionInterface server;
if (rootRegion) {
metaRegionName = HGlobals.rootRegionInfo.regionName;
- waitForRootRegion(); // Make sure root region available
+ if(waitForRootRegionOrClose()) {// Make sure root region available
+ return; // We're shutting down. Forget it.
+ }
server = client.getHRegionConnection(rootRegionLocation);
} else {
@@ -1404,15 +1570,19 @@
}
}
- /** PendingOpenReport is an open message that is saved in a different thread. */
+ /**
+ * PendingOpenReport is instantiated when a region server reports that it is
+ * serving a region. This applies to all meta and user regions except the
+ * root region which is handled specially.
+ */
private class PendingOpenReport extends PendingOperation {
private boolean rootRegion;
private Text regionName;
private BytesWritable serverAddress;
private BytesWritable startCode;
- public PendingOpenReport(HServerInfo info, Text regionName) {
- if(regionName.find(HGlobals.metaTableDesc.getName().toString()) == 0) {
+ PendingOpenReport(HServerInfo info, HRegionInfo region) {
+ if(region.tableDesc.getName().equals(META_TABLE_NAME)) {
// The region which just came on-line is a META region.
// We need to look in the ROOT region for its information.
@@ -1425,7 +1595,7 @@
this.rootRegion = false;
}
- this.regionName = regionName;
+ this.regionName = region.regionName;
try {
this.serverAddress = new BytesWritable(
@@ -1440,13 +1610,16 @@
}
- public void process() throws IOException {
+ @Override
+ void process() throws IOException {
for(int tries = 0; tries < numRetries; tries ++) {
// We can not access any meta region if they have not already been assigned
// and scanned.
- metaScanner.waitForMetaScan();
+ if(metaScanner.waitForMetaScanOrClose()) {
+ return; // We're shutting down. Forget it.
+ }
if(LOG.isDebugEnabled()) {
LOG.debug(regionName + " open on "
@@ -1459,7 +1632,9 @@
HRegionInterface server;
if(rootRegion) {
metaRegionName = HGlobals.rootRegionInfo.regionName;
- waitForRootRegion(); // Make sure root region available
+ if(waitForRootRegionOrClose()) {// Make sure root region available
+ return; // We're shutting down. Forget it.
+ }
server = client.getHRegionConnection(rootRegionLocation);
} else {
@@ -1489,24 +1664,7 @@
throw e;
}
}
- }
- }
- }
-
- synchronized void waitForRootRegion() {
- while (rootRegionLocation == null) {
- try {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Wait for root region");
- }
- wait();
- if (LOG.isDebugEnabled()) {
- LOG.debug("Wake from wait for root region");
- }
- } catch(InterruptedException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Wake from wait for root region (IE)");
- }
+ pendingRegions.remove(regionName);
}
}
}
@@ -1515,10 +1673,34 @@
// HMasterInterface
//////////////////////////////////////////////////////////////////////////////
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.hbase.HMasterInterface#isMasterRunning()
+ */
public boolean isMasterRunning() {
return !closed;
}
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.hbase.HMasterInterface#shutdown()
+ */
+ public void shutdown() {
+ TimerTask tt = new TimerTask() {
+ @Override
+ public void run() {
+ closed = true;
+ synchronized(msgQueue) {
+ msgQueue.clear(); // Empty the queue
+ msgQueue.notifyAll(); // Wake main thread
+ }
+ }
+ };
+ Timer t = new Timer("Shutdown");
+ t.schedule(tt, 10);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.hbase.HMasterInterface#createTable(org.apache.hadoop.hbase.HTableDescriptor)
+ */
public void createTable(HTableDescriptor desc) throws IOException {
if (!isMasterRunning()) {
throw new MasterNotRunningException();
@@ -1530,7 +1712,9 @@
// We can not access any meta region if they have not already been assigned
// and scanned.
- metaScanner.waitForMetaScan();
+ if(metaScanner.waitForMetaScanOrClose()) {
+ return; // We're shutting down. Forget it.
+ }
// 1. Check to see if table already exists
@@ -1630,25 +1814,6 @@
new ChangeTableState(tableName, true).process();
}
- /**
- * Turn off the HMaster. Sets a flag so that the main thread know to shut
- * things down in an orderly fashion.
- */
- public void shutdown() {
- TimerTask tt = new TimerTask() {
- @Override
- public void run() {
- closed = true;
- synchronized(msgQueue) {
- msgQueue.clear(); // Empty the queue
- msgQueue.notifyAll(); // Wake main thread
- }
- }
- };
- Timer t = new Timer("Shutdown");
- t.schedule(tt, 10);
- }
-
/* (non-Javadoc)
* @see org.apache.hadoop.hbase.HMasterInterface#findRootRegion()
*/
@@ -1682,7 +1847,9 @@
// We can not access any meta region if they have not already been
// assigned and scanned.
- metaScanner.waitForMetaScan();
+ if(metaScanner.waitForMetaScanOrClose()) {
+ return; // We're shutting down. Forget it.
+ }
Text firstMetaRegion = null;
if(knownMetaRegions.size() == 1) {
@@ -1698,7 +1865,7 @@
this.metaRegions.addAll(knownMetaRegions.tailMap(firstMetaRegion).values());
}
- public void process() throws IOException {
+ void process() throws IOException {
for(int tries = 0; tries < numRetries; tries++) {
boolean tableExists = false;
try {
@@ -1722,9 +1889,8 @@
String serverName = null;
long startCode = -1L;
- LabelledData[] values = null;
- HStoreKey key = new HStoreKey();
- values = server.next(scannerId, key);
+ KeyedData[] values = null;
+ values = server.next(scannerId);
if(values == null || values.length == 0) {
break;
}
@@ -1736,12 +1902,13 @@
}
System.arraycopy(values[i].getData().get(), 0, bytes, 0, bytes.length);
- if(values[i].getLabel().equals(COL_REGIONINFO)) {
+ Text column = values[i].getKey().getColumn();
+ if(column.equals(COL_REGIONINFO)) {
haveRegionInfo = true;
inbuf.reset(bytes, bytes.length);
info.readFields(inbuf);
- } else if(values[i].getLabel().equals(COL_SERVER)) {
+ } else if(column.equals(COL_SERVER)) {
try {
serverName = new String(bytes, UTF8_ENCODING);
@@ -1749,7 +1916,7 @@
assert(false);
}
- } else if(values[i].getLabel().equals(COL_STARTCODE)) {
+ } else if(column.equals(COL_STARTCODE)) {
try {
startCode = Long.valueOf(new String(bytes, UTF8_ENCODING));
@@ -1828,6 +1995,7 @@
throws IOException;
}
+ /** Instantiated to enable or disable a table */
private class ChangeTableState extends TableOperation {
private boolean online;
@@ -1836,11 +2004,12 @@
protected long lockid;
protected long clientId;
- public ChangeTableState(Text tableName, boolean onLine) throws IOException {
+ ChangeTableState(Text tableName, boolean onLine) throws IOException {
super(tableName);
this.online = onLine;
}
+ @Override
protected void processScanItem(String serverName, long startCode,
HRegionInfo info)
throws IOException {
@@ -1854,6 +2023,7 @@
}
}
+ @Override
protected void postProcessMeta(MetaRegion m, HRegionInterface server)
throws IOException {
// Process regions not being served
@@ -1958,13 +2128,19 @@
}
}
-
+
+ /**
+ * Instantiated to delete a table
+ * Note that it extends ChangeTableState, which takes care of disabling
+ * the table.
+ */
private class TableDelete extends ChangeTableState {
- public TableDelete(Text tableName) throws IOException {
+ TableDelete(Text tableName) throws IOException {
super(tableName, false);
}
+ @Override
protected void postProcessMeta(MetaRegion m, HRegionInterface server)
throws IOException {
// For regions that are being served, mark them for deletion
@@ -2001,9 +2177,11 @@
super(tableName);
}
+ @Override
protected void processScanItem(
@SuppressWarnings("unused") String serverName,
- @SuppressWarnings("unused") long startCode, final HRegionInfo info)
+ @SuppressWarnings("unused") long startCode,
+ final HRegionInfo info)
throws IOException {
if(isEnabled(info)) {
throw new TableNotDisabledException(tableName.toString());
@@ -2050,15 +2228,17 @@
}
}
}
-
+
+ /** Instantiated to remove a column family from a table */
private class DeleteColumn extends ColumnOperation {
private Text columnName;
- public DeleteColumn(Text tableName, Text columnName) throws IOException {
+ DeleteColumn(Text tableName, Text columnName) throws IOException {
super(tableName);
this.columnName = columnName;
}
+ @Override
protected void postProcessMeta(MetaRegion m, HRegionInterface server)
throws IOException {
@@ -2085,25 +2265,27 @@
}
}
}
-
+
+ /** Instantiated to add a column family to a table */
private class AddColumn extends ColumnOperation {
private HColumnDescriptor newColumn;
- public AddColumn(Text tableName, HColumnDescriptor newColumn)
+ AddColumn(Text tableName, HColumnDescriptor newColumn)
throws IOException {
super(tableName);
this.newColumn = newColumn;
}
+ @Override
protected void postProcessMeta(MetaRegion m, HRegionInterface server)
throws IOException {
for(HRegionInfo i: unservedRegions) {
- //TODO: I *think* all we need to do to add a column is add it to
- // the table descriptor. When the region is brought on-line, it
- // should find the column missing and create it.
+ // All we need to do to add a column is add it to the table descriptor.
+ // When the region is brought on-line, it will find the column missing
+ // and create it.
i.tableDesc.addFamily(newColumn);
updateRegionInfo(server, m.regionName, i);
@@ -2114,19 +2296,32 @@
//////////////////////////////////////////////////////////////////////////////
// Managing leases
//////////////////////////////////////////////////////////////////////////////
-
- private class ServerExpirer extends LeaseListener {
+
+ /** Instantiated to monitor the health of a region server */
+ private class ServerExpirer implements LeaseListener {
private String server;
- public ServerExpirer(String server) {
+ ServerExpirer(String server) {
this.server = server;
}
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.hbase.LeaseListener#leaseExpired()
+ */
public void leaseExpired() {
LOG.info(server + " lease expired");
HServerInfo storedInfo = serversToServerInfo.remove(server);
+ if(rootRegionLocation != null
+ && rootRegionLocation.toString().equals(
+ storedInfo.getServerAddress().toString())) {
+
+ rootRegionLocation = null;
+ unassignedRegions.put(HGlobals.rootRegionInfo.regionName,
+ HGlobals.rootRegionInfo);
+ assignAttempts.put(HGlobals.rootRegionInfo.regionName, 0L);
+ }
synchronized(msgQueue) {
- msgQueue.add(new PendingServerShutdown(storedInfo));
+ msgQueue.addLast(new PendingServerShutdown(storedInfo));
msgQueue.notifyAll();
}
}
@@ -2142,6 +2337,10 @@
System.exit(0);
}
+ /**
+ * Main program
+ * @param args
+ */
public static void main(String [] args) {
if (args.length < 1) {
printUsageAndExit();