Fix various CDH5 integration problems.

4 files changed, 25 insertions, 17 deletions

diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java b/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java
index 332bbd7..ae9db71 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java
@@ -2,6 +2,8 @@ package parallelai.spyglass.hbase;
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -14,7 +16,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionLocation;
-import org.apache.hadoop.hbase.HServerAddress;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.util.Addressing;
@@ -110,15 +111,15 @@ public class HBaseInputFormatGranular extends HBaseInputFormatBase {
 					: maxKey;
 
 			HRegionLocation regionLoc = table.getRegionLocation(keys.getFirst()[i]);
-            HServerAddress regionServerAddress = regionLoc.getServerAddress();
-			InetAddress regionAddress = regionServerAddress.getInetSocketAddress().getAddress();
+            String regionServerHostnamePort = regionLoc.getHostnamePort();
+			InetAddress regionAddress = toInetAddress(regionServerHostnamePort);
 			String regionLocation;
 			try {
 				regionLocation = reverseDNS(regionAddress);
 			} catch (NamingException e) {
 				LOG.error("Cannot resolve the host name for " + regionAddress
 						+ " because of " + e);
-				regionLocation = regionServerAddress.getHostname();
+				regionLocation = toHostname(regionServerHostnamePort);
 			}
 
             regionNames[i] = regionLoc.getRegionInfo().getRegionNameAsString();
@@ -189,17 +190,17 @@ public class HBaseInputFormatGranular extends HBaseInputFormatBase {
 						byte[] rStart = cRegion.getRegionInfo().getStartKey();
 						byte[] rStop = cRegion.getRegionInfo().getEndKey();
 
-						HServerAddress regionServerAddress = cRegion
-								.getServerAddress();
-						InetAddress regionAddress = regionServerAddress
-								.getInetSocketAddress().getAddress();
+						String regionServerHostnamePort = cRegion
+								.getHostnamePort();
+						InetAddress regionAddress =
+								toInetAddress(regionServerHostnamePort);
 						String regionLocation;
 						try {
 							regionLocation = reverseDNS(regionAddress);
 						} catch (NamingException e) {
 							LOG.error("Cannot resolve the host name for "
 									+ regionAddress + " because of " + e);
-							regionLocation = regionServerAddress.getHostname();
+							regionLocation = toHostname(regionServerHostnamePort);
 						}
 
                         String regionName = cRegion.getRegionInfo().getRegionNameAsString();
@@ -361,6 +362,14 @@ public class HBaseInputFormatGranular extends HBaseInputFormatBase {
 		}
 	}
 
+	private String toHostname(String regionServerHostnamePort) {
+		return Addressing.parseHostname(regionServerHostnamePort);
+	}
+
+	private InetAddress toInetAddress(String regionServerHostnamePort) throws UnknownHostException {
+		return InetAddress.getByName(toHostname(regionServerHostnamePort));
+	}
+
 	private String reverseDNS(InetAddress ipAddress) throws NamingException {
 		String hostName = this.reverseDNSCacheMap.get(ipAddress);
 		if (hostName == null) {
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java b/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java
index efe548d..ece7e61 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java
@@ -156,7 +156,7 @@ public class HBaseRawTap extends Tap<JobConf, RecordReader, OutputCollector> {
 		return new Path(SCHEME + ":/" + tableName.replaceAll(":", "_"));
 	}
 
-	protected HBaseAdmin getHBaseAdmin(JobConf conf) throws MasterNotRunningException, ZooKeeperConnectionException {
+	protected HBaseAdmin getHBaseAdmin(JobConf conf) throws IOException {
 		if (hBaseAdmin == null) {
 			Configuration hbaseConf = HBaseConfiguration.create(conf);
 			hBaseAdmin = new HBaseAdmin(hbaseConf);
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java
index b34ed3f..8a9559f 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java
@@ -10,7 +10,6 @@ import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
 import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Writables;
 import org.apache.hadoop.util.StringUtils;
 
 import java.io.IOException;
@@ -251,7 +250,7 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
           }
           
           lastSuccessfulRow = key.get();
-          Writables.copyWritable(result, value);
+          value.copyFrom(result);
           return true;
         }
         return false;
@@ -302,8 +301,8 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
               key.set(result.getRow());
             }
             lastSuccessfulRow = key.get();
-            Writables.copyWritable(result, value);
-            
+            value.copyFrom(result);
+
             return true;
           } else {
             LOG.debug(" Key ("+ Bytes.toString(nextKey)+ ") return an EMPTY result. Get ("+theGet.getId()+")" ); //alg0
@@ -352,7 +351,7 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
             key.set(result.getRow());
           }
           lastSuccessfulRow = key.get();
-          Writables.copyWritable(result, value);
+          value.copyFrom(result);
 
           return true;
         } else {
@@ -408,7 +407,7 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase {
                 key.set(result.getRow());
               }
               lastSuccessfulRow = key.get();
-              Writables.copyWritable(result, value);
+              value.copyFrom(result);
               return true;
             } else {
               LOG.debug(String.format("+ Key (%s) return an EMPTY result. Get (%s)", Bytes.toString(nextKey), theGet.getId()) ); //alg0
diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
index e4a3f78..e1897ef 100644
--- a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
+++ b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java
@@ -143,7 +143,7 @@ public class HBaseTap extends Tap<JobConf, RecordReader, OutputCollector> {
     return new Path(SCHEME + ":/" + tableName.replaceAll(":", "_"));
   }
 
-  protected HBaseAdmin getHBaseAdmin(JobConf conf) throws MasterNotRunningException, ZooKeeperConnectionException {
+  protected HBaseAdmin getHBaseAdmin(JobConf conf) throws IOException {
     if (hBaseAdmin == null) {
       Configuration hbaseConf = HBaseConfiguration.create(conf);
       hBaseAdmin = new HBaseAdmin(hbaseConf);