NGDATA/hbase-indexer — pull request #17 by rvs (2013-10-14 22:28:30+00:00)
Title: 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 getServices() {
+ List bssi = new ArrayList(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 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 keyValuesPerRowKey = ArrayListMultimap.create();
final Map 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 keyValues = (List)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> futures) throws IOException {
```