/*
 * Copyright 2012-2013 Continuuity,Inc. All Rights Reserved.
 */
package com.continuuity.data2.transaction.queue;

import com.google.common.base.Stopwatch;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableExistsException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 *
 */
public final class HBaseQueueUtils {

  private static final Logger LOG = LoggerFactory.getLogger(HBaseQueueUtils.class);

  /**
   * Creates a HBase table if the table doesn't exists.
   * @param admin
   * @param tableName
   * @param maxWaitMs
   * @throws IOException
   */
  public static void createTableIfNotExists(HBaseAdmin admin, String tableName,
                                            byte[] columnFamily, long maxWaitMs) throws IOException {
    if (!admin.tableExists(tableName)) {
      HTableDescriptor htd = new HTableDescriptor(tableName);
      HColumnDescriptor hcd = new HColumnDescriptor(columnFamily);
      htd.addFamily(hcd);
      try {
        LOG.info("Creating queue table '{}'", tableName);
        admin.createTable(htd);
        return;
      } catch (TableExistsException e) {
        // table may exist because someone else is creating it at the same
        // time. But it may not be available yet, and opening it might fail.
        LOG.info("Failed to create queue table '{}'. {}.", tableName, e.getMessage(), e);
      }

      // Wait for table to materialize
      try {
        Stopwatch stopwatch = new Stopwatch();
        stopwatch.start();
        while (stopwatch.elapsedTime(TimeUnit.MILLISECONDS) < maxWaitMs) {
          if (admin.tableExists(tableName)) {
            LOG.info("Table '{}' exists now. Assuming that another process concurrently created it.", tableName);
            return;
          } else {
            TimeUnit.MILLISECONDS.sleep(100);
          }
        }
      } catch (InterruptedException e) {
        LOG.warn("Sleeping thread interrupted.");
      }
      LOG.error("Table '{}' does not exist after waiting {} ms. Giving up.", tableName, maxWaitMs);
    }
  }

}

@@ -111,16 +111,63 @@
 queue;%0A%0A
+import com.continuuity.common.queue.QueueName;%0A
 import c
@@ -483,24 +483,49 @@
 OException;%0A
+import java.util.Arrays;%0A
 import java.
@@ -2305,10 +2305,349 @@
  %7D%0A  %7D%0A%0A
+%0A  public static byte%5B%5D getQueueRowPrefix(QueueName queueName) %7B%0A    byte%5B%5D bytes = Arrays.copyOf(queueName.toBytes(), queueName.toBytes().length);%0A    int i = 0;%0A    int j = bytes.length - 1;%0A    while (i %3C j) %7B%0A      byte tmp = bytes%5Bi%5D;%0A      bytes%5Bi%5D = bytes%5Bj%5D;%0A      bytes%5Bj%5D = tmp;%0A      i++;%0A      j--;%0A    %7D%0A    return bytes;%0A  %7D%0A
 %7D%0A

commit the missing function