HADOOP-1509 Open HRegionServer/HClient for extension

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@550634 13f79535-47bb-0310-9956-ffa450edef68

src/contrib/hbase/CHANGES.txt

@@ -35,3 +35,8 @@ Trunk (unreleased changes)
  20. HADOOP-1465 Add cluster stop/start scripts for hbase
  21. HADOOP-1415 Provide configurable per-column bloom filters - part 2.
  22. HADOOP-1498. Replace boxed types with primitives in many places.
+ 23. HADOOP-1509.  Made methods/inner classes in HRegionServer and HClient protected
+     instead of private for easier extension. Also made HRegion and HRegionInfo public too.
+     Added an hbase-default.xml property for specifying what HRegionInterface extension to use
+     for proxy server connection.
+ 

src/contrib/hbase/conf/hbase-default.xml

@@ -14,6 +14,13 @@
     <description>The host and port a HBase region server runs at.
     </description>
   </property>
+  <property>
+    <name>hbase.regionserver.class</name>
+    <value>org.apache.hadoop.hbase.HRegionInterface</value>
+    <description>An interface that is assignable to HRegionInterface.  Used in HClient for
+    opening proxy to remote region server.
+    </description>
+  </property>
   <property>
     <name>hbase.rootdir</name>
     <value>${hadoop.tmp.dir}/hbase</value>

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HClient.java

@@ -58,11 +58,12 @@ public class HClient implements HConstants {
   int numRetries;
   private HMasterInterface master;
   private final Configuration conf;
+  private Class<? extends HRegionInterface> serverInterfaceClass;
   
   /*
    * Data structure that holds current location for a region and its info.
    */
-  static class RegionLocation {
+  protected static class RegionLocation {
     HRegionInfo regionInfo;
     HServerAddress serverAddress;
 
@@ -76,6 +77,14 @@ public class HClient implements HConstants {
       return "address: " + this.serverAddress.toString() + ", regioninfo: " +
         this.regionInfo;
     }
+    
+    public HRegionInfo getRegionInfo(){
+      return regionInfo;
+    }
+
+    public HServerAddress getServerAddress(){
+      return serverAddress;
+    }
   }
   
   // Map tableName -> (Map startRow -> (HRegionInfo, HServerAddress)
@@ -94,6 +103,7 @@ public class HClient implements HConstants {
   Random rand;
   long clientid;
 
+
   /** 
    * Creates a new HClient
    * @param conf - Configuration object
@@ -116,7 +126,7 @@ public class HClient implements HConstants {
     this.rand = new Random();
   }
   
-  private void handleRemoteException(RemoteException e) throws IOException {
+  protected void handleRemoteException(RemoteException e) throws IOException {
     String msg = e.getMessage();
     if(e.getClassName().equals("org.apache.hadoop.hbase.InvalidColumnNameException")) {
       throw new InvalidColumnNameException(msg);
@@ -143,7 +153,7 @@ public class HClient implements HConstants {
   
   /* Find the address of the master and connect to it
    */
-  private void checkMaster() throws MasterNotRunningException {
+  protected void checkMaster() throws MasterNotRunningException {
     if (this.master != null) {
       return;
     }
@@ -531,7 +541,7 @@ public class HClient implements HConstants {
    * @param tableName - the table name to be checked
    * @throws IllegalArgumentException - if the table name is reserved
    */
-  private void checkReservedTableName(Text tableName) {
+  protected void checkReservedTableName(Text tableName) {
     if(tableName.equals(ROOT_TABLE_NAME)
         || tableName.equals(META_TABLE_NAME)) {
       
@@ -547,7 +557,7 @@ public class HClient implements HConstants {
   //////////////////////////////////////////////////////////////////////////////
   // Client API
   //////////////////////////////////////////////////////////////////////////////
-
+  
   /**
    * Loads information so that a table can be manipulated.
    * 
@@ -558,15 +568,29 @@ public class HClient implements HConstants {
     if(tableName == null || tableName.getLength() == 0) {
       throw new IllegalArgumentException("table name cannot be null or zero length");
     }
-    this.tableServers = tablesToServers.get(tableName);
-    if (this.tableServers == null ) {
+    this.tableServers = getTableServers(tableName);
+  }
+  
+  /**
+   * Gets the servers of the given table.
+   * 
+   * @param tableName - the table to be located
+   * @throws IOException - if the table can not be located after retrying
+   */
+  protected synchronized SortedMap<Text, RegionLocation> getTableServers(Text tableName) throws IOException {
+    if(tableName == null || tableName.getLength() == 0) {
+      throw new IllegalArgumentException("table name cannot be null or zero length");
+    }
+    SortedMap<Text, RegionLocation> serverResult  = tablesToServers.get(tableName);
+    if (serverResult == null ) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("No servers for " + tableName + ". Doing a find...");
       }
       // We don't know where the table is.
       // Load the information from meta.
-      this.tableServers = findServersForTable(tableName);
+      serverResult = findServersForTable(tableName);
     }
+    return serverResult;
   }
 
   /*
@@ -832,24 +856,39 @@ public class HClient implements HConstants {
     return servers;
   }
 
-  /* 
-   * Establishes a connection to the region server at the specified address
+  /** 
+   * Establishes a connection to the region server at the specified address.
    * @param regionServer - the server to connect to
    * @throws IOException
    */
-  synchronized HRegionInterface getHRegionConnection(HServerAddress regionServer)
-      throws IOException {
+  protected synchronized HRegionInterface getHRegionConnection(
+      HServerAddress regionServer) throws IOException{
 
-      // See if we already have a connection
+    getRegionServerInterface();
 
+    // See if we already have a connection
     HRegionInterface server = this.servers.get(regionServer.toString());
-    
-    if(server == null) {                                // Get a connection
-      
-      server = (HRegionInterface)RPC.waitForProxy(HRegionInterface.class, 
-          HRegionInterface.versionID, regionServer.getInetSocketAddress(),
-          this.conf);
-      
+
+    if (server == null) { // Get a connection
+      long versionId = 0;
+      try {
+        versionId = serverInterfaceClass.getDeclaredField("versionID").getLong(server);
+      } catch (IllegalAccessException e) {
+        // Should never happen unless visibility of versionID changes
+        throw new UnsupportedOperationException(
+            "Unable to open a connection to a " + serverInterfaceClass.getName() + " server.", e);
+      } catch (NoSuchFieldException e) {
+        // Should never happen unless versionID field name changes in HRegionInterface
+        throw new UnsupportedOperationException(
+            "Unable to open a connection to a " + serverInterfaceClass.getName() + " server.", e);
+      }
+
+      server = (HRegionInterface) RPC.waitForProxy(
+               serverInterfaceClass,
+               versionId,
+               regionServer.getInetSocketAddress(),
+               this.conf);
+
       this.servers.put(regionServer.toString(), server);
     }
     return server;
@@ -917,7 +956,7 @@ public class HClient implements HConstants {
    * @param row Row to find.
    * @return Location of row.
    */
-  synchronized RegionLocation getRegionLocation(Text row) {
+  protected synchronized RegionLocation getRegionLocation(Text row) {
     if(this.tableServers == null) {
       throw new IllegalStateException("Must open table first");
     }
@@ -1547,6 +1586,35 @@ public class HClient implements HConstants {
     }
     
     return errCode;
+  }    
+
+  /**
+   * Determine the region server interface to use from configuration properties.
+   *
+   */
+  @SuppressWarnings("unchecked")
+  private void getRegionServerInterface() {
+    if (this.serverInterfaceClass != null) {
+      return;
+    }
+
+    String serverClassName = this.conf.get(REGION_SERVER_CLASS,
+                                           DEFAULT_REGION_SERVER_CLASS);
+
+    try {
+      this.serverInterfaceClass = (Class<? extends HRegionInterface>) Class
+                                                                           .forName(serverClassName);
+    } catch (ClassNotFoundException e) {
+      throw new UnsupportedOperationException(
+            "Unable to find region server interface " + serverClassName, e);
+    }
+  }
+
+  /**
+   * @return the configuration for this client
+   */
+  protected Configuration getConf(){
+    return conf;
   }
   
   /**
@@ -1558,4 +1626,5 @@ public class HClient implements HConstants {
     int errCode = (new HClient(c)).doCommandLine(args);
     System.exit(errCode);
   }
+
 }

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java

@@ -47,7 +47,13 @@ public interface HConstants {
   static final String REGIONSERVER_ADDRESS = "hbase.regionserver";
   
   /** Default region server address */
-  static final String DEFAULT_REGIONSERVER_ADDRESS = DEFAULT_HOST + ":60010";
+  static final String DEFAULT_REGIONSERVER_ADDRESS = DEFAULT_HOST + ":60010";  
+
+  /** Parameter name for what region server interface to use. */
+  static final String REGION_SERVER_CLASS = "hbase.regionserver.class";
+  
+  /** Default region server interface class name. */
+  static final String DEFAULT_REGION_SERVER_CLASS = HRegionInterface.class.getName();
 
   /** Parameter name for how often threads should wake up */
   static final String THREAD_WAKE_FREQUENCY = "hbase.server.thread.wakefrequency";

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java

@@ -659,7 +659,7 @@ public class HMaster implements HConstants, HMasterInterface,
   /** 
    * @return HServerAddress of the master server
    */
-  HServerAddress getMasterAddress() {
+  public HServerAddress getMasterAddress() {
     return address;
   }
 

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java

@@ -55,7 +55,7 @@ import java.util.*;
  * regionName is a unique identifier for this HRegion. (startKey, endKey]
  * defines the keyspace for this HRegion.
  */
-class HRegion implements HConstants {
+public class HRegion implements HConstants {
   static String SPLITDIR = "splits";
   static String MERGEDIR = "merges";
   static String TMPREGION_PREFIX = "tmpregion_";
@@ -289,7 +289,6 @@ class HRegion implements HConstants {
    * the supplied path.
    * 
    * @param rootDir root directory for HBase instance
-   * @param log HLog where changes should be committed
    * @param fs is the filesystem.  
    * @param conf is global configuration settings.
    * @param regionInfo - HRegionInfo that describes the region
@@ -298,7 +297,7 @@ class HRegion implements HConstants {
    * 
    * @throws IOException
    */
-  HRegion(Path rootDir, HLog log, FileSystem fs, Configuration conf, 
+  public HRegion(Path rootDir, HLog log, FileSystem fs, Configuration conf, 
       HRegionInfo regionInfo, Path initialFiles)
   throws IOException {
     
@@ -386,7 +385,7 @@ class HRegion implements HConstants {
    * This method could take some time to execute, so don't call it from a 
    * time-sensitive thread.
    */
-  Vector<HStoreFile> close() throws IOException {
+  public Vector<HStoreFile> close() throws IOException {
     lock.obtainWriteLock();
     try {
       boolean shouldClose = false;
@@ -548,43 +547,43 @@ class HRegion implements HConstants {
   // HRegion accessors
   //////////////////////////////////////////////////////////////////////////////
 
-  Text getStartKey() {
+  public Text getStartKey() {
     return regionInfo.startKey;
   }
   
-  Text getEndKey() {
+  public Text getEndKey() {
     return regionInfo.endKey;
   }
   
-  long getRegionId() {
+  public long getRegionId() {
     return regionInfo.regionId;
   }
 
-  Text getRegionName() {
+  public Text getRegionName() {
     return regionInfo.regionName;
   }
   
-  Path getRootDir() {
+  public Path getRootDir() {
     return rootDir;
   }
  
-  HTableDescriptor getTableDesc() {
+  public HTableDescriptor getTableDesc() {
     return regionInfo.tableDesc;
   }
   
-  HLog getLog() {
+  public HLog getLog() {
     return log;
   }
   
-  Configuration getConf() {
+  public Configuration getConf() {
     return conf;
   }
   
-  Path getRegionDir() {
+  public Path getRegionDir() {
     return regiondir;
   }
   
-  FileSystem getFilesystem() {
+  public FileSystem getFilesystem() {
     return fs;
   }
 
@@ -973,7 +972,7 @@ class HRegion implements HConstants {
    * Return an iterator that scans over the HRegion, returning the indicated 
    * columns.  This Iterator must be closed by the caller.
    */
-  HInternalScannerInterface getScanner(Text[] cols, Text firstRow)
+  public HInternalScannerInterface getScanner(Text[] cols, Text firstRow)
   throws IOException {
     lock.obtainReadLock();
     try {
@@ -1009,9 +1008,9 @@ class HRegion implements HConstants {
    * 
    * @param row Row to update
    * @return lockid
-   * @see #put(long, Text, BytesWritable)
+   * @see #put(long, Text, byte[])
    */
-  long startUpdate(Text row) throws IOException {
+  public long startUpdate(Text row) throws IOException {
     // We obtain a per-row lock, so other clients will block while one client
     // performs an update.  The read lock is released by the client calling
     // #commit or #abort or if the HRegionServer lease on the lock expires.
@@ -1029,7 +1028,7 @@ class HRegion implements HConstants {
    * This method really just tests the input, then calls an internal localput() 
    * method.
    */
-  void put(long lockid, Text targetCol, byte [] val) throws IOException {
+  public void put(long lockid, Text targetCol, byte [] val) throws IOException {
     if (DELETE_BYTES.compareTo(val) == 0) {
       throw new IOException("Cannot insert value: " + val);
     }
@@ -1039,7 +1038,7 @@ class HRegion implements HConstants {
   /**
    * Delete a value or write a value. This is a just a convenience method for put().
    */
-  void delete(long lockid, Text targetCol) throws IOException {
+  public void delete(long lockid, Text targetCol) throws IOException {
     localput(lockid, targetCol, DELETE_BYTES.get());
   }
 
@@ -1090,7 +1089,7 @@ class HRegion implements HConstants {
    * writes associated with the given row-lock.  These values have not yet
    * been placed in memcache or written to the log.
    */
-  void abort(long lockid) throws IOException {
+  public void abort(long lockid) throws IOException {
     Text row = getRowFromLock(lockid);
     if(row == null) {
       throw new LockException("No write lock for lockid " + lockid);
@@ -1124,7 +1123,7 @@ class HRegion implements HConstants {
    * @param lockid Lock for row we're to commit.
    * @throws IOException
    */
-  void commit(final long lockid) throws IOException {
+  public void commit(final long lockid) throws IOException {
     // Remove the row from the pendingWrites list so 
     // that repeated executions won't screw this up.
     Text row = getRowFromLock(lockid);

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java

@@ -139,6 +139,41 @@ public class HRegionInfo implements WritableComparable {
     this.regionName.readFields(in);
     this.offLine = in.readBoolean();
   }
+  
+  /**
+   * @return the endKey
+   */
+  public Text getEndKey(){
+    return endKey;
+  }
+
+  /**
+   * @return the regionId
+   */
+  public long getRegionId(){
+    return regionId;
+  }
+
+  /**
+   * @return the regionName
+   */
+  public Text getRegionName(){
+    return regionName;
+  }
+
+  /**
+   * @return the startKey
+   */
+  public Text getStartKey(){
+    return startKey;
+  }
+
+  /**
+   * @return the tableDesc
+   */
+  public HTableDescriptor getTableDesc(){
+    return tableDesc;
+  }
 
   //////////////////////////////////////////////////////////////////////////////
   // Comparable

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java

@@ -468,7 +468,7 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
    * Sets a flag that will cause all the HRegionServer threads to shut down
    * in an orderly fashion.
    */
-  synchronized void stop() {
+  public synchronized void stop() {
     stopRequested = true;
     notifyAll();                        // Wakes run() if it is sleeping
   }
@@ -1079,25 +1079,25 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
   }
 
   /** 
-   * Private utility method for safely obtaining an HRegion handle.
+   * Protected utility method for safely obtaining an HRegion handle.
    * @param regionName Name of online {@link HRegion} to return
    * @return {@link HRegion} for <code>regionName</code>
    * @throws NotServingRegionException
    */
-  private HRegion getRegion(final Text regionName)
+  protected HRegion getRegion(final Text regionName)
   throws NotServingRegionException {
     return getRegion(regionName, false);
   }
   
   /** 
-   * Private utility method for safely obtaining an HRegion handle.
+   * Protected utility method for safely obtaining an HRegion handle.
    * @param regionName Name of online {@link HRegion} to return
    * @param checkRetiringRegions Set true if we're to check retiring regions
    * as well as online regions.
    * @return {@link HRegion} for <code>regionName</code>
    * @throws NotServingRegionException
    */
-  private HRegion getRegion(final Text regionName,
+  protected HRegion getRegion(final Text regionName,
       final boolean checkRetiringRegions)
   throws NotServingRegionException {
     HRegion region = null;