/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.s2graph.core.storage.hbase


import java.util
import java.util.Base64
import java.util.concurrent.{ExecutorService, Executors, TimeUnit}

import com.stumbleupon.async.{Callback, Deferred}
import com.typesafe.config.Config
import org.apache.commons.io.FileUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client.{Admin, ConnectionFactory, Durability}
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding
import org.apache.hadoop.hbase.regionserver.BloomType
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.security.UserGroupInformation
import org.apache.s2graph.core._
import org.apache.s2graph.core.mysqls.{Label, LabelMeta, ServiceColumn}
import org.apache.s2graph.core.storage._
import org.apache.s2graph.core.storage.hbase.AsynchbaseStorage.{AsyncRPC, ScanWithRange}
import org.apache.s2graph.core.types.{HBaseType, VertexId}
import org.apache.s2graph.core.utils._
import org.hbase.async.FilterList.Operator.MUST_PASS_ALL
import org.hbase.async._

import scala.collection.JavaConversions._
import scala.collection.mutable.ArrayBuffer
import scala.concurrent._
import scala.concurrent.duration.Duration
import scala.util.Try
import scala.util.control.NonFatal
import scala.util.hashing.MurmurHash3


object AsynchbaseStorage {
  val vertexCf = Serializable.vertexCf
  val edgeCf = Serializable.edgeCf
  val emptyKVs = new util.ArrayList[KeyValue]()

  AsynchbasePatcher.init()

  def makeClient(config: Config, overrideKv: (String, String)*) = {
    val asyncConfig: org.hbase.async.Config =
      if (config.hasPath("hbase.security.auth.enable") && config.getBoolean("hbase.security.auth.enable")) {
        val krb5Conf = config.getString("java.security.krb5.conf")
        val jaas = config.getString("java.security.auth.login.config")

        System.setProperty("java.security.krb5.conf", krb5Conf)
        System.setProperty("java.security.auth.login.config", jaas)
        new org.hbase.async.Config()
      } else {
        new org.hbase.async.Config()
      }

    for (entry <- config.entrySet() if entry.getKey.contains("hbase")) {
      asyncConfig.overrideConfig(entry.getKey, entry.getValue.unwrapped().toString)
    }

    for ((k, v) <- overrideKv) {
      asyncConfig.overrideConfig(k, v)
    }

    val client = new HBaseClient(asyncConfig)
    logger.info(s"Asynchbase: ${client.getConfig.dumpConfiguration()}")
    client
  }

  def shutdown(client: HBaseClient): Unit = {
    client.shutdown().join()
  }

  case class ScanWithRange(scan: Scanner, offset: Int, limit: Int)
  type AsyncRPC = Either[GetRequest, ScanWithRange]

  def initLocalHBase(config: Config,
                     overwrite: Boolean = true): ExecutorService = {
    import java.io.{File, IOException}
    import java.net.Socket

    lazy val hbaseExecutor = {
      val executor = Executors.newSingleThreadExecutor()

      Runtime.getRuntime.addShutdownHook(new Thread() {
        override def run(): Unit = {
          executor.shutdown()
        }
      })

      val hbaseAvailable = try {
        val (host, port) = config.getString("hbase.zookeeper.quorum").split(":") match {
          case Array(h, p) => (h, p.toInt)
          case Array(h) => (h, 2181)
        }

        val socket = new Socket(host, port)
        socket.close()
        logger.info(s"HBase is available.")
        true
      } catch {
        case e: IOException =>
          logger.info(s"HBase is not available.")
          false
      }

      if (!hbaseAvailable) {
        // start HBase
        executor.submit(new Runnable {
          override def run(): Unit = {
            logger.info(s"HMaster starting...")
            val ts = System.currentTimeMillis()
            val cwd = new File(".").getAbsolutePath
            if (overwrite) {
              val dataDir = new File(s"$cwd/storage/s2graph")
              FileUtils.deleteDirectory(dataDir)
            }

            System.setProperty("proc_master", "")
            System.setProperty("hbase.log.dir", s"$cwd/storage/s2graph/hbase/")
            System.setProperty("hbase.log.file", s"$cwd/storage/s2graph/hbase.log")
            System.setProperty("hbase.tmp.dir", s"$cwd/storage/s2graph/hbase/")
            System.setProperty("hbase.home.dir", "")
            System.setProperty("hbase.id.str", "s2graph")
            System.setProperty("hbase.root.logger", "INFO,RFA")

            org.apache.hadoop.hbase.master.HMaster.main(Array[String]("start"))
            logger.info(s"HMaster startup finished: ${System.currentTimeMillis() - ts}")
          }
        })
      }

      executor
    }

    hbaseExecutor
  }
}


class AsynchbaseStorage(override val graph: S2Graph,
                        override val config: Config)(implicit ec: ExecutionContext)
  extends Storage[AsyncRPC, Deferred[StepResult]](graph, config) {

  import Extensions.DeferOps

  val hbaseExecutor: ExecutorService  =
    if (config.getString("hbase.zookeeper.quorum") == "localhost")
      AsynchbaseStorage.initLocalHBase(config)
    else
      null

  /**
   * Asynchbase client setup.
   * note that we need two client, one for bulk(withWait=false) and another for withWait=true
   */
  private val clientFlushInterval = config.getInt("hbase.rpcs.buffered_flush_interval").toString().toShort

  /**
   * since some runtime environment such as spark cluster has issue with guava version, that is used in Asynchbase.
   * to fix version conflict, make this as lazy val for clients that don't require hbase client.
   */
  lazy val client = AsynchbaseStorage.makeClient(config)
  lazy val clientWithFlush = AsynchbaseStorage.makeClient(config, "hbase.rpcs.buffered_flush_interval" -> "0")
  lazy val clients = Seq(client, clientWithFlush)

  private val emptyKeyValues = new util.ArrayList[KeyValue]()
  private val emptyKeyValuesLs = new util.ArrayList[util.ArrayList[KeyValue]]()
  private val emptyStepResult = new util.ArrayList[StepResult]()

  private def client(withWait: Boolean): HBaseClient = if (withWait) clientWithFlush else client

  import CanDefer._

  /** Future Cache to squash request */
  lazy private val futureCache = new DeferCache[StepResult, Deferred, Deferred](config, StepResult.Empty, "AsyncHbaseFutureCache", useMetric = true)

  /** Simple Vertex Cache */
  lazy private val vertexCache = new DeferCache[Seq[SKeyValue], Promise, Future](config, Seq.empty[SKeyValue])

  private val zkQuorum = config.getString("hbase.zookeeper.quorum")
  private val zkQuorumSlave =
    if (config.hasPath("hbase.slave.zookeeper.quorum")) Option(config.getString("hbase.slave.zookeeper.quorum"))
    else None

  /** v4 max next row size */
  private val v4_max_num_rows = 10000
  private def getV4MaxNumRows(limit : Int): Int = {
    if (limit < v4_max_num_rows) limit
    else v4_max_num_rows
  }

  /**
   * fire rpcs into proper hbase cluster using client and
   * return true on all mutation success. otherwise return false.
   */
  override def writeToStorage(cluster: String, kvs: Seq[SKeyValue], withWait: Boolean): Future[Boolean] = {
    if (kvs.isEmpty) Future.successful(true)
    else {
      val _client = client(withWait)
      val (increments, putAndDeletes) = kvs.partition(_.operation == SKeyValue.Increment)

      /* Asynchbase IncrementRequest does not implement HasQualifiers */
      val incrementsFutures = increments.map { kv =>
        val inc = new AtomicIncrementRequest(kv.table, kv.row, kv.cf, kv.qualifier, Bytes.toLong(kv.value))
        val defer = _client.atomicIncrement(inc)
        val future = defer.toFuture(Long.box(0)).map(_ => true).recover { case ex: Exception =>
          logger.error(s"mutation failed. $kv", ex)
          false
        }
        if (withWait) future else Future.successful(true)
      }

      /* PutRequest and DeleteRequest accept byte[][] qualifiers/values. */
      val othersFutures = putAndDeletes.groupBy { kv =>
        (kv.table.toSeq, kv.row.toSeq, kv.cf.toSeq, kv.operation, kv.timestamp)
      }.map { case ((table, row, cf, operation, timestamp), groupedKeyValues) =>

        val durability = groupedKeyValues.head.durability
        val qualifiers = new ArrayBuffer[Array[Byte]]()
        val values = new ArrayBuffer[Array[Byte]]()

        groupedKeyValues.foreach { kv =>
          if (kv.qualifier != null) qualifiers += kv.qualifier
          if (kv.value != null) values += kv.value
        }
        val defer = operation match {
          case SKeyValue.Put =>
            val put = new PutRequest(table.toArray, row.toArray, cf.toArray, qualifiers.toArray, values.toArray, timestamp)
            put.setDurable(durability)
            _client.put(put)
          case SKeyValue.Delete =>
            val delete =
              if (qualifiers.isEmpty)
                new DeleteRequest(table.toArray, row.toArray, cf.toArray, timestamp)
              else
                new DeleteRequest(table.toArray, row.toArray, cf.toArray, qualifiers.toArray, timestamp)
            delete.setDurable(durability)
            _client.delete(delete)
        }
        if (withWait) {
          defer.toFuture(new AnyRef()).map(_ => true).recover { case ex: Exception =>
            groupedKeyValues.foreach { kv => logger.error(s"mutation failed. $kv", ex) }
            false
          }
        } else Future.successful(true)
      }
      for {
        incrementRets <- Future.sequence(incrementsFutures)
        otherRets <- Future.sequence(othersFutures)
      } yield (incrementRets ++ otherRets).forall(identity)
    }
  }

  private def fetchKeyValues(rpc: AsyncRPC): Future[Seq[SKeyValue]] = {
    val defer = fetchKeyValuesInner(rpc)
    defer.toFuture(emptyKeyValues).map { kvsArr =>
      kvsArr.map { kv =>
        implicitly[CanSKeyValue[KeyValue]].toSKeyValue(kv)
      }
    }
  }

  override def fetchSnapshotEdgeKeyValues(queryRequest: QueryRequest): Future[Seq[SKeyValue]] = {
    val edge = toRequestEdge(queryRequest, Nil)
    val rpc = buildRequest(queryRequest, edge)

    fetchKeyValues(rpc)
  }

  /**
   * since HBase natively provide CheckAndSet on storage level, implementation becomes simple.
   * @param rpc: key value that is need to be stored on storage.
   * @param expectedOpt: last valid value for rpc's KeyValue.value from fetching.
   * @return return true if expected value matches and our rpc is successfully applied, otherwise false.
   *         note that when some other thread modified same cell and have different value on this KeyValue,
   *         then HBase atomically return false.
   */
  override def writeLock(rpc: SKeyValue, expectedOpt: Option[SKeyValue]): Future[Boolean] = {
    val put = new PutRequest(rpc.table, rpc.row, rpc.cf, rpc.qualifier, rpc.value, rpc.timestamp)
    val expected = expectedOpt.map(_.value).getOrElse(Array.empty)
    client(withWait = true).compareAndSet(put, expected).map(true.booleanValue())(ret => ret.booleanValue()).toFuture(true)
  }


  /**
   * given queryRequest, build storage specific RPC Request.
   * In HBase case, we either build Scanner or GetRequest.
   *
   * IndexEdge layer:
   *    Tall schema(v4): use scanner.
   *    Wide schema(label's schema version in v1, v2, v3): use GetRequest with columnRangeFilter
   *                                                       when query is given with itnerval option.
   * SnapshotEdge layer:
   *    Tall schema(v3, v4): use GetRequest without column filter.
   *    Wide schema(label's schema version in v1, v2): use GetRequest with columnRangeFilter.
   * Vertex layer:
   *    all version: use GetRequest without column filter.
   * @param queryRequest
   * @return Scanner or GetRequest with proper setup with StartKey, EndKey, RangeFilter.
   */
  override def buildRequest(queryRequest: QueryRequest, edge: S2Edge): AsyncRPC = {
    import Serializable._
    val queryParam = queryRequest.queryParam
    val label = queryParam.label

    val serializer = if (queryParam.tgtVertexInnerIdOpt.isDefined) {
      val snapshotEdge = edge.toSnapshotEdge
      snapshotEdgeSerializer(snapshotEdge)
    } else {
      val indexEdge = edge.toIndexEdge(queryParam.labelOrderSeq)
      indexEdgeSerializer(indexEdge)
    }

    val rowKey = serializer.toRowKey
    val (minTs, maxTs) = queryParam.durationOpt.getOrElse((0L, Long.MaxValue))

    val (intervalMaxBytes, intervalMinBytes) = queryParam.buildInterval(Option(edge))

    label.schemaVersion match {
      case HBaseType.VERSION4 if queryParam.tgtVertexInnerIdOpt.isEmpty =>
        val scanner = AsynchbasePatcher.newScanner(client, label.hbaseTableName)
        scanner.setFamily(edgeCf)

        /*
         * TODO: remove this part.
         */
        val indexEdgeOpt = edge.edgesWithIndex.find(edgeWithIndex => edgeWithIndex.labelIndex.seq == queryParam.labelOrderSeq)
        val indexEdge = indexEdgeOpt.getOrElse(throw new RuntimeException(s"Can`t find index for query $queryParam"))

        val srcIdBytes = VertexId.toSourceVertexId(indexEdge.srcVertex.id).bytes
        val labelWithDirBytes = indexEdge.labelWithDir.bytes
        val labelIndexSeqWithIsInvertedBytes = StorageSerializable.labelOrderSeqWithIsInverted(indexEdge.labelIndexSeq, isInverted = false)
        val baseKey = Bytes.add(srcIdBytes, labelWithDirBytes, labelIndexSeqWithIsInvertedBytes)

        val (startKey, stopKey) =
          if (queryParam.intervalOpt.isDefined) {
            // interval is set.
            val _startKey = queryParam.cursorOpt match {
              case Some(cursor) => Base64.getDecoder.decode(cursor)
              case None => Bytes.add(baseKey, intervalMaxBytes)
            }
            (_startKey , Bytes.add(baseKey, intervalMinBytes))
          } else {
             /*
              * note: since propsToBytes encode size of property map at first byte, we are sure about max value here
              */
            val _startKey = queryParam.cursorOpt match {
              case Some(cursor) => Base64.getDecoder.decode(cursor)
              case None => baseKey
            }
            (_startKey, Bytes.add(baseKey, Array.fill(1)(-1)))
          }

        scanner.setStartKey(startKey)
        scanner.setStopKey(stopKey)

        if (queryParam.limit == Int.MinValue) logger.debug(s"MinValue: $queryParam")

        scanner.setMaxVersions(1)
        // TODO: exclusive condition innerOffset with cursorOpt
        if (queryParam.cursorOpt.isDefined) {
          scanner.setMaxNumRows(getV4MaxNumRows(queryParam.limit))
        } else {
          scanner.setMaxNumRows(getV4MaxNumRows(queryParam.innerOffset + queryParam.innerLimit))
        }
        scanner.setMaxTimestamp(maxTs)
        scanner.setMinTimestamp(minTs)
        scanner.setRpcTimeout(queryParam.rpcTimeout)

        // SET option for this rpc properly.
        if (queryParam.cursorOpt.isDefined) Right(ScanWithRange(scanner, 0, queryParam.limit))
        else Right(ScanWithRange(scanner, 0, queryParam.innerOffset + queryParam.innerLimit))

      case _ =>
        val get = if (queryParam.tgtVertexInnerIdOpt.isDefined) {
          new GetRequest(label.hbaseTableName.getBytes, rowKey, edgeCf, serializer.toQualifier)
        } else {
          new GetRequest(label.hbaseTableName.getBytes, rowKey, edgeCf)
        }

        get.maxVersions(1)
        get.setFailfast(true)
        get.setMinTimestamp(minTs)
        get.setMaxTimestamp(maxTs)
        get.setTimeout(queryParam.rpcTimeout)

        val pagination = new ColumnPaginationFilter(queryParam.limit, queryParam.offset)
        val columnRangeFilterOpt = queryParam.intervalOpt.map { interval =>
          new ColumnRangeFilter(intervalMaxBytes, true, intervalMinBytes, true)
        }
        get.setFilter(new FilterList(pagination +: columnRangeFilterOpt.toSeq, MUST_PASS_ALL))
        Left(get)
    }
  }

  /**
   * we are using future cache to squash requests into same key on storage.
   *
   * @param queryRequest
   * @param isInnerCall
   * @param parentEdges
   * @return we use Deferred here since it has much better performrance compared to scala.concurrent.Future.
   *         seems like map, flatMap on scala.concurrent.Future is slower than Deferred's addCallback
   */
  override def fetch(queryRequest: QueryRequest,
                     isInnerCall: Boolean,
                     parentEdges: Seq[EdgeWithScore]): Deferred[StepResult] = {

    def fetchInner(hbaseRpc: AsyncRPC): Deferred[StepResult] = {
      val prevStepScore = queryRequest.prevStepScore
      val fallbackFn: (Exception => StepResult) = { ex =>
        logger.error(s"fetchInner failed. fallback return. $hbaseRpc}", ex)
        StepResult.Failure
      }

      val queryParam = queryRequest.queryParam
      fetchKeyValuesInner(hbaseRpc).mapWithFallback(emptyKeyValues)(fallbackFn) { kvs =>
        val (startOffset, len) = queryParam.label.schemaVersion match {
          case HBaseType.VERSION4 =>
            val offset = if (queryParam.cursorOpt.isDefined) 0 else queryParam.offset
            (offset, queryParam.limit)
          case _ => (0, kvs.length)
        }

        toEdges(kvs, queryRequest, prevStepScore, isInnerCall, parentEdges, startOffset, len)
      }
    }

    val queryParam = queryRequest.queryParam
    val cacheTTL = queryParam.cacheTTLInMillis
    /* with version 4, request's type is (Scanner, (Int, Int)). otherwise GetRequest. */

    val edge = toRequestEdge(queryRequest, parentEdges)
    val request = buildRequest(queryRequest, edge)

    val (intervalMaxBytes, intervalMinBytes) = queryParam.buildInterval(Option(edge))
    val requestCacheKey = Bytes.add(toCacheKeyBytes(request), intervalMaxBytes, intervalMinBytes)

    if (cacheTTL <= 0) fetchInner(request)
    else {
      val cacheKeyBytes = Bytes.add(queryRequest.query.queryOption.cacheKeyBytes, requestCacheKey)

//      val cacheKeyBytes = toCacheKeyBytes(request)
      val cacheKey = queryParam.toCacheKey(cacheKeyBytes)
      futureCache.getOrElseUpdate(cacheKey, cacheTTL)(fetchInner(request))
    }
  }

  override def fetches(queryRequests: Seq[QueryRequest],
                       prevStepEdges: Map[VertexId, Seq[EdgeWithScore]]): Future[Seq[StepResult]] = {
    val defers: Seq[Deferred[StepResult]] = for {
      queryRequest <- queryRequests
    } yield {
        val queryOption = queryRequest.query.queryOption
        val queryParam = queryRequest.queryParam
        val shouldBuildParents = queryOption.returnTree || queryParam.whereHasParent
        val parentEdges = if (shouldBuildParents) prevStepEdges.getOrElse(queryRequest.vertex.id, Nil) else Nil
        fetch(queryRequest, isInnerCall = false, parentEdges)
      }

    val grouped: Deferred[util.ArrayList[StepResult]] = Deferred.groupInOrder(defers)
    grouped.map(emptyStepResult) { queryResults: util.ArrayList[StepResult] =>
      queryResults.toSeq
    }.toFuture(emptyStepResult)
  }


  def fetchVertexKeyValues(request: QueryRequest): Future[Seq[SKeyValue]] = {
    val edge = toRequestEdge(request, Nil)
    fetchKeyValues(buildRequest(request, edge))
  }


  def fetchVertexKeyValues(request: AsyncRPC): Future[Seq[SKeyValue]] = fetchKeyValues(request)

  /**
   * when withWait is given, we use client with flushInterval set to 0.
   * if we are not using this, then we are adding extra wait time as much as flushInterval in worst case.
   *
   * @param edges
   * @param withWait
   * @return
   */
  override def incrementCounts(edges: Seq[S2Edge], withWait: Boolean): Future[Seq[(Boolean, Long, Long)]] = {

    val _client = client(withWait)
    val defers: Seq[Deferred[(Boolean, Long, Long)]] = for {
      edge <- edges
    } yield {
        val futures: List[Deferred[(Boolean, Long, Long)]] = for {
          relEdge <- edge.relatedEdges
          edgeWithIndex <- EdgeMutate.filterIndexOption(relEdge.edgesWithIndexValid)
        } yield {
          val countWithTs = edge.propertyValueInner(LabelMeta.count)
          val countVal = countWithTs.innerVal.toString().toLong
          val kv = buildIncrementsCountAsync(edgeWithIndex, countVal).head
          val request = new AtomicIncrementRequest(kv.table, kv.row, kv.cf, kv.qualifier, Bytes.toLong(kv.value))
          val fallbackFn: (Exception => (Boolean, Long, Long)) = { ex =>
            logger.error(s"mutation failed. $request", ex)
            (false, -1L, -1L)
          }
          val defer = _client.bufferAtomicIncrement(request).mapWithFallback(0L)(fallbackFn) { resultCount: java.lang.Long =>
            (true, resultCount.longValue(), countVal)
          }
          if (withWait) defer
          else Deferred.fromResult((true, -1L, -1L))
        }

        val grouped: Deferred[util.ArrayList[(Boolean, Long, Long)]] = Deferred.group(futures)
        grouped.map(new util.ArrayList[(Boolean, Long, Long)]()) { resultLs => resultLs.head }
      }

    val grouped: Deferred[util.ArrayList[(Boolean, Long, Long)]] = Deferred.groupInOrder(defers)
    grouped.toFuture(new util.ArrayList[(Boolean, Long, Long)]()).map(_.toSeq)
  }


  override def flush(): Unit = clients.foreach { client =>
    super.flush()
    val timeout = Duration((clientFlushInterval + 10) * 20, duration.MILLISECONDS)
    Await.result(client.flush().toFuture(new AnyRef), timeout)
  }


  override def shutdown(): Unit = {
    flush()
    clients.foreach { client =>
      AsynchbaseStorage.shutdown(client)
    }
    if (hbaseExecutor != null) {
      hbaseExecutor.shutdown()
      hbaseExecutor.awaitTermination(1, TimeUnit.MINUTES)
    }
  }

  override def createTable(_zkAddr: String,
                           tableName: String,
                           cfs: List[String],
                           regionMultiplier: Int,
                           ttl: Option[Int],
                           compressionAlgorithm: String,
                           replicationScopeOpt: Option[Int] = None,
                           totalRegionCount: Option[Int] = None): Unit = {
    /* TODO: Decide if we will allow each app server to connect to multiple hbase cluster */
    for {
      zkAddr <- Seq(zkQuorum) ++ zkQuorumSlave.toSeq
    } {
      logger.info(s"create table: $tableName on $zkAddr, $cfs, $regionMultiplier, $compressionAlgorithm")
      withAdmin(zkAddr) { admin =>
        val regionCount = totalRegionCount.getOrElse(admin.getClusterStatus.getServersSize * regionMultiplier)
        try {
          if (!admin.tableExists(TableName.valueOf(tableName))) {
            val desc = new HTableDescriptor(TableName.valueOf(tableName))
            desc.setDurability(Durability.ASYNC_WAL)
            for (cf <- cfs) {
              val columnDesc = new HColumnDescriptor(cf)
                .setCompressionType(Algorithm.valueOf(compressionAlgorithm.toUpperCase))
                .setBloomFilterType(BloomType.ROW)
                .setDataBlockEncoding(DataBlockEncoding.FAST_DIFF)
                .setMaxVersions(1)
                .setTimeToLive(2147483647)
                .setMinVersions(0)
                .setBlocksize(32768)
                .setBlockCacheEnabled(true)
                // FIXME: For test!!
                .setInMemory(true)
              if (ttl.isDefined) columnDesc.setTimeToLive(ttl.get)
              if (replicationScopeOpt.isDefined) columnDesc.setScope(replicationScopeOpt.get)
              desc.addFamily(columnDesc)
            }

            if (regionCount <= 1) admin.createTable(desc)
            else admin.createTable(desc, getStartKey(regionCount), getEndKey(regionCount), regionCount)
          } else {
            logger.info(s"$zkAddr, $tableName, $cfs already exist.")
          }
        } catch {
          case e: Throwable =>
            logger.error(s"$zkAddr, $tableName failed with $e", e)
            throw e
        }
      }
    }
  }

  override def truncateTable(zkAddr: String, tableNameStr: String): Unit = {
    withAdmin(zkAddr) { admin =>
      val tableName = TableName.valueOf(tableNameStr)
      if (!Try(admin.tableExists(tableName)).getOrElse(false)) {
        logger.info(s"No table to truncate ${tableNameStr}")
        return
      }

      Try(admin.isTableDisabled(tableName)).map {
        case true =>
          logger.info(s"${tableNameStr} is already disabled.")

        case false =>
          logger.info(s"Before disabling to trucate ${tableNameStr}")
          Try(admin.disableTable(tableName)).recover {
            case NonFatal(e) =>
              logger.info(s"Failed to disable ${tableNameStr}: ${e}")
          }
          logger.info(s"After disabling to trucate ${tableNameStr}")
      }

      logger.info(s"Before truncating ${tableNameStr}")
      Try(admin.truncateTable(tableName, true)).recover {
        case NonFatal(e) =>
          logger.info(s"Failed to truncate ${tableNameStr}: ${e}")
      }
      logger.info(s"After truncating ${tableNameStr}")
      Try(admin.close()).recover {
        case NonFatal(e) =>
          logger.info(s"Failed to close admin ${tableNameStr}: ${e}")
      }
      Try(admin.getConnection.close()).recover {
        case NonFatal(e) =>
          logger.info(s"Failed to close connection ${tableNameStr}: ${e}")
      }
    }
  }

  override def deleteTable(zkAddr: String, tableNameStr: String): Unit = {
    withAdmin(zkAddr) { admin =>
      val tableName = TableName.valueOf(tableNameStr)
      if (!admin.tableExists(tableName)) {
        return
      }
      if (admin.isTableEnabled(tableName)) {
        admin.disableTable(tableName)
      }
      admin.deleteTable(tableName)
    }
  }

  /** Asynchbase implementation override default getVertices to use future Cache */
  override def getVertices(vertices: Seq[S2Vertex]): Future[Seq[S2Vertex]] = {
    def fromResult(kvs: Seq[SKeyValue],
                   version: String): Option[S2Vertex] = {
      if (kvs.isEmpty) None
      else vertexDeserializer.fromKeyValues(kvs, None)
    }

    val futures = vertices.map { vertex =>
      val kvs = vertexSerializer(vertex).toKeyValues
      val get = new GetRequest(vertex.hbaseTableName.getBytes, kvs.head.row, Serializable.vertexCf)
      //      get.setTimeout(this.singleGetTimeout.toShort)
      get.setFailfast(true)
      get.maxVersions(1)

      fetchVertexKeyValues(Left(get)).map { kvs =>
        fromResult(kvs, vertex.serviceColumn.schemaVersion)
      }
//      val cacheKey = MurmurHash3.stringHash(get.toString)
//      vertexCache.getOrElseUpdate(cacheKey, cacheTTL = -1)(fetchVertexKeyValues(Left(get))).map { kvs =>
//        fromResult(kvs, vertex.serviceColumn.schemaVersion)
//      }
    }

    Future.sequence(futures).map { result => result.toList.flatten }
  }

  //TODO: Limited to 100000 edges per hbase table. fix this later.
  override def fetchEdgesAll(): Future[Seq[S2Edge]] = {
    val futures = Label.findAll().groupBy(_.hbaseTableName).toSeq.map { case (hTableName, labels) =>
      val distinctLabels = labels.toSet
      val scan = AsynchbasePatcher.newScanner(client, hTableName)
      scan.setFamily(Serializable.edgeCf)
      scan.setMaxVersions(1)

      scan.nextRows(S2Graph.FetchAllLimit).toFuture(emptyKeyValuesLs).map {
        case null => Seq.empty
        case kvsLs =>
          kvsLs.flatMap { kvs =>
            kvs.flatMap { kv =>
              val sKV = implicitly[CanSKeyValue[KeyValue]].toSKeyValue(kv)

              indexEdgeDeserializer.fromKeyValues(Seq(kv), None)
                .filter(e => distinctLabels(e.innerLabel) && e.direction == "out" && !e.isDegree)
            }
          }
      }
    }

    Future.sequence(futures).map(_.flatten)
  }

  override def fetchVerticesAll(): Future[Seq[S2Vertex]] = {
    val futures = ServiceColumn.findAll().groupBy(_.service.hTableName).toSeq.map { case (hTableName, columns) =>
      val distinctColumns = columns.toSet
      val scan = AsynchbasePatcher.newScanner(client, hTableName)
      scan.setFamily(Serializable.vertexCf)
      scan.setMaxVersions(1)

      scan.nextRows(S2Graph.FetchAllLimit).toFuture(emptyKeyValuesLs).map {
        case null => Seq.empty
        case kvsLs =>
          kvsLs.flatMap { kvs =>
            vertexDeserializer.fromKeyValues(kvs, None)
              .filter(v => distinctColumns(v.serviceColumn))
          }
      }
    }
    Future.sequence(futures).map(_.flatten)
  }

  class V4ResultHandler(scanner: Scanner, defer: Deferred[util.ArrayList[KeyValue]], offset: Int, limit : Int) extends Callback[Object, util.ArrayList[util.ArrayList[KeyValue]]] {
    val results = new util.ArrayList[KeyValue]()
    var offsetCount = 0

    override def call(kvsLs: util.ArrayList[util.ArrayList[KeyValue]]): Object = {
      try {
        if (kvsLs == null) {
          defer.callback(results)
          Try(scanner.close())
        } else {
          val curRet = new util.ArrayList[KeyValue]()
          kvsLs.foreach(curRet.addAll(_))
          val prevOffset = offsetCount
          offsetCount += curRet.size()

          val nextRet = if(offsetCount > offset){
            if(prevOffset < offset ) {
              curRet.subList(offset - prevOffset, curRet.size())
            } else{
              curRet
            }
          } else{
            emptyKeyValues
          }

          val needCount = limit - results.size()
          if (needCount >= nextRet.size()) {
            results.addAll(nextRet)
          } else {
            results.addAll(nextRet.subList(0, needCount))
          }

          if (results.size() < limit) {
            scanner.nextRows().addCallback(this)
          } else {
            defer.callback(results)
            Try(scanner.close())
          }
        }
      } catch{
        case ex: Exception =>
          logger.error(s"fetchKeyValuesInner failed.", ex)
          defer.callback(ex)
          Try(scanner.close())
      }
    }
  }

  /**
   * Private Methods which is specific to Asynchbase implementation.
   */
  private def fetchKeyValuesInner(rpc: AsyncRPC): Deferred[util.ArrayList[KeyValue]] = {
    rpc match {
      case Left(get) => client.get(get)
      case Right(ScanWithRange(scanner, offset, limit)) =>
        val deferred = new Deferred[util.ArrayList[KeyValue]]()
        scanner.nextRows().addCallback(new V4ResultHandler(scanner, deferred, offset, limit))
        deferred
      case _ => Deferred.fromError(new RuntimeException(s"fetchKeyValues failed. $rpc"))
    }
  }

  private def toCacheKeyBytes(hbaseRpc: AsyncRPC): Array[Byte] = {
    /* with version 4, request's type is (Scanner, (Int, Int)). otherwise GetRequest. */
    hbaseRpc match {
      case Left(getRequest) => getRequest.key
      case Right(ScanWithRange(scanner, offset, limit)) =>
        Bytes.add(scanner.getCurrentKey, Bytes.add(Bytes.toBytes(offset), Bytes.toBytes(limit)))
      case _ =>
        logger.error(s"toCacheKeyBytes failed. not supported class type. $hbaseRpc")
        throw new RuntimeException(s"toCacheKeyBytes: $hbaseRpc")
    }
  }

  private def getSecureClusterAdmin(zkAddr: String) = {
    val jaas = config.getString("java.security.auth.login.config")
    val krb5Conf = config.getString("java.security.krb5.conf")
    val realm = config.getString("realm")
    val principal = config.getString("principal")
    val keytab = config.getString("keytab")

    System.setProperty("java.security.auth.login.config", jaas)
    System.setProperty("java.security.krb5.conf", krb5Conf)
    // System.setProperty("sun.security.krb5.debug", "true")
    // System.setProperty("sun.security.spnego.debug", "true")
    val conf = new Configuration(true)
    val hConf = HBaseConfiguration.create(conf)

    hConf.set("hbase.zookeeper.quorum", zkAddr)

    hConf.set("hadoop.security.authentication", "Kerberos")
    hConf.set("hbase.security.authentication", "Kerberos")
    hConf.set("hbase.master.kerberos.principal", "hbase/_HOST@" + realm)
    hConf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@" + realm)

    System.out.println("Connecting secure cluster, using keytab\n")
    UserGroupInformation.setConfiguration(hConf)
    UserGroupInformation.loginUserFromKeytab(principal, keytab)
    val currentUser = UserGroupInformation.getCurrentUser()
    System.out.println("current user : " + currentUser + "\n")

    // get table list
    val conn = ConnectionFactory.createConnection(hConf)
    conn.getAdmin
  }

  private def withAdmin(zkAddr: String)(op: Admin => Unit): Unit = {
    val admin = getAdmin(zkAddr)
    try {
      op(admin)
    } finally {
      admin.close()
      admin.getConnection.close()
    }
  }
  /**
   * following configuration need to come together to use secured hbase cluster.
   * 1. set hbase.security.auth.enable = true
   * 2. set file path to jaas file java.security.auth.login.config
   * 3. set file path to kerberos file java.security.krb5.conf
   * 4. set realm
   * 5. set principal
   * 6. set file path to keytab
   * @param zkAddr
   * @return
   */
  private def getAdmin(zkAddr: String) = {
    if (config.hasPath("hbase.security.auth.enable") && config.getBoolean("hbase.security.auth.enable")) {
      getSecureClusterAdmin(zkAddr)
    } else {
      val conf = HBaseConfiguration.create()
      conf.set("hbase.zookeeper.quorum", zkAddr)
      val conn = ConnectionFactory.createConnection(conf)
      conn.getAdmin
    }
  }

  private def enableTable(zkAddr: String, tableName: String) = {
    withAdmin(zkAddr) { admin =>
      admin.enableTable(TableName.valueOf(tableName))
    }
  }

  private def disableTable(zkAddr: String, tableName: String) = {
    withAdmin(zkAddr) { admin =>
      admin.disableTable(TableName.valueOf(tableName))
    }
  }

  private def dropTable(zkAddr: String, tableName: String) = {
    withAdmin(zkAddr) { admin =>
      admin.disableTable(TableName.valueOf(tableName))
      admin.deleteTable(TableName.valueOf(tableName))
    }
  }

  private def getStartKey(regionCount: Int): Array[Byte] = {
    Bytes.toBytes((Int.MaxValue / regionCount))
  }

  private def getEndKey(regionCount: Int): Array[Byte] = {
    Bytes.toBytes((Int.MaxValue / regionCount * (regionCount - 1)))
  }
}