Polishing support of kerberized HBase clusters

Gabriel, please see the latest patch attached. What I did, I unified your work (on HBase 0.94) with mine (HBase 0.95+). Here's an interesting diff between the two implementations for you to consider (as you can see the two are now very close to each other where it counts):
```
$ diff -u hbase-sep/hbase-sep-impl-0.9*/src/main/java/com/ngdata/sep/impl/SepConsumer.java
@@ -91,7 +104,7 @@
      *                 that the RPC will bind to.
      */
     public SepConsumer(String subscriptionId, long subscriptionTimestamp, EventListener listener, int threadCnt,
-            String hostName, ZooKeeperItf zk, Configuration hbaseConf) throws IOException {
+            String hostName, ZooKeeperItf zk, Configuration hbaseConf) throws IOException, InterruptedException {
         this(subscriptionId, subscriptionTimestamp, listener, threadCnt, hostName, zk, hbaseConf, null);
     }
 
@@ -104,7 +117,7 @@
      * @param payloadExtractor extracts payloads to include in SepEvents
      */
     public SepConsumer(String subscriptionId, long subscriptionTimestamp, EventListener listener, int threadCnt,
-            String hostName, ZooKeeperItf zk, Configuration hbaseConf, PayloadExtractor payloadExtractor) throws IOException {
+            String hostName, ZooKeeperItf zk, Configuration hbaseConf, PayloadExtractor payloadExtractor) throws IOException, InterruptedException {
         Preconditions.checkArgument(threadCnt > 0, "Thread count must be > 0");
         this.subscriptionId = SepModelImpl.toInternalSubscriptionName(subscriptionId);
         this.subscriptionTimestamp = subscriptionTimestamp;
@@ -114,14 +127,18 @@
         this.sepMetrics = new SepMetrics(subscriptionId);
         this.payloadExtractor = payloadExtractor;
         this.executors = Lists.newArrayListWithCapacity(threadCnt);
-        
-        // TODO see same call in HBase's HRegionServer:
-        // - should we do HBaseRPCErrorHandler ?
-        rpcServer = HBaseRPC.getServer(this, new Class<?>[] { HRegionInterface.class }, hostName, 0, /* ephemeral port */
-                10, // TODO how many handlers do we need? make it configurable?
-                10, false, // TODO make verbose flag configurable
-                hbaseConf, 0); // TODO need to check what this parameter is for
-        
+
+        InetSocketAddress initialIsa = new InetSocketAddress(hostName, 0);
+        if (initialIsa.getAddress() == null) {
+          throw new IllegalArgumentException("Failed resolve of " + initialIsa);
+        }
+        String name = "regionserver/" + initialIsa.toString();
+        this.rpcServer = new RpcServer(this, name, getServices(),
+        /*HBaseRPCErrorHandler.class, OnlineRegions.class},*/
+          initialIsa, // BindAddress is IP we got for this server.
+          hbaseConf.getInt("hbase.regionserver.handler.count", 10),
+          hbaseConf.getInt("hbase.regionserver.metahandler.count", 10),
+          hbaseConf, HConstants.QOS_THRESHOLD);
         this.serverName = new ServerName(hostName, rpcServer.getListenerAddress().getPort(), System.currentTimeMillis());
         this.zkWatcher = new ZooKeeperWatcher(hbaseConf, this.serverName.toString(), null);
 
@@ -153,6 +170,14 @@
         this.running = true;
     }
 
+    private List<RpcServer.BlockingServiceAndInterface> getServices() {
+      List<RpcServer.BlockingServiceAndInterface> bssi = new ArrayList<RpcServer.BlockingServiceAndInterface>(1);
+      bssi.add(new RpcServer.BlockingServiceAndInterface(
+          AdminProtos.AdminService.newReflectiveBlockingService(this),
+          AdminProtos.AdminService.BlockingInterface.class));
+      return bssi;
+    }
+
     public void stop() {
         Closer.close(zkWatcher);
         if (running) {
@@ -177,27 +202,41 @@
     }
 
     @Override
-    public void replicateLogEntries(HLog.Entry[] entries) throws IOException {
+    public AdminProtos.ReplicateWALEntryResponse replicateWALEntry(final RpcController controller,
+                                  final AdminProtos.ReplicateWALEntryRequest request) throws ServiceException {
+      try {
 
         // TODO Recording of last processed timestamp won't work if two batches of log entries are sent out of order
         long lastProcessedTimestamp = -1;
 
         SepEventExecutor eventExecutor = new SepEventExecutor(listener, executors, 100, sepMetrics);
-        
-        for (final HLog.Entry entry : entries) {
-            final HLogKey entryKey = entry.getKey();
-            if (entryKey.getWriteTime() < subscriptionTimestamp) {
-                continue;
-            }
-            byte[] tableName = entryKey.getTablename();
+
+        List<AdminProtos.WALEntry> entries = request.getEntryList();
+        CellScanner cells = ((PayloadCarryingRpcController)controller).cellScanner();
+
+        for (final AdminProtos.WALEntry entry : entries) {
+            TableName tableName = (entry.getKey().getWriteTime() < subscriptionTimestamp) ? null :
+                                  TableName.valueOf(entry.getKey().getTableName().toByteArray());
             Multimap<ByteBuffer, KeyValue> keyValuesPerRowKey = ArrayListMultimap.create();
             final Map<ByteBuffer, byte[]> payloadPerRowKey = Maps.newHashMap();
-            for (final KeyValue kv : entry.getEdit().getKeyValues()) {
-                ByteBuffer rowKey = ByteBuffer.wrap(kv.getBuffer(), kv.getRowOffset(), kv.getRowLength());
+            int count = entry.getAssociatedCellCount();
+            for (int i = 0; i < count; i++) {
+                if (!cells.advance()) {
+                    throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i);
+                }
+
+                // this signals to us that we simply need to skip over count of cells
+                if (tableName == null) {
+                    continue;
+                }
+
+                Cell cell = cells.current();
+                ByteBuffer rowKey = ByteBuffer.wrap(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                 byte[] payload;
-                if (payloadExtractor != null && (payload = payloadExtractor.extractPayload(tableName, kv)) != null) {
+                KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
+                if (payloadExtractor != null && (payload = payloadExtractor.extractPayload(tableName.toBytes(), kv)) != null) {
                     if (payloadPerRowKey.containsKey(rowKey)) {
-                        log.error("Multiple payloads encountered for row " + Bytes.toStringBinary(kv.getRow())
+                        log.error("Multiple payloads encountered for row " + Bytes.toStringBinary(rowKey)
                                 + ", choosing " + Bytes.toStringBinary(payloadPerRowKey.get(rowKey)));
                     } else {
                         payloadPerRowKey.put(rowKey, payload);
@@ -209,7 +248,7 @@
             for (final ByteBuffer rowKeyBuffer : keyValuesPerRowKey.keySet()) {
                 final List<KeyValue> keyValues = (List<KeyValue>)keyValuesPerRowKey.get(rowKeyBuffer);
 
-                final SepEvent sepEvent = new SepEvent(tableName, keyValues.get(0).getRow(), keyValues,
+                final SepEvent sepEvent = new SepEvent(tableName.toBytes(), keyValues.get(0).getRow(), keyValues,
                         payloadPerRowKey.get(rowKeyBuffer));
                 eventExecutor.scheduleSepEvent(sepEvent);
                 lastProcessedTimestamp = Math.max(lastProcessedTimestamp, entry.getKey().getWriteTime());
@@ -222,6 +261,10 @@
         if (lastProcessedTimestamp > 0) {
             sepMetrics.reportSepTimestamp(lastProcessedTimestamp);
         }
+        return AdminProtos.ReplicateWALEntryResponse.newBuilder().build();
+      } catch (IOException ie) {
+        throw new ServiceException(ie);
+      }
     }
 
     private void waitOnSepEventCompletion(List<Future<?>> futures) throws IOException {
```