<|start_filename|>hbase-native-client/src/hbase/examples/load-client.cc<|end_filename|>
/*
 * 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.
 *
 */

#include <folly/Logging.h>
#include <folly/Random.h>
#include <gflags/gflags.h>

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>

#include "hbase/client/client.h"
#include "hbase/client/get.h"
#include "hbase/client/put.h"
#include "hbase/client/table.h"
#include "hbase/serde/table-name.h"
#include "hbase/utils/time-util.h"

using hbase::Client;
using hbase::Configuration;
using hbase::Get;
using hbase::Put;
using hbase::Table;
using hbase::pb::TableName;
using hbase::TimeUtil;
using folly::Random;

DEFINE_string(table, "load_test_table", "What table to do the reads and writes with");
DEFINE_string(families, "f", "comma separated list of column family names");
DEFINE_string(conf, "", "Conf directory to read the config from (optional)");
DEFINE_string(zookeeper, "localhost:2181", "What zk quorum to talk to");
DEFINE_string(znode, "/hbase", "parent znode");
DEFINE_uint64(num_rows, 1'000'000, "How many rows to write and read");
DEFINE_uint64(num_cols, 1000, "How many columns there are in a row");
DEFINE_int32(threads, 10, "How many client threads");
DEFINE_int32(batch_num_rows, 100, "number of rows in one multi-get / multi-put");
DEFINE_uint64(report_num_rows, 5000, "How frequent we should report the progress");
DEFINE_bool(gets, true, "perform gets");
DEFINE_bool(scans, true, "perform scans");
DEFINE_bool(puts, true, "perform put's");
DEFINE_bool(appends, true, "perform append's");

static constexpr const char *kNumColumn = "num";
static constexpr const char *incrPrefix = "i";
static constexpr const char *appendPrefix = "a";

std::string PrefixZero(int total_width, int num) {
  std::string str = std::to_string(num);
  int prefix_len = total_width - str.length();
  if (prefix_len > 0) {
    return std::string(prefix_len, '0') + str;
  }
  return str;
}

bool Verify(std::shared_ptr<hbase::Result> result, std::string family, int m) {
  auto col = std::to_string(m);
  if (!result->Value(family, col)) {
    LOG(ERROR) << "Column:" << col << " is not found for " << result->Row();
    return false;
  }
  auto l = *(result->Value(family, col));
  if (l != col) {
    LOG(ERROR) << "value " << *(result->Value(family, "1")) << " is not " << col;
    return false;
  }
  if (FLAGS_appends) {
    if (!result->Value(family, incrPrefix + col)) {
      LOG(ERROR) << "Column:" << (incrPrefix + col) << " is not found for " << result->Row();
      return false;
    }
    auto int_val = hbase::BytesUtil::ToInt64(*(result->Value(family, incrPrefix + col)));
    if (int_val != m) {
      LOG(ERROR) << "value is not " << col << " for " << result->Row();
      return false;
    }
    if (!result->Value(family, appendPrefix + col)) {
      LOG(ERROR) << "Column:" << (appendPrefix + col) << " is not found for " << result->Row();
      return false;
    }
    l = *(result->Value(family, appendPrefix + col));
    if (l != col) {
      LOG(ERROR) << "value " << *(result->Value(family, "1")) << " is not " << col;
      return false;
    }
  }

  return true;
}

bool Verify(std::shared_ptr<hbase::Result> result, const std::string &row,
            const std::vector<std::string> &families) {
  if (result == nullptr || result->IsEmpty()) {
    LOG(ERROR) << "didn't get result";
    return false;
  }
  if (result->Row().compare(row) != 0) {
    LOG(ERROR) << "row " << result->Row() << " is not the expected: " << row;
    return false;
  }
  // Test the values
  for (auto family : families) {
    if (!result->Value(family, kNumColumn)) {
      LOG(ERROR) << "Column:" << kNumColumn << " is not found for " << result->Row();
      return false;
    }
    auto cols = std::stoi(*(result->Value(family, kNumColumn)));
    VLOG(3) << "Result for row:" << row << " contains " << std::to_string(cols) << " columns";
    for (int m = 1; m <= cols; m++) {
      if (!Verify(result, family, m)) return false;
    }
  }
  return true;
}

bool DoScan(int iteration, uint64_t max_row, uint64_t rows, std::unique_ptr<Table> table,
            const std::vector<std::string> &families) {
  hbase::Scan scan{};
  auto start = iteration * rows;
  auto end = start + rows;
  auto width = std::to_string(max_row).length();
  scan.SetStartRow(PrefixZero(width, start));
  if (end != max_row && end != max_row + 1) {
    scan.SetStopRow(PrefixZero(width, end));
  }

  auto start_ns = TimeUtil::GetNowNanos();
  auto scanner = table->Scan(scan);

  auto cnt = 0;
  auto r = scanner->Next();
  while (r != nullptr) {
    auto row = PrefixZero(width, start + cnt);
    if (!Verify(r, row, families)) {
      return false;
    }
    cnt++;
    r = scanner->Next();
    if (cnt != 0 && cnt % FLAGS_report_num_rows == 0) {
      LOG(INFO) << "(Thread " << iteration << ") "
                << "Scan iterated over " << cnt << " results in "
                << TimeUtil::ElapsedMillis(start_ns) << " ms.";
    }
  }
  if (cnt != rows) {
    LOG(ERROR) << "(Thread " << iteration << ") "
               << "Expected number of results does not match. expected:" << rows
               << ", actual:" << cnt;
    return false;
  }
  LOG(INFO) << "(Thread " << iteration << ") "
            << "scanned " << std::to_string(cnt) << " rows in " << TimeUtil::ElapsedMillis(start_ns)
            << " ms.";
  return true;
}

bool DoGet(int iteration, uint64_t max_row, uint64_t rows, std::unique_ptr<Table> table,
           const std::vector<std::string> &families, uint64_t batch_num_rows) {
  auto width = std::to_string(max_row).length();
  auto start_ns = TimeUtil::GetNowNanos();
  for (uint64_t k = iteration; k <= max_row;) {
    uint64_t total_read = 0;
    std::vector<hbase::Get> gets;
    for (uint64_t i = 0; i < batch_num_rows && k <= max_row; ++i, k += FLAGS_threads) {
      std::string row = PrefixZero(width, k);
      hbase::Get get(row);
      gets.push_back(get);
    }
    VLOG(3) << "getting for " << batch_num_rows << " rows";
    auto results = table->Get(gets);
    if (results.size() != gets.size()) {
      LOG(ERROR) << "(Thread " << iteration << ") "
                 << "Expected number of results does not match. expected:" << gets.size()
                 << ", actual:" << results.size();
      return false;
    }
    for (uint64_t i = 0; i < batch_num_rows && i < results.size(); ++i) {
      if (!Verify(results[i], gets[i].row(), families)) {
        return false;
      }
    }
    total_read += gets.size();
    if (total_read != 0 && total_read % FLAGS_report_num_rows == 0) {
      LOG(INFO) << "(Thread " << iteration << ") "
                << "Sent  " << total_read << " Multi-Get requests in "
                << TimeUtil::ElapsedMillis(start_ns) << " ms.";
    }
    k += batch_num_rows;
  }
  LOG(INFO) << "(Thread " << iteration << ") "
            << "Sent " << rows << " gets"
            << " in " << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  return true;
}

void DoPut(int iteration, uint64_t max_row, uint64_t rows, int cols, std::unique_ptr<Table> table,
           const std::vector<std::string> &families) {
  auto start_ns = TimeUtil::GetNowNanos();
  auto width = std::to_string(max_row).length();
  for (uint64_t j = 0; j < rows; j++) {
    std::string row = PrefixZero(width, iteration * rows + j);
    auto put = Put{row};
    for (auto family : families) {
      auto n_cols = Random::rand32(1, cols);
      put.AddColumn(family, kNumColumn, std::to_string(n_cols));
      for (unsigned int k = 1; k <= n_cols; k++) {
        put.AddColumn(family, std::to_string(k), std::to_string(k));
      }
    }
    table->Put(put);
    if ((j + 1) % FLAGS_report_num_rows == 0) {
      LOG(INFO) << "(Thread " << iteration << ") "
                << "Written " << std::to_string(j + 1) << " rows in "
                << TimeUtil::ElapsedMillis(start_ns) << " ms.";
    }
  }
  LOG(INFO) << "(Thread " << iteration << ") "
            << "written " << std::to_string(rows) << " rows"
            << " in " << TimeUtil::ElapsedMillis(start_ns) << " ms.";
}

bool DoAppendIncrement(int iteration, uint64_t max_row, uint64_t rows, int cols,
                       std::unique_ptr<Table> table, const std::vector<std::string> &families) {
  auto start_ns = TimeUtil::GetNowNanos();
  auto width = std::to_string(max_row).length();
  for (uint64_t j = 0; j < rows; j++) {
    std::string row = PrefixZero(width, iteration * rows + j);
    hbase::Get get(row);
    auto result = table->Get(get);
    for (auto family : families) {
      auto n_cols = std::stoi(*(result->Value(family, kNumColumn)));
      for (unsigned int k = 1; k <= n_cols; k++) {
        table->Increment(
            hbase::Increment{row}.AddColumn(family, incrPrefix + std::to_string(k), k));
        if (!table->Append(hbase::Append{row}.Add(family, appendPrefix + std::to_string(k),
                                                  std::to_string(k)))) {
          LOG(ERROR) << "(Thread " << iteration << ") "
                     << "append for " << row << " family: " << family << " failed";
          return false;
        }
      }
    }
    if ((j + 1) % FLAGS_report_num_rows == 0)
      LOG(INFO) << "(Thread " << iteration << ") "
                << "Written " << std::to_string(j + 1) << " increments"
                << " in " << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }
  LOG(INFO) << "(Thread " << iteration << ") "
            << "written " << std::to_string(rows) << " increments"
            << " in " << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  return true;
}

int main(int argc, char *argv[]) {
  gflags::SetUsageMessage("Load client to manipulate multiple rows from HBase on the comamnd line");
  gflags::ParseCommandLineFlags(&argc, &argv, true);
  google::InitGoogleLogging(argv[0]);
  google::InstallFailureSignalHandler();
  FLAGS_logtostderr = 1;
  FLAGS_stderrthreshold = 1;

  if (FLAGS_batch_num_rows < 1) {
    LOG(ERROR) << "size of multi get should be positive";
    return -1;
  }
  if (!FLAGS_gets && !FLAGS_scans && !FLAGS_puts) {
    LOG(ERROR) << "Must perform at least Get or Put operations";
    return -1;
  }
  std::shared_ptr<Configuration> conf = nullptr;
  if (FLAGS_conf == "") {
    // Configuration
    conf = std::make_shared<Configuration>();
    conf->Set("hbase.zookeeper.quorum", FLAGS_zookeeper);
    conf->Set("zookeeper.znode.parent", FLAGS_znode);
  } else {
    setenv("HBASE_CONF", FLAGS_conf.c_str(), 1);
    hbase::HBaseConfigurationLoader loader;
    conf = std::make_shared<Configuration>(loader.LoadDefaultResources().value());
  }
  auto tn = std::make_shared<TableName>(folly::to<TableName>(FLAGS_table));
  auto num_puts = FLAGS_num_rows;

  auto client = std::make_unique<Client>(*conf);

  // Do the Put requests

  std::vector<std::string> families;
  std::size_t pos = 0, found;
  while ((found = FLAGS_families.find_first_of(',', pos)) != std::string::npos) {
    families.push_back(FLAGS_families.substr(pos, found - pos));
    pos = found + 1;
  }
  families.push_back(FLAGS_families.substr(pos));

  int rows = FLAGS_num_rows / FLAGS_threads;
  if (FLAGS_num_rows % FLAGS_threads != 0) rows++;
  int cols = FLAGS_num_cols;
  std::atomic<int8_t> succeeded{1};  // not using bool since we want atomic &=
  if (FLAGS_puts) {
    LOG(INFO) << "Sending put requests";
    auto start_ns = TimeUtil::GetNowNanos();
    std::vector<std::thread> writer_threads;
    for (int i = 0; i < FLAGS_threads; i++) {
      writer_threads.push_back(std::thread([&, i] {
        auto table = client->Table(*tn);
        DoPut(i, FLAGS_num_rows - 1, rows, cols, std::move(table), families);
      }));
    }
    for (auto &t : writer_threads) {
      t.join();
    }
    LOG(INFO) << "Successfully sent  " << num_puts << " Put requests in "
              << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }
  if (FLAGS_appends) {
    LOG(INFO) << "Sending append/increment requests";
    auto start_ns = TimeUtil::GetNowNanos();
    std::vector<std::thread> writer_threads;
    for (int i = 0; i < FLAGS_threads; i++) {
      writer_threads.push_back(std::thread([&, i] {
        auto table = client->Table(*tn);
        succeeded &=
            DoAppendIncrement(i, FLAGS_num_rows - 1, rows, cols, std::move(table), families);
      }));
    }
    for (auto &t : writer_threads) {
      t.join();
    }
    LOG(INFO) << "Successfully sent  " << num_puts << " append requests in "
              << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }

  if (FLAGS_scans) {
    LOG(INFO) << "Sending scan requests";
    auto start_ns = TimeUtil::GetNowNanos();
    std::vector<std::thread> reader_threads;
    for (int i = 0; i < FLAGS_threads; i++) {
      reader_threads.push_back(std::thread([&, i] {
        auto table1 = client->Table(*tn);
        succeeded &= DoScan(i, FLAGS_num_rows - 1, rows, std::move(table1), families);
      }));
    }
    for (auto &t : reader_threads) {
      t.join();
    }

    LOG(INFO) << (succeeded.load() ? "Successfully " : "Failed. ") << " scannned " << num_puts
              << " rows in " << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }

  if (FLAGS_gets) {
    LOG(INFO) << "Sending get requests";
    auto start_ns = TimeUtil::GetNowNanos();
    std::vector<std::thread> reader_threads;
    for (int i = 0; i < FLAGS_threads; i++) {
      reader_threads.push_back(std::thread([&, i] {
        auto table1 = client->Table(*tn);
        succeeded &=
            DoGet(i, FLAGS_num_rows - 1, rows, std::move(table1), families, FLAGS_batch_num_rows);
      }));
    }
    for (auto &t : reader_threads) {
      t.join();
    }

    LOG(INFO) << (succeeded.load() ? "Successful. " : "Failed. ") << " sent multi-get requests for "
              << num_puts << " rows in " << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }
  client->Close();

  return succeeded.load() ? 0 : -1;
}


<|start_filename|>hbase-native-client/include/hbase/client/region-request.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <memory>
#include <queue>
#include <vector>
#include "hbase/client/action.h"
#include "hbase/client/region-location.h"

namespace hbase {

class RegionRequest {
 public:
  // Concurrent
  using ActionList = std::vector<std::shared_ptr<Action>>;
  explicit RegionRequest(const std::shared_ptr<hbase::RegionLocation> &region_loc)
      : region_loc_(region_loc) {}
  ~RegionRequest() {}
  void AddAction(std::shared_ptr<Action> action) { actions_.push_back(action); }
  std::shared_ptr<hbase::RegionLocation> region_location() const { return region_loc_; }
  const ActionList &actions() const { return actions_; }

 private:
  std::shared_ptr<hbase::RegionLocation> region_loc_;
  ActionList actions_;
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/utils/bytes-util.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <memory>
#include <string>

namespace hbase {

class BytesUtil {
 private:
  static const constexpr char kHexChars[] = "0123456789ABCDEF";

 public:
  static std::string ToStringBinary(const std::string& b) { return ToStringBinary(b, 0, b.size()); }
  /**
    * Write a printable representation of a byte array. Non-printable
    * characters are hex escaped in the format \\x%02X, eg:
    * \x00 \x05 etc
    *
    * @param b array to write out
    * @param off offset to start at
    * @param len length to write
    * @return string output
    */
  static std::string ToStringBinary(const std::string& b, size_t off, size_t len);

  static std::string ToString(int64_t value);

  static int64_t ToInt64(std::string str);

  static bool IsEmptyStartRow(const std::string& row) { return row == ""; }

  static bool IsEmptyStopRow(const std::string& row) { return row == ""; }

  static int32_t CompareTo(const std::string& a, const std::string& b) {
    if (a < b) {
      return -1;
    }
    if (a == b) {
      return 0;
    }
    return 1;
  }

  /**
   * Create the closest row after the specified row
   */
  static std::string CreateClosestRowAfter(std::string row) { return row.append(1, '\0'); }
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/connection/rpc-test-server.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once
#include <folly/SocketAddress.h>
#include <wangle/concurrent/CPUThreadPoolExecutor.h>
#include <wangle/service/ExecutorFilter.h>
#include <wangle/service/Service.h>

#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/exceptions/exception.h"

using namespace hbase;
using namespace folly;
using namespace wangle;

namespace hbase {
using RpcTestServerSerializePipeline = wangle::Pipeline<IOBufQueue&, std::unique_ptr<Response>>;

class RpcTestException : public IOException {
 public:
  RpcTestException() {}
  RpcTestException(const std::string& what) : IOException(what) {}
  RpcTestException(const std::string& what, const folly::exception_wrapper& cause)
      : IOException(what, cause) {}
  RpcTestException(const folly::exception_wrapper& cause) : IOException("", cause) {}
};

class RpcTestService : public Service<std::unique_ptr<Request>, std::unique_ptr<Response>> {
 public:
  RpcTestService(std::shared_ptr<folly::SocketAddress> socket_address)
      : socket_address_(socket_address) {}
  virtual ~RpcTestService() = default;
  Future<std::unique_ptr<Response>> operator()(std::unique_ptr<Request> request) override;

 private:
  std::shared_ptr<folly::SocketAddress> socket_address_;
};

class RpcTestServerPipelineFactory : public PipelineFactory<RpcTestServerSerializePipeline> {
 public:
  RpcTestServerSerializePipeline::Ptr newPipeline(
      std::shared_ptr<AsyncTransportWrapper> sock) override;

 private:
  void initService(std::shared_ptr<AsyncTransportWrapper> sock);

 private:
  std::shared_ptr<ExecutorFilter<std::unique_ptr<Request>, std::unique_ptr<Response>>> service_{
      nullptr};
};
}  // end of namespace hbase


<|start_filename|>hbase-native-client/src/hbase/serde/zk.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/serde/zk.h"

#include <folly/io/Cursor.h>
#include <folly/io/IOBuf.h>
#include <google/protobuf/message.h>

#include <string>

using std::runtime_error;

namespace hbase {

static const std::string MAGIC_STRING = "PBUF";

bool ZkDeserializer::Parse(folly::IOBuf *buf, google::protobuf::Message *out) {
  // The format is like this
  // 1 byte of magic number. 255
  // 4 bytes of id length.
  // id_length number of bytes for the id of who put up the znode
  // 4 bytes of a magic string PBUF
  // Then the protobuf serialized without a varint header.

  folly::io::Cursor c{buf};

  // There should be a magic number for recoverable zk
  uint8_t magic_num = c.read<uint8_t>();
  if (magic_num != 255) {
    LOG(ERROR) << "Magic number not in ZK znode data expected 255 got =" << unsigned(magic_num);
    throw runtime_error("Magic number not in znode data");
  }
  // How long is the id?
  uint32_t id_len = c.readBE<uint32_t>();

  if (id_len >= c.length()) {
    LOG(ERROR) << "After skiping the if from zookeeper data there's not enough "
                  "left to read anything else";
    throw runtime_error("Not enough bytes to decode from zookeeper");
  }

  // Skip the id
  c.skip(id_len);

  // Make sure that the magic string is there.
  if (MAGIC_STRING != c.readFixedString(4)) {
    LOG(ERROR) << "There was no PBUF magic string.";
    throw runtime_error("No PBUF magic string in the zookpeeper data.");
  }

  // Try to decode the protobuf.
  // If there's an error bail out.
  if (out->ParseFromArray(c.data(), c.length()) == false) {
    LOG(ERROR) << "Error parsing Protobuf Message";
    throw runtime_error("Error parsing protobuf");
  }

  return true;
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/increment.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <cstdint>
#include <map>
#include <memory>
#include <string>
#include <vector>
#include "hbase/client/cell.h"
#include "hbase/client/mutation.h"

namespace hbase {

class Increment : public Mutation {
 public:
  /**
   * Constructors
   */
  explicit Increment(const std::string& row) : Mutation(row) {}
  Increment(const Increment& cincrement) : Mutation(cincrement) {}
  Increment& operator=(const Increment& cincrement) {
    Mutation::operator=(cincrement);
    return *this;
  }

  ~Increment() = default;

  /**
   *  @brief Increment the column from the specific family with the specified qualifier
   * by the specified amount.
   *  @param family family name
   *  @param qualifier column qualifier
   *  @param amount amount to increment by
   */
  Increment& AddColumn(const std::string& family, const std::string& qualifier, int64_t amount);
  Increment& Add(std::unique_ptr<Cell> cell);
};

}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/sasl-handler.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <glog/logging.h>
#include <sasl/sasl.h>
#include <sasl/saslplug.h>
#include <sasl/saslutil.h>

#include <memory>
#include <string>
#include <vector>

#include "hbase/connection/sasl-util.h"
#include "hbase/connection/service.h"
#include "hbase/security/user.h"
#include "hbase/serde/rpc-serde.h"

namespace hbase {

/**
 * Class to perform SASL handshake with server (currently works with regionserver principals only)
 * It is inserted between EventBaseHandler and LengthFieldBasedFrameDecoder in the pipeline
 * SaslHandler would intercept writes to server by buffering the IOBuf's and start the handshake
 * process
 *   (via sasl_client_XX calls provided by Cyrus)
 * After handshake is complete, SaslHandler would send the buffered IOBuf's to server and
 *   act as pass-thru from then on
 */
class SaslHandler
    : public wangle::HandlerAdapter<folly::IOBufQueue&, std::unique_ptr<folly::IOBuf>> {
 public:
  explicit SaslHandler(std::string user_name, std::shared_ptr<Configuration> conf);
  SaslHandler(const SaslHandler& hdlr);
  ~SaslHandler();

  // from HandlerAdapter
  void read(Context* ctx, folly::IOBufQueue& buf) override;
  folly::Future<folly::Unit> write(Context* ctx, std::unique_ptr<folly::IOBuf> buf) override;
  void transportActive(Context* ctx) override;

 private:
  // used by Cyrus
  sasl_conn_t* sconn_ = nullptr;
  std::string user_name_;
  std::string service_name_;
  std::string host_name_;
  bool secure_;
  std::atomic_flag sasl_connection_setup_started_;
  std::atomic<bool> sasl_connection_setup_in_progress_{true};
  // vector of folly::IOBuf which buffers client writes before handshake is complete
  std::vector<std::unique_ptr<folly::IOBuf>> iobuf_;
  SaslUtil sasl_util_;

  // writes the output returned by sasl_client_XX to server
  folly::Future<folly::Unit> WriteSaslOutput(Context* ctx, const char* out, unsigned int outlen);
  folly::Future<folly::Unit> SaslInit(Context* ctx);
  void FinishAuth(Context* ctx, folly::IOBufQueue* bufQueue);
  void ContinueSaslNegotiation(Context* ctx, folly::IOBufQueue* buf);
  std::string ParseServiceName(std::shared_ptr<Configuration> conf, bool secure);
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/exceptions/exception.cc<|end_filename|>
/*
 * 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.
 *
 */
#include "hbase/exceptions/exception.h"

namespace hbase {
const std::vector<const char*> ExceptionUtil::kAllDoNotRetryIOExceptions = {
    kDoNotRetryIOException,
    kTableNotFoundException,
    kTableNotEnabledException,
    kCoprocessorException,
    kBypassCoprocessorException,
    kInvalidFamilyOperationException,
    kServerTooBusyException,
    kFailedSanityCheckException,
    kCorruptHFileException,
    kLabelAlreadyExistsException,
    kFatalConnectionException,
    kUnsupportedCryptoException,
    kUnsupportedCellCodecException,
    kEmptyServiceNameException,
    kUnknownServiceException,
    kWrongVersionException,
    kBadAuthException,
    kUnsupportedCompressionCodecException,
    kDoNotRetryRegionException,
    kRowTooBigException,
    kRowTooBigExceptionDeprecated,
    kUnknownRegionException,
    kMergeRegionException,
    kNoServerForRegionException,
    kQuotaExceededException,
    kSpaceLimitingException,
    kThrottlingException,
    kAccessDeniedException,
    kUnknownProtocolException,
    kRequestTooBigException,
    kNotAllMetaRegionsOnlineException,
    kConstraintException,
    kNoSuchColumnFamilyException,
    kLeaseException,
    kInvalidLabelException,
    kUnknownScannerException,
    kScannerResetException,
    kOutOfOrderScannerNextException};

bool ExceptionUtil::ShouldRetry(const folly::exception_wrapper& error) {
  bool do_not_retry = false;
  error.with_exception(
      [&](const IOException& ioe) { do_not_retry = do_not_retry || ioe.do_not_retry(); });
  error.with_exception([&](const RemoteException& remote_ex) {
    do_not_retry = do_not_retry || IsJavaDoNotRetryException(remote_ex.exception_class_name());
  });
  return !do_not_retry;
}

/**
 * Returns whether the java exception class extends DoNotRetryException.
 * In the java side, we just have a hierarchy of Exception classes that we use
 * both client side and server side. On the client side, we rethrow the server
 * side exception by un-wrapping the exception from a RemoteException or a ServiceException
 * (see ConnectionUtils.translateException() in Java).
 * Since this object-hierarchy info is not available in C++ side, we are doing a
 * very fragile catch-all list of all exception types in Java that extend the
 * DoNotRetryException class type.
 */
bool ExceptionUtil::IsJavaDoNotRetryException(const std::string& java_class_name) {
  for (auto exception : kAllDoNotRetryIOExceptions) {
    if (java_class_name == exception) {
      return true;
    }
  }
  return false;
}

/**
 * Returns whether the scanner is closed when the client received the
 * remote exception.
 * Since the object-hierarchy info is not available in C++ side, we are doing a
 * very fragile catch-all list of all exception types in Java that extend these
 * three base classes: UnknownScannerException, NotServingRegionException,
 * RegionServerStoppedException
 */
bool ExceptionUtil::IsScannerClosed(const folly::exception_wrapper& exception) {
  bool scanner_closed = false;
  exception.with_exception([&](const RemoteException& remote_ex) {
    auto java_class = remote_ex.exception_class_name();
    if (java_class == kUnknownScannerException || java_class == kNotServingRegionException ||
        java_class == kRegionInRecoveryException || java_class == kRegionOpeningException ||
        java_class == kRegionMovedException || java_class == kRegionServerStoppedException ||
        java_class == kRegionServerAbortedException) {
      scanner_closed = true;
    }
  });
  return scanner_closed;
}

/**
 * Returns whether the wrapped exception is a java exception of type OutOfOrderScannerNextException
 * or ScannerResetException. These two exception types are thrown from the server side when the
 * scanner on the server side is closed.
 */
bool ExceptionUtil::IsScannerOutOfOrder(const folly::exception_wrapper& exception) {
  bool scanner_out_of_order = false;
  exception.with_exception([&](const RemoteException& remote_ex) {
    auto java_class = remote_ex.exception_class_name();
    if (java_class == kOutOfOrderScannerNextException || java_class == kScannerResetException) {
      scanner_out_of_order = true;
    }
  });
  return scanner_out_of_order;
}
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/async-batch-rpc-retrying-caller.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/async-batch-rpc-retrying-caller.h"
#include <glog/logging.h>
#include <limits>

using folly::Future;
using folly::Promise;
using folly::Try;
using hbase::pb::ServerName;
using hbase::pb::TableName;
using hbase::security::User;
using std::chrono::nanoseconds;
using std::chrono::milliseconds;

namespace hbase {

template <typename REQ, typename RESP>
AsyncBatchRpcRetryingCaller<REQ, RESP>::AsyncBatchRpcRetryingCaller(
    std::shared_ptr<AsyncConnection> conn, std::shared_ptr<folly::HHWheelTimer> retry_timer,
    std::shared_ptr<TableName> table_name, const std::vector<REQ> &actions, nanoseconds pause_ns,
    int32_t max_attempts, nanoseconds operation_timeout_ns, nanoseconds rpc_timeout_ns,
    int32_t start_log_errors_count)
    : conn_(conn),
      retry_timer_(retry_timer),
      table_name_(table_name),
      pause_ns_(pause_ns),
      operation_timeout_ns_(operation_timeout_ns),
      rpc_timeout_ns_(rpc_timeout_ns),
      start_log_errors_count_(start_log_errors_count) {
  CHECK(conn_ != nullptr);
  CHECK(retry_timer_ != nullptr);
  location_cache_ = conn_->region_locator();
  rpc_client_ = conn_->rpc_client();
  cpu_pool_ = conn_->cpu_executor();
  CHECK(location_cache_ != nullptr);
  CHECK(rpc_client_ != nullptr);
  CHECK(cpu_pool_ != nullptr);

  max_attempts_ = ConnectionUtils::Retries2Attempts(max_attempts);
  uint32_t index = 0;
  for (auto row : actions) {
    actions_.push_back(std::make_shared<Action>(row, index));
    Promise<RESP> prom{};
    action2promises_.insert(std::pair<uint64_t, Promise<RESP>>(index, std::move(prom)));
    action2futures_.push_back(action2promises_[index++].getFuture());
  }
}

template <typename REQ, typename RESP>
AsyncBatchRpcRetryingCaller<REQ, RESP>::~AsyncBatchRpcRetryingCaller() {}

template <typename REQ, typename RESP>
Future<std::vector<Try<RESP>>> AsyncBatchRpcRetryingCaller<REQ, RESP>::Call() {
  GroupAndSend(actions_, 1);
  return collectAll(action2futures_);
}

template <typename REQ, typename RESP>
int64_t AsyncBatchRpcRetryingCaller<REQ, RESP>::RemainingTimeNs() {
  return operation_timeout_ns_.count() - (TimeUtil::GetNowNanos() - start_ns_);
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::LogException(
    int32_t tries, std::shared_ptr<RegionRequest> region_request,
    const folly::exception_wrapper &ew, std::shared_ptr<ServerName> server_name) {
  if (tries > start_log_errors_count_) {
    std::string regions;
    regions += region_request->region_location()->region_name() + ", ";
    LOG(WARNING) << "Process batch for " << regions << " in " << table_name_->namespace_() << ":"
                 << table_name_->qualifier() << " from " << server_name->host_name()
                 << " failed, tries=" << tries << ":- " << ew.what().toStdString();
  }
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::LogException(
    int32_t tries, const std::vector<std::shared_ptr<RegionRequest>> &region_requests,
    const folly::exception_wrapper &ew, std::shared_ptr<ServerName> server_name) {
  if (tries > start_log_errors_count_) {
    std::string regions;
    for (const auto region_request : region_requests) {
      regions += region_request->region_location()->region_name() + ", ";
    }
    LOG(WARNING) << "Process batch for " << regions << " in " << table_name_->namespace_() << ":"
                 << table_name_->qualifier() << " from " << server_name->host_name()
                 << " failed, tries=" << tries << ew.what().toStdString();
  }
}

template <typename REQ, typename RESP>
const std::string AsyncBatchRpcRetryingCaller<REQ, RESP>::GetExtraContextForError(
    std::shared_ptr<ServerName> server_name) {
  return server_name ? server_name->ShortDebugString() : "";
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::AddError(const std::shared_ptr<Action> &action,
                                                      const folly::exception_wrapper &ew,
                                                      std::shared_ptr<ServerName> server_name) {
  ThrowableWithExtraContext twec(ew, TimeUtil::GetNowNanos(), GetExtraContextForError(server_name));
  AddAction2Error(action->original_index(), twec);
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::AddError(
    const std::vector<std::shared_ptr<Action>> &actions, const folly::exception_wrapper &ew,
    std::shared_ptr<ServerName> server_name) {
  for (const auto action : actions) {
    AddError(action, ew, server_name);
  }
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::FailOne(const std::shared_ptr<Action> &action,
                                                     int32_t tries,
                                                     const folly::exception_wrapper &ew,
                                                     int64_t current_time,
                                                     const std::string extras) {
  auto action_index = action->original_index();
  auto itr = action2promises_.find(action_index);
  if (itr != action2promises_.end()) {
    if (itr->second.isFulfilled()) {
      return;
    }
  }
  ThrowableWithExtraContext twec(ew, current_time, extras);
  AddAction2Error(action_index, twec);
  action2promises_[action_index].setException(
      RetriesExhaustedException(tries - 1, action2errors_[action_index]));
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::FailAll(
    const std::vector<std::shared_ptr<Action>> &actions, int32_t tries,
    const folly::exception_wrapper &ew, std::shared_ptr<ServerName> server_name) {
  for (const auto action : actions) {
    FailOne(action, tries, ew, TimeUtil::GetNowNanos(), GetExtraContextForError(server_name));
  }
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::FailAll(
    const std::vector<std::shared_ptr<Action>> &actions, int32_t tries) {
  for (const auto action : actions) {
    auto action_index = action->original_index();
    auto itr = action2promises_.find(action_index);
    if (itr->second.isFulfilled()) {
      return;
    }
    action2promises_[action_index].setException(
        RetriesExhaustedException(tries, action2errors_[action_index]));
  }
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::AddAction2Error(
    uint64_t action_index, const ThrowableWithExtraContext &twec) {
  auto erritr = action2errors_.find(action_index);
  if (erritr != action2errors_.end()) {
    erritr->second->push_back(twec);
  } else {
    action2errors_[action_index] = std::make_shared<std::vector<ThrowableWithExtraContext>>();
    action2errors_[action_index]->push_back(twec);
  }
  return;
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::OnError(const ActionsByRegion &actions_by_region,
                                                     int32_t tries,
                                                     const folly::exception_wrapper &ew,
                                                     std::shared_ptr<ServerName> server_name) {
  std::vector<std::shared_ptr<Action>> copied_actions;
  std::vector<std::shared_ptr<RegionRequest>> region_requests;
  for (const auto &action_by_region : actions_by_region) {
    region_requests.push_back(action_by_region.second);
    for (const auto &action : action_by_region.second->actions()) {
      copied_actions.push_back(action);
    }
  }

  LogException(tries, region_requests, ew, server_name);
  if ((tries >= max_attempts_) || !ExceptionUtil::ShouldRetry(ew)) {
    FailAll(copied_actions, tries, ew, server_name);
    return;
  }
  AddError(copied_actions, ew, server_name);
  TryResubmit(copied_actions, tries);
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::TryResubmit(
    const std::vector<std::shared_ptr<Action>> &actions, int32_t tries) {
  int64_t delay_ns;
  if (operation_timeout_ns_.count() > 0) {
    int64_t max_delay_ns = RemainingTimeNs() - ConnectionUtils::kSleepDeltaNs;
    if (max_delay_ns <= 0) {
      FailAll(actions, tries);
      return;
    }
    delay_ns = std::min(max_delay_ns, ConnectionUtils::GetPauseTime(pause_ns_.count(), tries - 1));
  } else {
    delay_ns = ConnectionUtils::GetPauseTime(pause_ns_.count(), tries - 1);
  }

  conn_->retry_executor()->add([=]() {
    retry_timer_->scheduleTimeoutFn(
        [=]() { conn_->cpu_executor()->add([=]() { GroupAndSend(actions, tries + 1); }); },
        milliseconds(TimeUtil::ToMillis(delay_ns)));
  });
}

template <typename REQ, typename RESP>
Future<std::vector<Try<std::shared_ptr<RegionLocation>>>>
AsyncBatchRpcRetryingCaller<REQ, RESP>::GetRegionLocations(
    const std::vector<std::shared_ptr<Action>> &actions, int64_t locate_timeout_ns) {
  auto locs = std::vector<Future<std::shared_ptr<RegionLocation>>>{};
  for (auto const &action : actions) {
    locs.push_back(location_cache_->LocateRegion(*table_name_, action->action()->row(),
                                                 RegionLocateType::kCurrent, locate_timeout_ns));
  }

  return collectAll(locs);
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::GroupAndSend(
    const std::vector<std::shared_ptr<Action>> &actions, int32_t tries) {
  int64_t locate_timeout_ns;
  if (operation_timeout_ns_.count() > 0) {
    locate_timeout_ns = RemainingTimeNs();
    if (locate_timeout_ns <= 0) {
      FailAll(actions, tries);
      return;
    }
  } else {
    locate_timeout_ns = -1L;
  }

  GetRegionLocations(actions, locate_timeout_ns)
      .then([=](std::vector<Try<std::shared_ptr<RegionLocation>>> &loc) {
        std::lock_guard<std::recursive_mutex> lck(multi_mutex_);
        ActionsByServer actions_by_server;
        std::vector<std::shared_ptr<Action>> locate_failed;

        for (uint64_t i = 0; i < loc.size(); ++i) {
          auto action = actions[i];
          if (loc[i].hasValue()) {
            auto region_loc = loc[i].value();
            // Add it to actions_by_server;
            auto search =
                actions_by_server.find(std::make_shared<ServerName>(region_loc->server_name()));
            if (search != actions_by_server.end()) {
              search->second->AddActionsByRegion(region_loc, action);
            } else {
              auto server_request = std::make_shared<ServerRequest>(region_loc);
              server_request->AddActionsByRegion(region_loc, action);
              auto server_name = std::make_shared<ServerName>(region_loc->server_name());
              actions_by_server[server_name] = server_request;
            }
            VLOG(5) << "rowkey [" << action->action()->row() << "] of table["
                    << table_name_->ShortDebugString() << "] found in ["
                    << region_loc->region_name() << "]; RS["
                    << region_loc->server_name().host_name() << ":"
                    << region_loc->server_name().port() << "];";
          } else if (loc[i].hasException()) {
            folly::exception_wrapper ew = loc[i].exception();
            VLOG(1) << "GetRegionLocations() exception: " << ew.what().toStdString()
                    << "for index:" << i << "; tries: " << tries
                    << "; max_attempts_: " << max_attempts_;
            // We might receive runtime error from location-cache.cc too, we are doing FailOne and
            // continue next one
            if (tries >= max_attempts_ || !ExceptionUtil::ShouldRetry(ew)) {
              FailOne(action, tries, ew, TimeUtil::GetNowNanos(), ew.what().toStdString());
            } else {
              AddError(action, loc[i].exception(), nullptr);
              locate_failed.push_back(action);
            }
          }
        }
        if (!actions_by_server.empty()) {
          Send(actions_by_server, tries);
        }

        if (!locate_failed.empty()) {
          TryResubmit(locate_failed, tries);
        }
      })
      .onError([=](const folly::exception_wrapper &ew) {
        VLOG(1) << "GetRegionLocations() exception: " << ew.what().toStdString()
                << "tries: " << tries << "; max_attempts_: " << max_attempts_;
        std::lock_guard<std::recursive_mutex> lck(multi_mutex_);
        if (tries >= max_attempts_ || !ExceptionUtil::ShouldRetry(ew)) {
          FailAll(actions, tries, ew, nullptr);
        } else {
          TryResubmit(actions, tries);
        }
      });
  return;
}

template <typename REQ, typename RESP>
Future<std::vector<Try<std::unique_ptr<Response>>>>
AsyncBatchRpcRetryingCaller<REQ, RESP>::GetMultiResponse(const ActionsByServer &actions_by_server) {
  auto multi_calls = std::vector<Future<std::unique_ptr<hbase::Response>>>{};
  auto user = User::defaultUser();
  for (const auto &action_by_server : actions_by_server) {
    std::unique_ptr<Request> multi_req =
        RequestConverter::ToMultiRequest(action_by_server.second->actions_by_region());
    auto host = action_by_server.first->host_name();
    int port = action_by_server.first->port();
    multi_calls.push_back(
        rpc_client_->AsyncCall(host, port, std::move(multi_req), user, "ClientService"));
  }
  return collectAll(multi_calls);
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::Send(const ActionsByServer &actions_by_server,
                                                  int32_t tries) {
  int64_t remaining_ns;
  if (operation_timeout_ns_.count() > 0) {
    remaining_ns = RemainingTimeNs();
    if (remaining_ns <= 0) {
      std::vector<std::shared_ptr<Action>> failed_actions;
      for (const auto &action_by_server : actions_by_server) {
        for (auto &value : action_by_server.second->actions_by_region()) {
          for (const auto &failed_action : value.second->actions()) {
            failed_actions.push_back(failed_action);
          }
        }
      }
      FailAll(failed_actions, tries);
      return;
    }
  } else {
    remaining_ns = std::numeric_limits<int64_t>::max();
  }

  std::vector<std::shared_ptr<Request>> multi_reqv;
  for (const auto &action_by_server : actions_by_server)
    multi_reqv.push_back(
        std::move(RequestConverter::ToMultiRequest(action_by_server.second->actions_by_region())));

  GetMultiResponse(actions_by_server)
      .then([=](const std::vector<Try<std::unique_ptr<hbase::Response>>> &completed_responses) {
        std::lock_guard<std::recursive_mutex> lck(multi_mutex_);
        uint64_t num = 0;
        for (const auto &action_by_server : actions_by_server) {
          if (completed_responses[num].hasValue()) {
            auto multi_response =
                ResponseConverter::GetResults(multi_reqv[num], *completed_responses[num].value(),
                                              action_by_server.second->actions_by_region());
            OnComplete(action_by_server.second->actions_by_region(), tries, action_by_server.first,
                       std::move(multi_response));
          } else if (completed_responses[num].hasException()) {
            folly::exception_wrapper ew = completed_responses[num].exception();
            VLOG(1) << "GetMultiResponse() exception: " << ew.what().toStdString()
                    << " from server for action index:" << num;
            OnError(action_by_server.second->actions_by_region(), tries, ew,
                    action_by_server.first);
          }
          num++;
        }
      })
      .onError([=](const folly::exception_wrapper &ew) {
        VLOG(1) << "GetMultiResponse() exception: " << ew.what().toStdString();
        std::lock_guard<std::recursive_mutex> lck(multi_mutex_);
        for (const auto &action_by_server : actions_by_server) {
          OnError(action_by_server.second->actions_by_region(), tries, ew, action_by_server.first);
        }
      });
  return;
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::OnComplete(
    const ActionsByRegion &actions_by_region, int32_t tries,
    const std::shared_ptr<ServerName> server_name,
    const std::unique_ptr<hbase::MultiResponse> multi_response) {
  std::vector<std::shared_ptr<Action>> failed_actions;
  const auto region_results = multi_response->RegionResults();
  for (const auto &action_by_region : actions_by_region) {
    auto region_result_itr = region_results.find(action_by_region.first);
    if (region_result_itr != region_results.end()) {
      for (const auto &action : action_by_region.second->actions()) {
        OnComplete(action, action_by_region.second, tries, server_name, region_result_itr->second,
                   failed_actions);
      }
    } else if (region_result_itr == region_results.end()) {
      auto region_exc = multi_response->RegionException(action_by_region.first);
      if (region_exc == nullptr) {
        // FailAll actions for this particular region as inconsistent server response. So we raise
        // this exception to the application
        std::string err_msg = "Invalid response: Server " + server_name->ShortDebugString() +
                              " sent us neither results nor exceptions for " +
                              action_by_region.first;
        VLOG(1) << err_msg;
        auto ew = folly::make_exception_wrapper<std::runtime_error>(err_msg);
        FailAll(action_by_region.second->actions(), tries, ew, server_name);
      } else {
        // Eg: org.apache.hadoop.hbase.NotServingRegionException:
        LogException(tries, action_by_region.second, *region_exc, server_name);
        if (tries >= max_attempts_ || !ExceptionUtil::ShouldRetry(*region_exc)) {
          FailAll(action_by_region.second->actions(), tries, *region_exc, server_name);
          return;
        }
        location_cache_->UpdateCachedLocation(*action_by_region.second->region_location(),
                                              *region_exc);
        AddError(action_by_region.second->actions(), *region_exc, server_name);
        for (const auto &action : action_by_region.second->actions()) {
          failed_actions.push_back(action);
        }
      }
    }
  }
  if (!failed_actions.empty()) {
    TryResubmit(failed_actions, tries);
  }

  return;
}

template <typename REQ, typename RESP>
void AsyncBatchRpcRetryingCaller<REQ, RESP>::OnComplete(
    const std::shared_ptr<Action> &action, const std::shared_ptr<RegionRequest> &region_request,
    int32_t tries, const std::shared_ptr<ServerName> &server_name,
    const std::shared_ptr<RegionResult> &region_result,
    std::vector<std::shared_ptr<Action>> &failed_actions) {
  std::string err_msg;
  try {
    auto result_or_exc = region_result->ResultOrException(action->original_index());
    auto result = std::get<0>(*result_or_exc);
    auto exc = std::get<1>(*result_or_exc);
    if (exc != nullptr) {
      LogException(tries, region_request, *exc, server_name);
      if (tries >= max_attempts_ || !ExceptionUtil::ShouldRetry(*exc)) {
        FailOne(action, tries, *exc, TimeUtil::GetNowNanos(), GetExtraContextForError(server_name));
      } else {
        failed_actions.push_back(action);
      }
    } else if (result != nullptr) {
      action2promises_[action->original_index()].setValue(std::move(result));
    } else {
      std::string err_msg = "Invalid response: Server " + server_name->ShortDebugString() +
                            " sent us neither results nor exceptions for request @ index " +
                            std::to_string(action->original_index()) + ", row " +
                            action->action()->row() + " of " +
                            region_request->region_location()->region_name();
      VLOG(1) << err_msg;
      auto ew = folly::make_exception_wrapper<std::runtime_error>(err_msg);
      AddError(action, ew, server_name);
      failed_actions.push_back(action);
    }
  } catch (const std::out_of_range &oor) {
    // This should never occur. Error in logic. Throwing std::runtime_error from here. Will be
    // retried or failed
    std::string err_msg = "ResultOrException not present @ index " +
                          std::to_string(action->original_index()) + ", row " +
                          action->action()->row() + " of " +
                          region_request->region_location()->region_name();
    throw std::runtime_error(err_msg);
  }
  return;
}

template class AsyncBatchRpcRetryingCaller<std::shared_ptr<hbase::Row>,
                                           std::shared_ptr<hbase::Result>>;
} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/client/time-range.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/time-range.h"
#include <limits>
#include <stdexcept>
#include <string>

namespace hbase {

TimeRange::TimeRange()
    : min_timestamp_(0L), max_timestamp_(std::numeric_limits<int64_t>::max()), all_time_(true) {}

TimeRange::TimeRange(const TimeRange &tr) {
  this->all_time_ = tr.all_time_;
  this->max_timestamp_ = tr.max_timestamp_;
  this->min_timestamp_ = tr.min_timestamp_;
}

TimeRange &TimeRange::operator=(const TimeRange &tr) {
  this->all_time_ = tr.all_time_;
  this->max_timestamp_ = tr.max_timestamp_;
  this->min_timestamp_ = tr.min_timestamp_;
  return *this;
}

TimeRange::~TimeRange() {}

TimeRange::TimeRange(int64_t min_timestamp) {
  this->min_timestamp_ = min_timestamp;
  this->max_timestamp_ = std::numeric_limits<int64_t>::max();
  this->all_time_ = false;
}

TimeRange::TimeRange(int64_t min_timestamp, int64_t max_timestamp) {
  if (min_timestamp < 0 || max_timestamp < 0) {
    throw std::runtime_error("Timestamp cannot be negative. min_timestamp: " +
                             std::to_string(min_timestamp) + ", max_timestamp:" +
                             std::to_string(max_timestamp));
  }
  if (max_timestamp < min_timestamp) {
    throw std::runtime_error("max_timestamp [" + std::to_string(max_timestamp) +
                             "] should be greater than min_timestamp [" +
                             std::to_string(min_timestamp) + "]");
  }

  this->min_timestamp_ = min_timestamp;
  this->max_timestamp_ = max_timestamp;
  this->all_time_ = false;
}

int64_t TimeRange::MinTimeStamp() const { return this->min_timestamp_; }

int64_t TimeRange::MaxTimeStamp() const { return this->max_timestamp_; }

bool TimeRange::IsAllTime() const { return this->all_time_; }
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/connection/connection-factory.cc<|end_filename|>
/*
 * 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.
 *
 */

#include <folly/Conv.h>
#include <glog/logging.h>
#include <wangle/channel/Handler.h>

#include <chrono>

#include <folly/ExceptionWrapper.h>
#include <folly/SocketAddress.h>
#include <folly/io/async/AsyncSocketException.h>

#include "hbase/connection/client-dispatcher.h"
#include "hbase/connection/connection-factory.h"
#include "hbase/connection/pipeline.h"
#include "hbase/connection/sasl-handler.h"
#include "hbase/connection/service.h"
#include "hbase/exceptions/exception.h"

using std::chrono::milliseconds;
using std::chrono::nanoseconds;

namespace hbase {

ConnectionFactory::ConnectionFactory(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
                                     std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
                                     std::shared_ptr<Codec> codec,
                                     std::shared_ptr<Configuration> conf,
                                     nanoseconds connect_timeout)
    : connect_timeout_(connect_timeout),
      io_executor_(io_executor),
      cpu_executor_(cpu_executor),
      conf_(conf),
      pipeline_factory_(std::make_shared<RpcPipelineFactory>(codec, conf)) {}

std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>> ConnectionFactory::MakeBootstrap() {
  auto client = std::make_shared<wangle::ClientBootstrap<SerializePipeline>>();
  client->group(io_executor_);
  client->pipelineFactory(pipeline_factory_);

  // TODO: Opened https://github.com/facebook/wangle/issues/85 in wangle so that we can set socket
  //  options like TCP_NODELAY, SO_KEEPALIVE, CONNECT_TIMEOUT_MILLIS, etc.

  return client;
}

std::shared_ptr<HBaseService> ConnectionFactory::Connect(
    std::shared_ptr<RpcConnection> rpc_connection,
    std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>> client_bootstrap,
    const std::string &hostname, uint16_t port) {
  // connection should happen from an IO thread
  try {
    auto future = via(io_executor_.get()).then([=]() {
      VLOG(1) << "Connecting to server: " << hostname << ":" << port;
      return client_bootstrap->connect(folly::SocketAddress(hostname, port, true),
                                       std::chrono::duration_cast<milliseconds>(connect_timeout_));
    });

    // See about using shared promise for this.
    auto pipeline = future.get();

    VLOG(1) << "Connected to server: " << hostname << ":" << port;
    auto dispatcher =
        std::make_shared<ClientDispatcher>(hostname + ":" + folly::to<std::string>(port));
    dispatcher->setPipeline(pipeline);
    return dispatcher;
  } catch (const folly::AsyncSocketException &e) {
    throw ConnectionException(folly::exception_wrapper{e});
  }
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/client-dispatcher.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <folly/Logging.h>
#include <wangle/service/ClientDispatcher.h>

#include <atomic>
#include <map>
#include <memory>
#include <mutex>
#include <string>

#include "hbase/connection/pipeline.h"
#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/utils/concurrent-map.h"

namespace hbase {

/**
 * Dispatcher that assigns a call_id and then routes the response back to the
 * future.
 */
class ClientDispatcher
    : public wangle::ClientDispatcherBase<SerializePipeline, std::unique_ptr<Request>,
                                          std::unique_ptr<Response>> {
 public:
  /** Create a new ClientDispatcher */
  explicit ClientDispatcher(const std::string &server);
  /** Read a response off the pipeline. */
  void read(Context *ctx, std::unique_ptr<Response> in) override;
  void readException(Context *ctx, folly::exception_wrapper e) override;
  void readEOF(Context *ctx) override;
  /** Take a request as a call and send it down the pipeline. */
  folly::Future<std::unique_ptr<Response>> operator()(std::unique_ptr<Request> arg) override;
  /** Close the dispatcher and the associated pipeline. */
  folly::Future<folly::Unit> close(Context *ctx) override;
  /** Close the dispatcher and the associated pipeline. */
  folly::Future<folly::Unit> close() override;

 private:
  void CloseAndCleanUpCalls();

 private:
  std::recursive_mutex mutex_;
  concurrent_map<uint32_t, folly::Promise<std::unique_ptr<Response>>> requests_;
  // Start at some number way above what could
  // be there for un-initialized call id counters.
  //
  // This makes it easier to make sure that the're are
  // no access to un-initialized variables.
  //
  // uint32_t has a max of 4Billion so 10 more or less is
  // not a big deal.
  std::atomic<uint32_t> current_call_id_;
  std::string server_;
  bool is_closed_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/client-handler.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <wangle/channel/Handler.h>

#include <atomic>
#include <memory>
#include <mutex>
#include <string>
#include <utility>

#include "hbase/client/configuration.h"
#include "hbase/exceptions/exception.h"
#include "hbase/serde/codec.h"
#include "hbase/serde/rpc-serde.h"
#include "hbase/utils/concurrent-map.h"

// Forward decs.
namespace hbase {
class Request;
class Response;
class HeaderInfo;
}
namespace google {
namespace protobuf {
class Message;
}
}

namespace hbase {

/**
 * wangle::Handler implementation to convert hbase::Request to IOBuf and
 * convert IOBuf to hbase::Response.
 *
 * This class deals with sending the connection header and preamble
 * on first request.
 */
class ClientHandler
    : public wangle::Handler<std::unique_ptr<folly::IOBuf>, std::unique_ptr<Response>,
                             std::unique_ptr<Request>, std::unique_ptr<folly::IOBuf>> {
 public:
  /**
   * Create the handler
   * @param user_name the user name of the user running this process.
   */
  ClientHandler(std::string user_name, std::shared_ptr<Codec> codec,
                std::shared_ptr<Configuration> conf, const std::string &server);

  /**
   * Get bytes from the wire.
   * This should be the full message as the length field decoder should be
   * in the pipeline before this.
   */
  void read(Context *ctx, std::unique_ptr<folly::IOBuf> msg) override;

  /**
   * Write the data down the wire.
   */
  folly::Future<folly::Unit> write(Context *ctx, std::unique_ptr<Request> r) override;

 private:
  std::unique_ptr<std::once_flag> once_flag_;
  std::string user_name_;
  RpcSerde serde_;
  std::string server_;  // for logging
  std::shared_ptr<Configuration> conf_;

  // in flight requests
  std::unique_ptr<concurrent_map<uint32_t, std::shared_ptr<google::protobuf::Message>>> resp_msgs_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/zk-util.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <cstdlib>
#include <string>
#include "hbase/client/configuration.h"

namespace hbase {

class ZKUtil {
 public:
  static constexpr const char* kHBaseZookeeperQuorum_ = "hbase.zookeeper.quorum";
  static constexpr const char* kDefHBaseZookeeperQuorum_ = "localhost:2181";
  static constexpr const char* kHBaseZookeeperClientPort_ = "hbase.zookeeper.property.clientPort";
  static constexpr const int32_t kDefHBaseZookeeperClientPort_ = 2181;
  static constexpr const char* kHBaseZnodeParent_ = "zookeeper.znode.parent";
  static constexpr const char* kDefHBaseZnodeParent_ = "/hbase";
  static constexpr const char* kHBaseMetaRegionServer_ = "meta-region-server";

  static constexpr const char* kHBaseZookeeperSessionTimeout_ = "zookeeper.session.timeout";
  static constexpr const int32_t kDefHBaseZookeeperSessionTimeout_ = 90000;

  static std::string ParseZooKeeperQuorum(const hbase::Configuration& conf);

  static std::string MetaZNode(const hbase::Configuration& conf);

  static int32_t SessionTimeout(const hbase::Configuration& conf);
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/request.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/Conv.h>
#include <google/protobuf/message.h>

#include <cstdint>
#include <memory>
#include <string>

namespace hbase {

/**
 * Main request class.
 * This holds the request object and the un-filled in approriatley typed
 * response object.
 */
class Request {
 public:
  /** Create a request object for a get */
  static std::unique_ptr<Request> get();
  /** Create a request object for a mutate */
  static std::unique_ptr<Request> mutate();
  /** Create a request object for a scan */
  static std::unique_ptr<Request> scan();
  /** Create a request object for a multi */
  static std::unique_ptr<Request> multi();

  /**
   * This should be private. Do not use this.
   *
   *
   * Constructor that's public for make_unique. This sets all the messages and
   * method name.
   */
  Request(std::shared_ptr<google::protobuf::Message> req,
          std::shared_ptr<google::protobuf::Message> resp, std::string method);

  /** Get the call id. */
  uint32_t call_id() { return call_id_; }
  /** Set the call id. This should only be set once. */
  void set_call_id(uint32_t call_id) { call_id_ = call_id; }
  /** Get the backing request protobuf message. */
  std::shared_ptr<google::protobuf::Message> req_msg() { return req_msg_; }
  /** Get the backing response protobuf message. */
  std::shared_ptr<google::protobuf::Message> resp_msg() { return resp_msg_; }
  /** Get the method name. This is used to the the receiving rpc server what
   * method type to decode. */
  std::string method() { return method_; }

  std::string DebugString() {
    return "call_id:" + folly::to<std::string>(call_id_) + ", req_msg:" +
           req_msg_->ShortDebugString() + ", method:" + method_;
  }

 private:
  uint32_t call_id_;
  std::shared_ptr<google::protobuf::Message> req_msg_ = nullptr;
  std::shared_ptr<google::protobuf::Message> resp_msg_ = nullptr;
  std::string method_ = "Get";
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/rpc-connection.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <memory>
#include <mutex>
#include <utility>

#include "hbase/connection/connection-factory.h"
#include "hbase/connection/connection-id.h"
#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/connection/service.h"

namespace hbase {

class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
 public:
  RpcConnection(std::shared_ptr<ConnectionId> connection_id, std::shared_ptr<ConnectionFactory> cf)
      : connection_id_(connection_id), cf_(cf), hbase_service_(nullptr) {}

  virtual ~RpcConnection() { Close(); }

  virtual std::shared_ptr<ConnectionId> remote_id() const { return connection_id_; }

  virtual folly::Future<std::unique_ptr<Response>> SendRequest(std::unique_ptr<Request> req) {
    std::lock_guard<std::recursive_mutex> lock(mutex_);
    if (hbase_service_ == nullptr) {
      Connect();
    }
    VLOG(5) << "Calling RpcConnection::SendRequest()";  // TODO
    return (*hbase_service_)(std::move(req));
  }

  virtual void Close() {
    std::lock_guard<std::recursive_mutex> lock(mutex_);
    if (hbase_service_) {
      hbase_service_->close();
      hbase_service_ = nullptr;
    }
    if (client_bootstrap_) {
      client_bootstrap_ = nullptr;
    }
  }

 private:
  void Connect() {
    client_bootstrap_ = cf_->MakeBootstrap();
    auto dispatcher = cf_->Connect(shared_from_this(), client_bootstrap_, remote_id()->host(),
                                   remote_id()->port());
    hbase_service_ = std::move(dispatcher);
  }

 private:
  std::recursive_mutex mutex_;
  std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor_;
  std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor_;
  std::shared_ptr<ConnectionId> connection_id_;
  std::shared_ptr<HBaseService> hbase_service_;
  std::shared_ptr<ConnectionFactory> cf_;
  std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>> client_bootstrap_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/serde/cell-scanner.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once
#include <folly/io/IOBuf.h>
#include <memory>

namespace hbase {

class Cell;

/**
 * @brief Interface for iterating over a sequence of Cells
 */
class CellScanner {
 public:
  virtual ~CellScanner() {}

  /**
   * @brief This method will be used to iterate the cells.
   * Typical usage will be :-
   * while(cell_scanner.Advance()){
   *  auto current_cell = cell_scanner.Current();
   * }
   */
  virtual bool Advance() = 0;

  /**
   * @brief returns the current cell
   */
  virtual const std::shared_ptr<Cell> Current() const = 0;
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/result.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <functional>
#include <map>
#include <memory>
#include <string>
#include <vector>

#include "hbase/client/cell.h"
#include "hbase/utils/optional.h"

namespace hbase {

/**
 * @brief Map of families to all versions of its qualifiers and values
 * We need to have a reverse ordered map, when storing TS -> value, so that the
 * most recent value is stored first
 */
using ResultMap =
    std::map<std::string,
             std::map<std::string, std::map<int64_t, std::string, std::greater<int64_t> > > >;

class Result {
 public:
  /**
   * Constructors
   */
  Result(const std::vector<std::shared_ptr<Cell> > &cells, bool exists, bool stale, bool partial);
  Result(const Result &result);
  ~Result();

  /**
   * @brief Return the vector of Cells backing this Result instance. This vector
   * will be ordered in the same manner
   * as the one which was passed while creation of the Result instance.
   */
  const std::vector<std::shared_ptr<Cell> > &Cells() const;

  /**
   * @brief  Return a vector of Cells for the family and qualifier or empty list
   * if the column
   * did not exist in the result.
   * @param family - column family
   * @param qualifier - column qualifier
   */
  std::vector<std::shared_ptr<Cell> > ColumnCells(const std::string &family,
                                                  const std::string &qualifier) const;

  /**
   * @brief Returns the Cell for the most recent timestamp for a given family
   * and qualifier.
   * Returns map of qualifiers to values, only includes latest values
   * @param family - column family.
   * @param qualifier - column qualifier
   */
  const std::shared_ptr<Cell> ColumnLatestCell(const std::string &family,
                                               const std::string &qualifier) const;

  /**
   * @brief Get the latest version of the specified family and qualifier.
   * @param family - column family
   * @param qualifier - column qualifier
   */
  optional<std::string> Value(const std::string &family, const std::string &qualifier) const;

  /**
   * @brief Returns if the underlying Cell vector is empty or not
   */
  bool IsEmpty() const;

  /**
   * @brief Retrieves the row key that corresponds to the row from which this
   * Result was created.
   */
  const std::string &Row() const;

  /**
   * @brief Returns the size of the underlying Cell vector
   */
  int Size() const;

  /**
   * @brief Map of families to all versions of its qualifiers and values.
   * Returns a three level Map of the form:
   * Map<family,Map<qualifier,Map<timestamp,value>>>>
   * All other map returning methods make use of this map internally
   * The Map is created when the Result instance is created
   */
  ResultMap Map() const;

  /**
   * @brief Map of qualifiers to values.
   * Returns a Map of the form: Map<qualifier,value>
   * @param family - column family to get
   */
  std::map<std::string, std::string> FamilyMap(const std::string &family) const;

  std::string DebugString() const;

  bool Exists() const { return exists_; }

  bool Stale() const { return stale_; }

  bool Partial() const { return partial_; }

  /** Returns estimated size of the Result object including deep heap space usage
   * of its Cells and data. Notice that this is a very rough estimate. */
  size_t EstimatedSize() const;

 private:
  bool exists_ = false;
  bool stale_ = false;
  bool partial_ = false;
  std::string row_ = "";
  std::vector<std::shared_ptr<Cell> > cells_;
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/connection/rpc-client.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/connection/rpc-client.h"

#include <folly/Format.h>
#include <folly/Logging.h>
#include <folly/futures/Future.h>
#include <unistd.h>
#include <wangle/concurrent/IOThreadPoolExecutor.h>
#include "hbase/exceptions/exception.h"

using hbase::security::User;
using std::chrono::nanoseconds;

namespace hbase {

RpcClient::RpcClient(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
                     std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
                     std::shared_ptr<Codec> codec, std::shared_ptr<Configuration> conf,
                     nanoseconds connect_timeout)
    : io_executor_(io_executor), conf_(conf) {
  cp_ = std::make_shared<ConnectionPool>(io_executor_, cpu_executor, codec, conf, connect_timeout);
}

void RpcClient::Close() { io_executor_->stop(); }

std::unique_ptr<Response> RpcClient::SyncCall(const std::string& host, uint16_t port,
                                              std::unique_ptr<Request> req,
                                              std::shared_ptr<User> ticket) {
  return AsyncCall(host, port, std::move(req), ticket).get();
}

std::unique_ptr<Response> RpcClient::SyncCall(const std::string& host, uint16_t port,
                                              std::unique_ptr<Request> req,
                                              std::shared_ptr<User> ticket,
                                              const std::string& service_name) {
  return AsyncCall(host, port, std::move(req), ticket, service_name).get();
}

folly::Future<std::unique_ptr<Response>> RpcClient::AsyncCall(const std::string& host,
                                                              uint16_t port,
                                                              std::unique_ptr<Request> req,
                                                              std::shared_ptr<User> ticket) {
  auto remote_id = std::make_shared<ConnectionId>(host, port, ticket);
  return SendRequest(remote_id, std::move(req));
}

folly::Future<std::unique_ptr<Response>> RpcClient::AsyncCall(const std::string& host,
                                                              uint16_t port,
                                                              std::unique_ptr<Request> req,
                                                              std::shared_ptr<User> ticket,
                                                              const std::string& service_name) {
  auto remote_id = std::make_shared<ConnectionId>(host, port, ticket, service_name);
  return SendRequest(remote_id, std::move(req));
}

/**
 * There are two cases for ConnectionException:
 * 1. The first time connection
 * establishment, i.e. GetConnection(remote_id), AsyncSocketException being a cause.
 * 2. Writing request down the pipeline, i.e. RpcConnection::SendRequest, AsyncSocketException being
 * a cause as well.
 */
folly::Future<std::unique_ptr<Response>> RpcClient::SendRequest(
    std::shared_ptr<ConnectionId> remote_id, std::unique_ptr<Request> req) {
  try {
    return GetConnection(remote_id)
        ->SendRequest(std::move(req))
        .onError([&, this](const folly::exception_wrapper& ew) {
          VLOG(3) << folly::sformat("RpcClient Exception: {}", ew.what());
          ew.with_exception([&, this](const hbase::ConnectionException& re) {
            /* bad connection, remove it from pool. */
            cp_->Close(remote_id);
          });
          return GetFutureWithException(ew);
        });
  } catch (const ConnectionException& e) {
    CHECK(e.cause().get_exception() != nullptr);
    VLOG(3) << folly::sformat("RpcClient Exception: {}", e.cause().what());
    /* bad connection, remove it from pool. */
    cp_->Close(remote_id);
    return GetFutureWithException(e);
  }
}

template <typename EXCEPTION>
folly::Future<std::unique_ptr<Response>> RpcClient::GetFutureWithException(const EXCEPTION& e) {
  return GetFutureWithException(folly::exception_wrapper{e});
}

folly::Future<std::unique_ptr<Response>> RpcClient::GetFutureWithException(
    const folly::exception_wrapper& ew) {
  folly::Promise<std::unique_ptr<Response>> promise;
  auto future = promise.getFuture();
  promise.setException(ew);
  return future;
}

std::shared_ptr<RpcConnection> RpcClient::GetConnection(std::shared_ptr<ConnectionId> remote_id) {
  return cp_->GetConnection(remote_id);
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/table.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <chrono>
#include <memory>
#include <string>
#include <vector>

#include "hbase/connection/rpc-client.h"
#include "hbase/client/async-connection.h"
#include "hbase/client/client.h"
#include "hbase/client/configuration.h"
#include "hbase/client/get.h"
#include "hbase/client/location-cache.h"
#include "hbase/client/put.h"
#include "hbase/client/raw-async-table.h"
#include "hbase/client/result-scanner.h"
#include "hbase/client/result.h"
#include "hbase/serde/table-name.h"

namespace hbase {

class Client;

class Table {
 public:
  /**
   * Constructors
   */
  Table(const pb::TableName &table_name, std::shared_ptr<AsyncConnection> async_connection);
  ~Table();

  /**
   * @brief - Returns a Result object for the constructed Get.
   * @param - get Get object to perform HBase Get operation.
   */
  std::shared_ptr<hbase::Result> Get(const hbase::Get &get);

  std::vector<std::shared_ptr<hbase::Result>> Get(const std::vector<hbase::Get> &gets);

  /**
   * @brief - Puts some data in the table.
   * @param - put Put object to perform HBase Put operation.
   */
  void Put(const hbase::Put &put);

  /**
   * Atomically checks if a row/family/qualifier value matches the expected
   * value. If it does, it adds the put.  If the passed value is null, the check
   * is for the lack of column (ie: non-existance)
   *
   * @param row to check
   * @param family column family to check
   * @param qualifier column qualifier to check
   * @param value the expected value
   * @param put data to put if check succeeds
   * @param compare_op comparison operator to use
   * @throws IOException e
   * @return true if the new put was executed, false otherwise
   */
  bool CheckAndPut(const std::string &row, const std::string &family, const std::string &qualifier,
                   const std::string &value, const hbase::Put &put,
                   const pb::CompareType &compare_op = pb::CompareType::EQUAL);
  /**
   * @brief - Deletes some data in the table.
   * @param - del Delete object to perform HBase Delete operation.
   */
  void Delete(const hbase::Delete &del);

  /**
   * Atomically checks if a row/family/qualifier value matches the expected
   * value. If it does, it adds the delete.  If the passed value is null, the
   * check is for the lack of column (ie: non-existence)
   *
   * The expected value argument of this call is on the left and the current
   * value of the cell is on the right side of the comparison operator.
   *
   * Ie. eg. GREATER operator means expected value > existing <=> add the delete.
   *
   * @param row to check
   * @param family column family to check
   * @param qualifier column qualifier to check
   * @param compare_op comparison operator to use
   * @param value the expected value
   * @param del data to delete if check succeeds
   * @return true if the new delete was executed, false otherwise
   */
  bool CheckAndDelete(const std::string &row, const std::string &family,
                      const std::string &qualifier, const std::string &value,
                      const hbase::Delete &del,
                      const pb::CompareType &compare_op = pb::CompareType::EQUAL);

  /**
   * @brief - Increments some data in the table.
   * @param - increment Increment object to perform HBase Increment operation.
   */
  std::shared_ptr<hbase::Result> Increment(const hbase::Increment &increment);

  /**
   * @brief - Appends some data in the table.
   * @param - append Append object to perform HBase Append operation.
   */
  std::shared_ptr<hbase::Result> Append(const hbase::Append &append);

  std::shared_ptr<ResultScanner> Scan(const hbase::Scan &scan);

  /**
     * @brief - Multi Puts.
     * @param - puts vector of hbase::Put.
     */
  void Put(const std::vector<hbase::Put> &puts);
  /**
   * @brief - Close the client connection.
   */
  void Close();

  /**
   * @brief - Get region location for a row in current table.
   */
  std::shared_ptr<RegionLocation> GetRegionLocation(const std::string &row);

 private:
  std::shared_ptr<pb::TableName> table_name_;
  std::shared_ptr<AsyncConnection> async_connection_;
  std::shared_ptr<hbase::Configuration> conf_;
  std::unique_ptr<RawAsyncTable> async_table_;

 private:
  std::chrono::milliseconds operation_timeout() const;

  int64_t ResultSize2CacheSize(int64_t max_results_size) const;
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/connection/connection-id.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <boost/functional/hash.hpp>

#include <memory>
#include <string>
#include <utility>

#include "hbase/if/HBase.pb.h"
#include "hbase/security/user.h"

namespace hbase {

class ConnectionId {
 public:
  ConnectionId(const std::string &host, uint16_t port)
      : ConnectionId(host, port, security::User::defaultUser(), "") {}

  ConnectionId(const std::string &host, uint16_t port, std::shared_ptr<security::User> user)
      : ConnectionId(host, port, user, "") {}

  ConnectionId(const std::string &host, uint16_t port, std::shared_ptr<security::User> user,
               const std::string &service_name)
      : user_(user), service_name_(service_name), host_(host), port_(port) {}

  ConnectionId(const std::string &host, uint16_t port, const std::string &service_name)
      : user_(security::User::defaultUser()),
        service_name_(service_name),
        host_(host),
        port_(port) {}

  virtual ~ConnectionId() = default;

  std::shared_ptr<security::User> user() const { return user_; }
  std::string service_name() const { return service_name_; }
  std::string host() { return host_; }
  uint16_t port() { return port_; }

 private:
  std::shared_ptr<security::User> user_;
  std::string service_name_;
  std::string host_;
  uint16_t port_;
};

/* Equals function for ConnectionId */
struct ConnectionIdEquals {
  /** equals */
  bool operator()(const std::shared_ptr<ConnectionId> &lhs,
                  const std::shared_ptr<ConnectionId> &rhs) const {
    return userEquals(lhs->user(), rhs->user()) && lhs->host() == rhs->host() &&
           lhs->port() == rhs->port() && lhs->service_name() == rhs->service_name();
  }

 private:
  bool userEquals(const std::shared_ptr<security::User> &lhs,
                  const std::shared_ptr<security::User> &rhs) const {
    return lhs == nullptr ? rhs == nullptr
                          : (rhs == nullptr ? false : lhs->user_name() == rhs->user_name());
  }
};

/** Hash for ConnectionId. */
struct ConnectionIdHash {
  /** hash */
  std::size_t operator()(const std::shared_ptr<ConnectionId> &ci) const {
    std::size_t h = 0;
    boost::hash_combine(h, ci->user() == nullptr ? 0 : ci->user()->user_name());
    boost::hash_combine(h, ci->host());
    boost::hash_combine(h, ci->port());
    boost::hash_combine(h, ci->service_name());
    return h;
  }
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/connection-configuration.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <chrono>
#include <climits>
#include <string>

#include "hbase/client/configuration.h"

namespace hbase {

/**
 * Timeout configs.
 */
class ConnectionConfiguration {
 public:
  explicit ConnectionConfiguration(const Configuration& conf) {
    connect_timeout_ =
        ToNanos(conf.GetInt(kClientSocketConnectTimeout, kDefaultClientSocketConnectTimeout));
    meta_operation_timeout_ =
        ToNanos(conf.GetLong(kClientMetaOperationTimeout, kDefaultClientOperationTimeout));
    operation_timeout_ =
        ToNanos(conf.GetLong(kClientOperationTimeout, kDefaultClientOperationTimeout));
    rpc_timeout_ = ToNanos(conf.GetLong(kRpcTimeout, kDefaultRpcTimeout));
    read_rpc_timeout_ = ToNanos(conf.GetLong(kRpcReadTimeout, ToMillis(rpc_timeout_)));
    write_rpc_timeout_ = ToNanos(conf.GetLong(kRpcWriteTimeout, ToMillis(rpc_timeout_)));
    pause_ = ToNanos(conf.GetLong(kClientPause, kDefaultClientPause));
    max_retries_ = conf.GetInt(kClientRetriesNumber, kDefaultClientRetriesNumber);
    start_log_errors_count_ =
        conf.GetInt(kStartLogErrorsAfterCount, kDefaultStartLogErrorsAfterCount);
    scan_timeout_ =
        ToNanos(conf.GetLong(kClientScannerTimeoutPeriod, kDefaultClientScannerTimeoutPeriod));
    scanner_caching_ = conf.GetInt(kClientScannerCaching, kDefaultClientScannerCaching);
    scanner_max_result_size_ =
        conf.GetLong(kClientScannerMaxResultsSize, kDefaultClientScannerMaxResultsSize);
  }

  // Used by tests
  ConnectionConfiguration(std::chrono::nanoseconds connect_timeout,
                          std::chrono::nanoseconds operation_timeout,
                          std::chrono::nanoseconds rpc_timeout, std::chrono::nanoseconds pause,
                          uint32_t max_retries, uint32_t start_log_errors_count)
      : connect_timeout_(connect_timeout),
        operation_timeout_(operation_timeout),
        meta_operation_timeout_(operation_timeout),
        rpc_timeout_(rpc_timeout),
        read_rpc_timeout_(rpc_timeout),
        write_rpc_timeout_(rpc_timeout),
        pause_(pause),
        max_retries_(max_retries),
        start_log_errors_count_(start_log_errors_count) {}

  std::chrono::nanoseconds connect_timeout() const { return connect_timeout_; }

  std::chrono::nanoseconds meta_operation_timeout() const { return meta_operation_timeout_; }

  // timeout for a whole operation such as get, put or delete. Notice that scan will not be effected
  // by this value, see scanTimeoutNs.
  std::chrono::nanoseconds operation_timeout() const { return operation_timeout_; }

  // timeout for each rpc request. Can be overridden by a more specific config, such as
  // readRpcTimeout or writeRpcTimeout.
  std::chrono::nanoseconds rpc_timeout() const { return rpc_timeout_; }

  // timeout for each read rpc request
  std::chrono::nanoseconds read_rpc_timeout() const { return read_rpc_timeout_; }

  // timeout for each write rpc request
  std::chrono::nanoseconds write_rpc_timeout() const { return write_rpc_timeout_; }

  std::chrono::nanoseconds pause() const { return pause_; }

  uint32_t max_retries() const { return max_retries_; }

  /** How many retries are allowed before we start to log */
  uint32_t start_log_errors_count() const { return start_log_errors_count_; }

  // The scan timeout is used as operation timeout for every
  // operations in a scan, such as openScanner or next.
  std::chrono::nanoseconds scan_timeout() const { return scan_timeout_; }

  uint32_t scanner_caching() const { return scanner_caching_; }

  uint64_t scanner_max_result_size() const { return scanner_max_result_size_; }

 private:
  /** Parameter name for HBase client CPU thread pool size. Defaults to (2 * num cpus) */
  static constexpr const char* kClientSocketConnectTimeout =
      "hbase.ipc.client.socket.timeout.connect";
  /** Parameter name for HBase client CPU thread pool size. Defaults to (2 * num cpus) */
  static constexpr const uint32_t kDefaultClientSocketConnectTimeout = 10000;  // 10 secs

  /** Parameter name for HBase client operation timeout. */
  static constexpr const char* kClientOperationTimeout = "hbase.client.operation.timeout";

  /** Parameter name for HBase client meta operation timeout. */
  static constexpr const char* kClientMetaOperationTimeout = "hbase.client.meta.operation.timeout";

  /** Default HBase client operation timeout, which is tantamount to a blocking call */
  static constexpr const uint32_t kDefaultClientOperationTimeout = 1200000;

  /** timeout for each RPC */
  static constexpr const char* kRpcTimeout = "hbase.rpc.timeout";

  /** timeout for each read RPC */
  static constexpr const char* kRpcReadTimeout = "hbase.rpc.read.timeout";

  /** timeout for each write RPC */
  static constexpr const char* kRpcWriteTimeout = "hbase.rpc.write.timeout";

  static constexpr const uint32_t kDefaultRpcTimeout = 60000;

  /**
    * Parameter name for client pause value, used mostly as value to wait
    * before running a retry of a failed get, region lookup, etc.
    */
  static constexpr const char* kClientPause = "hbase.client.pause";

  static constexpr const uint64_t kDefaultClientPause = 100;

  /**
   * Parameter name for maximum retries, used as maximum for all retryable
   * operations such as fetching of the root region from root region server,
   * getting a cell's value, starting a row update, etc.
   */
  static constexpr const char* kClientRetriesNumber = "hbase.client.retries.number";

  static constexpr const uint32_t kDefaultClientRetriesNumber = 35;

  /**
    * Configure the number of failures after which the client will start logging. A few failures
    * is fine: region moved, then is not opened, then is overloaded. We try to have an acceptable
    * heuristic for the number of errors we don't log. 9 was chosen because we wait for 1s at
    * this stage.
    */
  static constexpr const char* kStartLogErrorsAfterCount = "hbase.client.start.log.errors.counter";
  static constexpr const uint32_t kDefaultStartLogErrorsAfterCount = 9;

  /** The client scanner timeout period in milliseconds. */
  static constexpr const char* kClientScannerTimeoutPeriod = "hbase.client.scanner.timeout.period";

  static constexpr const uint32_t kDefaultClientScannerTimeoutPeriod = 60000;

  /**
   * Parameter name to set the default scanner caching for all clients.
   */
  static constexpr const char* kClientScannerCaching = "hbase.client.scanner.caching";

  static constexpr const uint32_t kDefaultClientScannerCaching = INT_MAX;

  /**
   * Parameter name for maximum number of bytes returned when calling a scanner's next method.
   * Controlled by the client.
   */
  static constexpr const char* kClientScannerMaxResultsSize =
      "hbase.client.scanner.max.result.size";

  /**
   * Maximum number of bytes returned when calling a scanner's next method.
   * Note that when a single row is larger than this limit the row is still
   * returned completely.
   *
   * The default value is 2MB.
   */
  static constexpr const uint64_t kDefaultClientScannerMaxResultsSize = 2 * 1024 * 1024;

  std::chrono::nanoseconds connect_timeout_;
  std::chrono::nanoseconds meta_operation_timeout_;
  std::chrono::nanoseconds operation_timeout_;
  std::chrono::nanoseconds rpc_timeout_;
  std::chrono::nanoseconds read_rpc_timeout_;
  std::chrono::nanoseconds write_rpc_timeout_;
  std::chrono::nanoseconds pause_;
  uint32_t max_retries_;
  uint32_t start_log_errors_count_;
  std::chrono::nanoseconds scan_timeout_;
  uint32_t scanner_caching_;
  uint64_t scanner_max_result_size_;

  static std::chrono::nanoseconds ToNanos(const uint64_t& millis) {
    return std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::milliseconds(millis));
  }

  static uint64_t ToMillis(const std::chrono::nanoseconds& nanos) {
    return std::chrono::duration_cast<std::chrono::milliseconds>(nanos).count();
  }
};

}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/keyvalue-codec.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <folly/io/Cursor.h>
#include <folly/io/IOBuf.h>
#include <memory>

#include "hbase/client/cell.h"
#include "hbase/serde/codec.h"

namespace hbase {

/**
 * @brief Class for parsing sequence of Cells based on org.apache.hadoop.hbase.KeyValueCodec.java
 *
 * KeyValueCodec implements CellScanner interface. Sequence of cells are obtained from cell_block.
 * We have CreateEncoder and CreateDecoder public methods which will return Encoder/Decoder
 * instances which will be used to obtain individual cells in cell_block.
 * Usage:-
 * 1) Cell Decoding:-
 * unique_ptr<CellScanner> cell_scanner = KeyValueCodec::CreateDecoder(cell_block, cb_start_offset,
 *    cb_length);
 * while (cell_scanner->Advance()) {
 *  auto current_cell = cell_scanner->Current
 * }
 */
class KeyValueCodec : public Codec {
 public:
  /**
        * Constructor
        */
  KeyValueCodec() {}

  std::unique_ptr<Codec::Encoder> CreateEncoder() override { return std::make_unique<KVEncoder>(); }
  std::unique_ptr<Codec::Decoder> CreateDecoder(std::unique_ptr<folly::IOBuf> cell_block,
                                                uint32_t offset, uint32_t length) override {
    return std::make_unique<KVDecoder>(std::move(cell_block), offset, length);
  }

  /** @brief returns the java class name corresponding to this Codec implementation */
  virtual const char* java_class_name() const override { return kJavaClassName; }

  static constexpr const char* kJavaClassName = "org.apache.hadoop.hbase.codec.KeyValueCodec";

 private:
  class KVEncoder : public Codec::Encoder {
   public:
    KVEncoder() {}

    void Write(const Cell& cell) {
      // TODO: Encode Cells using KeyValueCodec wire format
    }

    void Flush() {}
  };

  class KVDecoder : public Codec::Decoder {
   public:
    KVDecoder(std::unique_ptr<folly::IOBuf> cell_block, uint32_t cell_block_start_offset,
              uint32_t cell_block_length);
    ~KVDecoder();

    /**
     * @brief Overridden from CellScanner. This method parses cell_block and stores the current in
     * current_cell_. Current cell can be obtained using cell_scanner.Current();
     */
    bool Advance();

    /**
     * @brief returns the current cell
     */
    const std::shared_ptr<Cell> Current() const { return current_cell_; }

    /**
     * @brief returns the total length of cell_meta_block
     */
    uint32_t CellBlockLength() const;

   private:
    std::shared_ptr<Cell> Decode(folly::io::Cursor& cursor);

    /**
     * Size of boolean in bytes
     */
    const int kHBaseSizeOfBoolean_ = sizeof(uint8_t) / sizeof(uint8_t);

    /**
     * Size of byte in bytes
     */
    const uint8_t kHBaseSizeOfByte_ = kHBaseSizeOfBoolean_;

    /**
     * Size of int in bytes
     */
    const uint32_t kHBaseSizeOfInt_ = sizeof(uint32_t) / kHBaseSizeOfByte_;

    /**
     * Size of long in bytes
     */
    const uint64_t kHBaseSizeOfLong_ = sizeof(uint64_t) / kHBaseSizeOfByte_;

    /**
     * Size of Short in bytes
     */
    const uint16_t kHBaseSizeOfShort_ = sizeof(uint16_t) / kHBaseSizeOfByte_;

    const uint32_t kHBaseSizeOfKeyLength_ = kHBaseSizeOfInt_;
    const uint32_t kHBaseSizeOfValueLength_ = kHBaseSizeOfInt_;
    const uint16_t kHBaseSizeOfRowLength_ = kHBaseSizeOfShort_;
    const uint8_t kHBaseSizeOfFamilyLength_ = kHBaseSizeOfByte_;
    const uint64_t kHBaseSizeOfTimestamp_ = kHBaseSizeOfLong_;
    const uint8_t kHBaseSizeOfKeyType_ = kHBaseSizeOfByte_;
    const uint32_t kHBaseSizeOfTimestampAndKey_ = kHBaseSizeOfTimestamp_ + kHBaseSizeOfKeyType_;
    const uint32_t kHBaseSizeOfKeyInfrastructure_ =
        kHBaseSizeOfRowLength_ + kHBaseSizeOfFamilyLength_ + kHBaseSizeOfTimestampAndKey_;
    const uint32_t kHBaseSizeOfKeyValueInfrastructure_ =
        kHBaseSizeOfKeyLength_ + kHBaseSizeOfValueLength_;

    std::unique_ptr<folly::IOBuf> cell_block_ = nullptr;
    uint32_t offset_ = 0;
    uint32_t length_ = 0;
    uint32_t cur_pos_ = 0;
    bool end_of_cell_block_ = false;

    std::shared_ptr<Cell> current_cell_ = nullptr;
  };
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/time-range.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <cstdint>

namespace hbase {
class TimeRange {
 public:
  /**
   * @brief  Default constructor. Represents interval [0,
   * std::numeric_limits<int64_t>::max())
   * (allTime)
   */
  TimeRange();
  TimeRange(const TimeRange &tr);
  TimeRange &operator=(const TimeRange &tr);
  /**
   * @brief Represents interval [minStamp, std::numeric_limits<int64_t>::max())
   * @param minStamp the minimum timestamp value, inclusive
   */
  explicit TimeRange(int64_t min_timestamp);
  /**
   * @brief Represents interval [minStamp, maxStamp)
   * @param minStamp the minimum timestamp, inclusive
   * @param maxStamp the maximum timestamp, exclusive
   * @throws std::runtime_error if min_timestamp < 0 or max_timestamp < 0 or
   * max_timestamp < min_timestamp
   */
  TimeRange(int64_t min_timestamp, int64_t max_timestamp);
  int64_t MinTimeStamp() const;
  int64_t MaxTimeStamp() const;
  bool IsAllTime() const;
  ~TimeRange();

 private:
  int64_t min_timestamp_;
  int64_t max_timestamp_;
  bool all_time_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/hbase-rpc-controller.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/ExceptionWrapper.h>
#include <google/protobuf/service.h>
#include <chrono>
#include <string>

namespace hbase {

class HBaseRpcController : public google::protobuf::RpcController {
 public:
  HBaseRpcController() {}
  virtual ~HBaseRpcController() = default;

  void set_call_timeout(const std::chrono::milliseconds& call_timeout) {
    // TODO:
  }

  void Reset() override {}

  bool Failed() const override { return false; }

  folly::exception_wrapper exception() { return exception_; }

  void set_exception(const folly::exception_wrapper& exception) { exception_ = exception; }

  std::string ErrorText() const override { return ""; }

  void StartCancel() override {}

  void SetFailed(const std::string& reason) override {}

  bool IsCanceled() const override { return false; }

  void NotifyOnCancel(google::protobuf::Closure* callback) override {}

 private:
  folly::exception_wrapper exception_;
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/serde/codec.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <folly/io/IOBuf.h>
#include <memory>

#include "hbase/serde/cell-outputstream.h"
#include "hbase/serde/cell-scanner.h"

namespace hbase {

/**
 * @brief Encoder / Decoder for Cells.
 */
class Codec {
 public:
  virtual ~Codec() {}

  class Encoder : public CellOutputStream {};

  class Decoder : public CellScanner {};

  virtual std::unique_ptr<Encoder> CreateEncoder() = 0;
  virtual std::unique_ptr<Decoder> CreateDecoder(std::unique_ptr<folly::IOBuf> cell_block,
                                                 uint32_t cell_block_start_offset,
                                                 uint32_t cell_block_length) = 0;

  /** @brief returns the java class name corresponding to this Codec implementation */
  virtual const char* java_class_name() const = 0;
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/client.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <memory>
#include <string>

#include "hbase/connection/rpc-client.h"
#include "hbase/client/async-connection.h"
#include "hbase/client/configuration.h"

#include "hbase/client/table.h"
#include "hbase/serde/table-name.h"

namespace hbase {

class Table;
/**
 * Client.
 *
 * This is the class that provides access to an HBase cluster.
 * It is thread safe and does connection pooling. Current recommendations are to
 * have only one Client per cluster around.
 */
class Client {
 public:
  /**
   * @brief Create a new client.
   * @param quorum_spec Where to connect to get Zookeeper bootstrap information.
   */
  Client();
  explicit Client(const Configuration& conf);
  ~Client() = default;

  /**
   * @brief Retrieve a Table implementation for accessing a table.
   * @param - table_name
   */
  std::unique_ptr<::hbase::Table> Table(const pb::TableName& table_name);

  /**
   * @brief Close the Client connection.
   */
  void Close();

  /**
   * @brief Internal. DO NOT USE.
   */
  std::shared_ptr<AsyncConnectionImpl> async_connection() { return async_connection_; }

 private:
  /** Data */
  std::shared_ptr<AsyncConnectionImpl> async_connection_;

 private:
  /** Methods */
  void Init(const Configuration& conf);
};

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/async-connection.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/async-connection.h"
#include "hbase/client/async-rpc-retrying-caller-factory.h"

namespace hbase {

void AsyncConnectionImpl::Init() {
  connection_conf_ = std::make_shared<hbase::ConnectionConfiguration>(*conf_);
  // start thread pools
  auto io_threads = conf_->GetInt(kClientIoThreadPoolSize, sysconf(_SC_NPROCESSORS_ONLN));
  auto cpu_threads = conf_->GetInt(kClientCpuThreadPoolSize, 2 * sysconf(_SC_NPROCESSORS_ONLN));
  cpu_executor_ = std::make_shared<wangle::CPUThreadPoolExecutor>(cpu_threads);
  io_executor_ = std::make_shared<wangle::IOThreadPoolExecutor>(io_threads);
  /*
   * We need a retry_executor for a thread pool of size 1 due to a possible bug in wangle/folly.
   * Otherwise, Assertion 'isInEventBaseThread()' always fails. See the comments
   * in async-rpc-retrying-caller.cc.
   */
  retry_executor_ = std::make_shared<wangle::IOThreadPoolExecutor>(1);
  retry_timer_ = folly::HHWheelTimer::newTimer(retry_executor_->getEventBase());

  std::shared_ptr<Codec> codec = nullptr;
  if (conf_->Get(kRpcCodec, hbase::KeyValueCodec::kJavaClassName) ==
      std::string(KeyValueCodec::kJavaClassName)) {
    codec = std::make_shared<hbase::KeyValueCodec>();
  } else {
    LOG(WARNING) << "Not using RPC Cell Codec";
  }
  rpc_client_ = std::make_shared<hbase::RpcClient>(io_executor_, cpu_executor_, codec, conf_,
                                                   connection_conf_->connect_timeout());
  location_cache_ = std::make_shared<hbase::LocationCache>(conf_, io_executor_, cpu_executor_,
                                                           rpc_client_->connection_pool());
  caller_factory_ =
      std::make_shared<AsyncRpcRetryingCallerFactory>(shared_from_this(), retry_timer_);
}

// We can't have the threads continue running after everything is done
// that leads to an error.
AsyncConnectionImpl::~AsyncConnectionImpl() { Close(); }

void AsyncConnectionImpl::Close() {
  if (is_closed_) return;

  cpu_executor_->stop();
  io_executor_->stop();
  retry_executor_->stop();
  retry_timer_->destroy();
  if (rpc_client_.get()) rpc_client_->Close();
  is_closed_ = true;
}

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/examples/simple-client.cc<|end_filename|>
/*
 * 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.
 *
 */

#include <folly/Logging.h>
#include <folly/Random.h>
#include <gflags/gflags.h>

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>

#include "hbase/connection/rpc-client.h"
#include "hbase/client/client.h"
#include "hbase/client/get.h"
#include "hbase/client/hbase-configuration-loader.h"
#include "hbase/client/put.h"
#include "hbase/client/scan.h"
#include "hbase/client/table.h"
#include "hbase/serde/server-name.h"
#include "hbase/serde/table-name.h"
#include "hbase/utils/time-util.h"

using hbase::Client;
using hbase::Configuration;
using hbase::Get;
using hbase::HBaseConfigurationLoader;
using hbase::Scan;
using hbase::Put;
using hbase::Result;
using hbase::Table;
using hbase::pb::TableName;
using hbase::pb::ServerName;
using hbase::TimeUtil;

DEFINE_string(table, "test_table", "What table to do the reads or writes");
DEFINE_string(row, "row_", "row prefix");
DEFINE_string(zookeeper, "localhost:2181", "What zk quorum to talk to");
DEFINE_string(conf, "", "Conf directory to read the config from (optional)");
DEFINE_uint64(num_rows, 10000, "How many rows to write and read");
DEFINE_uint64(batch_num_rows, 10000, "How many rows batch for multi-gets and multi-puts");
DEFINE_uint64(report_num_rows, 10000, "How frequent we should report the progress");
DEFINE_bool(puts, true, "Whether to perform puts");
DEFINE_bool(gets, true, "Whether to perform gets");
DEFINE_bool(multigets, true, "Whether to perform multi-gets");
DEFINE_bool(scans, true, "Whether to perform scans");
DEFINE_bool(display_results, false, "Whether to display the Results from Gets");
DEFINE_int32(threads, 6, "How many cpu threads");

std::unique_ptr<Put> MakePut(const std::string &row) {
  auto put = std::make_unique<Put>(row);
  put->AddColumn("f", "q", row);
  return std::move(put);
}

std::string Row(const std::string &prefix, uint64_t i) {
  auto suf = folly::to<std::string>(i);
  return prefix + suf;
}

void ValidateResult(const Result &result, const std::string &row) {
  CHECK(!result.IsEmpty());
  CHECK_EQ(result.Row(), row);
  CHECK_EQ(result.Size(), 1);
  CHECK_EQ(result.Value("f", "q").value(), row);
}

int main(int argc, char *argv[]) {
  gflags::SetUsageMessage("Simple client to get a single row from HBase on the comamnd line");
  gflags::ParseCommandLineFlags(&argc, &argv, true);
  google::InitGoogleLogging(argv[0]);
  google::InstallFailureSignalHandler();
  FLAGS_logtostderr = 1;
  FLAGS_stderrthreshold = 1;

  std::shared_ptr<Configuration> conf = nullptr;
  if (FLAGS_conf == "") {
    // Configuration
    conf = std::make_shared<Configuration>();
    conf->Set("hbase.zookeeper.quorum", FLAGS_zookeeper);
    conf->SetInt("hbase.client.cpu.thread.pool.size", FLAGS_threads);
  } else {
    setenv("HBASE_CONF", FLAGS_conf.c_str(), 1);
    hbase::HBaseConfigurationLoader loader;
    conf = std::make_shared<Configuration>(loader.LoadDefaultResources().value());
  }

  auto row = FLAGS_row;

  auto tn = std::make_shared<TableName>(folly::to<TableName>(FLAGS_table));
  auto num_puts = FLAGS_num_rows;

  auto client = std::make_unique<Client>(*conf);
  auto table = client->Table(*tn);

  auto start_ns = TimeUtil::GetNowNanos();

  // Do the Put requests
  if (FLAGS_puts) {
    LOG(INFO) << "Sending put requests";
    for (uint64_t i = 0; i < num_puts; i++) {
      table->Put(*MakePut(Row(FLAGS_row, i)));
      if (i != 0 && i % FLAGS_report_num_rows == 0) {
        LOG(INFO) << "Sent  " << i << " Put requests in " << TimeUtil::ElapsedMillis(start_ns)
                  << " ms.";
      }
    }

    LOG(INFO) << "Successfully sent  " << num_puts << " Put requests in "
              << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }

  // Do the Get requests
  if (FLAGS_gets) {
    LOG(INFO) << "Sending get requests";
    start_ns = TimeUtil::GetNowNanos();
    for (uint64_t i = 0; i < num_puts; i++) {
      auto row = Row(FLAGS_row, i);
      auto result = table->Get(Get{row});
      if (FLAGS_display_results) {
        LOG(INFO) << result->DebugString();
      } else if (i != 0 && i % FLAGS_report_num_rows == 0) {
        LOG(INFO) << "Sent  " << i << " Get requests in " << TimeUtil::ElapsedMillis(start_ns)
                  << " ms.";
      }
      ValidateResult(*result, row);
    }

    LOG(INFO) << "Successfully sent  " << num_puts << " Get requests in "
              << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }

  // Do the Multi-Gets
  if (FLAGS_multigets) {
    LOG(INFO) << "Sending multi-get requests";
    start_ns = TimeUtil::GetNowNanos();
    std::vector<hbase::Get> gets;

    for (uint64_t i = 0; i < num_puts;) {
      gets.clear();
      // accumulate batch_num_rows at a time
      for (uint64_t j = 0; j < FLAGS_batch_num_rows && i < num_puts; ++j) {
        hbase::Get get(Row(FLAGS_row, i));
        gets.push_back(get);
        i++;
      }
      auto results = table->Get(gets);

      if (FLAGS_display_results) {
        for (const auto &result : results) LOG(INFO) << result->DebugString();
      } else if (i != 0 && i % FLAGS_report_num_rows == 0) {
        LOG(INFO) << "Sent  " << i << " Multi-Get requests in " << TimeUtil::ElapsedMillis(start_ns)
                  << " ms.";
      }
    }

    LOG(INFO) << "Successfully sent  " << num_puts << " Multi-Get requests in "
              << TimeUtil::ElapsedMillis(start_ns) << " ms.";
  }

  // Do the Scan
  if (FLAGS_scans) {
    LOG(INFO) << "Starting scanner";
    start_ns = TimeUtil::GetNowNanos();
    Scan scan{};
    auto scanner = table->Scan(scan);

    uint64_t i = 0;
    auto r = scanner->Next();
    while (r != nullptr) {
      if (FLAGS_display_results) {
        LOG(INFO) << r->DebugString();
      }
      r = scanner->Next();
      i++;
      if (!FLAGS_display_results && i != 0 && i % FLAGS_report_num_rows == 0) {
        LOG(INFO) << "Scan iterated over " << i << " results " << TimeUtil::ElapsedMillis(start_ns)
                  << " ms.";
      }
    }

    LOG(INFO) << "Successfully iterated over  " << i << " Scan results in "
              << TimeUtil::ElapsedMillis(start_ns) << " ms.";
    scanner->Close();
  }

  table->Close();
  client->Close();

  return 0;
}


<|start_filename|>hbase-native-client/include/hbase/client/raw-async-table.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/Unit.h>
#include <folly/futures/Future.h>
#include <chrono>
#include <memory>
#include <string>
#include <vector>

#include "hbase/client/async-client-scanner.h"
#include "hbase/client/async-connection.h"
#include "hbase/client/async-rpc-retrying-caller-factory.h"
#include "hbase/client/async-rpc-retrying-caller.h"
#include "hbase/client/connection-configuration.h"
#include "hbase/client/delete.h"
#include "hbase/client/get.h"
#include "hbase/client/increment.h"
#include "hbase/client/put.h"
#include "hbase/client/result.h"
#include "hbase/client/scan.h"

namespace hbase {

/**
 * A low level asynchronous table that should not be used by user applications.The implementation
 * is required to be thread safe.
 */
class RawAsyncTable {
 public:
  RawAsyncTable(std::shared_ptr<pb::TableName> table_name,
                std::shared_ptr<AsyncConnection> connection)
      : connection_(connection),
        connection_conf_(connection->connection_conf()),
        table_name_(table_name),
        rpc_client_(connection->rpc_client()) {
    default_scanner_caching_ = connection_conf_->scanner_caching();
    default_scanner_max_result_size_ = connection_conf_->scanner_max_result_size();
  }
  virtual ~RawAsyncTable() = default;

  folly::Future<std::shared_ptr<Result>> Get(const hbase::Get& get);

  folly::Future<folly::Unit> Delete(const hbase::Delete& del);

  folly::Future<std::shared_ptr<hbase::Result>> Append(const hbase::Append& append);

  folly::Future<std::shared_ptr<hbase::Result>> Increment(const hbase::Increment& increment);

  folly::Future<folly::Unit> Put(const hbase::Put& put);

  folly::Future<bool> CheckAndPut(const std::string& row, const std::string& family,
                                  const std::string& qualifier, const std::string& value,
                                  const hbase::Put& put,
                                  const pb::CompareType& compare_op = pb::CompareType::EQUAL);

  folly::Future<bool> CheckAndDelete(const std::string& row, const std::string& family,
                                     const std::string& qualifier, const std::string& value,
                                     const hbase::Delete& del,
                                     const pb::CompareType& compare_op = pb::CompareType::EQUAL);

  void Scan(const hbase::Scan& scan, std::shared_ptr<RawScanResultConsumer> consumer);

  void Close() {}

  folly::Future<std::vector<folly::Try<std::shared_ptr<Result>>>> Get(
      const std::vector<hbase::Get>& gets);
  template <typename REQ, typename RESP>
  folly::Future<std::vector<folly::Try<RESP>>> Batch(const std::vector<REQ>& rows,
                                                     std::chrono::nanoseconds timeout);
  folly::Future<std::vector<folly::Try<std::shared_ptr<Result>>>> Put(
      const std::vector<hbase::Put>& puts);

 private:
  /* Data */
  std::shared_ptr<AsyncConnection> connection_;
  std::shared_ptr<ConnectionConfiguration> connection_conf_;
  std::shared_ptr<pb::TableName> table_name_;
  std::shared_ptr<RpcClient> rpc_client_;
  int32_t default_scanner_caching_;
  int64_t default_scanner_max_result_size_;

  /* Methods */
  template <typename REQ, typename PREQ, typename PRESP, typename RESP>
  folly::Future<RESP> Call(
      std::shared_ptr<RpcClient> rpc_client, std::shared_ptr<HBaseRpcController> controller,
      std::shared_ptr<RegionLocation> loc, const REQ& req,
      const ReqConverter<std::unique_ptr<PREQ>, REQ, std::string> req_converter,
      const RespConverter<RESP, PRESP> resp_converter);

  template <typename RESP>
  std::shared_ptr<SingleRequestCallerBuilder<RESP>> CreateCallerBuilder(
      std::string row, std::chrono::nanoseconds rpc_timeout);

  std::shared_ptr<hbase::Scan> SetDefaultScanConfig(const hbase::Scan& scan);
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/utils/bytes-util.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/utils/bytes-util.h"

#include <boost/predef.h>
#include <glog/logging.h>

#include <memory>
#include <string>

namespace hbase {

constexpr char BytesUtil::kHexChars[];

std::string BytesUtil::ToString(int64_t val) {
  std::string res;
#if BOOST_ENDIAN_BIG_BYTE || BOOST_ENDIAN_BIG_WORD
  for (int i = 7; i > 0; i--) {
    res += (int8_t)(val & 0xffu);
    val = val >> 8;
  }
  res += (int8_t)val;
#else
  int64_t mask = 0xff00000000000000u;
  for (int i = 56; i >= 1; i -= 8) {
    auto num = ((val & mask) >> i);
    res += num;
    mask = mask >> 8;
  }
  res += (val & 0xff);
#endif
  return res;
}

int64_t BytesUtil::ToInt64(std::string str) {
  if (str.length() < 8) {
    throw std::runtime_error("There are not enough bytes. Expected: 8, actual: " + str.length());
  }
  const unsigned char *bytes = reinterpret_cast<unsigned char *>(const_cast<char *>(str.c_str()));
  int64_t l = 0;
  for (int i = 0; i < 8; i++) {
    l <<= 8;
    l ^= bytes[i];
  }
  return l;
}

std::string BytesUtil::ToStringBinary(const std::string &b, size_t off, size_t len) {
  std::string result;
  // Just in case we are passed a 'len' that is > buffer length...
  if (off >= b.size()) {
    return result;
  }
  if (off + len > b.size()) {
    len = b.size() - off;
  }
  for (size_t i = off; i < off + len; ++i) {
    int32_t ch = b[i] & 0xFF;
    if (ch >= ' ' && ch <= '~' && ch != '\\') {
      result += ch;
    } else {
      result += "\\x";
      result += kHexChars[ch / 0x10];
      result += kHexChars[ch % 0x10];
    }
  }
  return result;
}

} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/client/cell.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/cell.h"
#include <climits>
#include <limits>
#include <stdexcept>

#include "folly/Conv.h"
#include "hbase/utils/bytes-util.h"

namespace hbase {

Cell::Cell(const std::string &row, const std::string &family, const std::string &qualifier,
           const int64_t timestamp, const std::string &value, const hbase::CellType &cell_type)
    : row_(row),
      family_(family),
      qualifier_(qualifier),
      timestamp_(timestamp),
      cell_type_(cell_type),
      value_(value),
      sequence_id_(0) {
  if (0 == row.size()) throw std::runtime_error("Row size should be greater than 0");

  if (0 >= timestamp) throw std::runtime_error("Timestamp should be greater than 0");
}

Cell::Cell(const Cell &cell)
    : row_(cell.row_),
      family_(cell.family_),
      qualifier_(cell.qualifier_),
      timestamp_(cell.timestamp_),
      cell_type_(cell.cell_type_),
      value_(cell.value_),
      sequence_id_(cell.sequence_id_) {}

Cell &Cell::operator=(const Cell &cell) {
  row_ = cell.row_;
  family_ = cell.family_;
  qualifier_ = cell.qualifier_;
  timestamp_ = cell.timestamp_;
  cell_type_ = cell.cell_type_;
  value_ = cell.value_;
  sequence_id_ = cell.sequence_id_;

  return *this;
}

Cell::~Cell() {}

const std::string &Cell::Row() const { return row_; }

const std::string &Cell::Family() const { return family_; }

const std::string &Cell::Qualifier() const { return qualifier_; }

int64_t Cell::Timestamp() const { return timestamp_; }

const std::string &Cell::Value() const { return value_; }

hbase::CellType Cell::Type() const { return cell_type_; }

int64_t Cell::SequenceId() const { return sequence_id_; }

std::string Cell::DebugString() const {
  std::string timestamp_str;
  if (timestamp_ == std::numeric_limits<int64_t>::max()) {
    timestamp_str = "LATEST_TIMESTAMP";
  } else {
    timestamp_str = folly::to<std::string>(timestamp_);
  }

  return BytesUtil::ToStringBinary(row_) + "/" + BytesUtil::ToStringBinary(family_) +
         (family_.empty() ? "" : ":") + BytesUtil::ToStringBinary(qualifier_) + "/" +
         timestamp_str + "/" + TypeToString(cell_type_) + "/vlen=" +
         folly::to<std::string>(value_.size()) + "/seqid=" + folly::to<std::string>(sequence_id_);
}

const char *Cell::TypeToString(CellType type) {
  switch (type) {
    case CellType::MINIMUM:
      return "MINIMUM";
    case CellType::PUT:
      return "PUT";
    case CellType::DELETE:
      return "DELETE";
    case CellType::DELETE_COLUMN:
      return "DELETE_COLUMN";
    case CellType::DELETE_FAMILY:
      return "DELETE_FAMILY";
    case CellType::MAXIMUM:
      return "MAXIMUM";
    default:
      return "UNKNOWN";
  }
}

size_t Cell::EstimatedSize() const {
  size_t s = sizeof(Cell);
  s += row_.capacity();
  s += family_.capacity();
  s += qualifier_.capacity();
  s += value_.capacity();
  return s;
}

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/result-scanner.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <functional>
#include <iterator>
#include <map>
#include <memory>
#include <string>
#include <vector>

#include "hbase/client/cell.h"
#include "hbase/client/result.h"

namespace hbase {

/**
 * Interface for client-side scanning. Use Table to obtain instances.
 */
class ResultScanner {
  // TODO: should we implement forward iterators?

 public:
  virtual ~ResultScanner() {}

  virtual void Close() = 0;

  virtual std::shared_ptr<Result> Next() = 0;
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/delete.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <cstdint>
#include <map>
#include <memory>
#include <string>
#include <vector>
#include "hbase/client/cell.h"
#include "hbase/client/mutation.h"

namespace hbase {

class Delete : public Mutation {
 public:
  /**
   * Constructors
   */
  /*
   * If no further operations are done, this will delete everything
   * associated with the specified row (all versions of all columns in all
   * families), with timestamp from current point in time to the past.
   * Cells defining timestamp for a future point in time
   * (timestamp > current time) will not be deleted.
   */
  explicit Delete(const std::string& row) : Mutation(row) {}

  Delete(const std::string& row, int64_t timestamp) : Mutation(row, timestamp) {}
  Delete(const Delete& cdelete) : Mutation(cdelete) {}
  Delete& operator=(const Delete& cdelete) {
    Mutation::operator=(cdelete);
    return *this;
  }

  ~Delete() = default;

  /**
   *  @brief Add the specified column to this Delete operation.
   *  @param family family name
   *  @param qualifier column qualifier
   */
  Delete& AddColumn(const std::string& family, const std::string& qualifier);

  /**
   *  @brief Add the specified column to this Delete operation.
   *  @param family family name
   *  @param qualifier column qualifier
   *  @param timestamp version timestamp
   */
  Delete& AddColumn(const std::string& family, const std::string& qualifier, int64_t timestamp);

  /**
   *  @brief Deletes all versions of the specified column
   *  @param family family name
   *  @param qualifier column qualifier
   */
  Delete& AddColumns(const std::string& family, const std::string& qualifier);
  /**
   *  @brief Deletes all versions of the specified column with a timestamp less than
   * or equal to the specified timestamp
   *  @param family family name
   *  @param qualifier column qualifier
   *  @param timestamp version timestamp
   */
  Delete& AddColumns(const std::string& family, const std::string& qualifier, int64_t timestamp);
  /**
   *  @brief Add the specified family to this Delete operation.
   *  @param family family name
   */
  Delete& AddFamily(const std::string& family);

  /**
   *  @brief Deletes all columns of the specified family with a timestamp less than
   * or equal to the specified timestamp
   *  @param family family name
   *  @param timestamp version timestamp
   */
  Delete& AddFamily(const std::string& family, int64_t timestamp);
  /**
   *  @brief Deletes all columns of the specified family with a timestamp
   *   equal to the specified timestamp
   *  @param family family name
   *  @param timestamp version timestamp
   */
  Delete& AddFamilyVersion(const std::string& family, int64_t timestamp);
  /**
   * Advanced use only.
   * Add an existing delete marker to this Delete object.
   */
  Delete& Add(std::unique_ptr<Cell> cell);
};

}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/location-cache.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/ExceptionWrapper.h>
#include <folly/Executor.h>
#include <folly/SharedMutex.h>
#include <folly/futures/Future.h>
#include <folly/futures/SharedPromise.h>
#include <wangle/concurrent/CPUThreadPoolExecutor.h>
#include <wangle/concurrent/IOThreadPoolExecutor.h>
#include <zookeeper/zookeeper.h>

#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>

#include "hbase/connection/connection-pool.h"
#include "hbase/client/async-region-locator.h"
#include "hbase/client/configuration.h"
#include "hbase/client/meta-utils.h"
#include "hbase/client/region-location.h"
#include "hbase/client/zk-util.h"
#include "hbase/serde/table-name.h"

namespace hbase {
// Forward
class Request;
class Response;
namespace pb {
class ServerName;
class TableName;
}

/** Equals function for TableName (uses namespace and table name) */
struct TableNameEquals {
  /** equals */
  bool operator()(const hbase::pb::TableName &lht, const hbase::pb::TableName &rht) const {
    return lht.namespace_() == rht.namespace_() && lht.qualifier() == rht.qualifier();
  }
};

/** Hash for TableName. */
struct TableNameHash {
  /** hash */
  std::size_t operator()(hbase::pb::TableName const &t) const {
    std::size_t h = 0;
    boost::hash_combine(h, t.namespace_());
    boost::hash_combine(h, t.qualifier());
    return h;
  }
};

// typedefs for location cache
typedef std::map<std::string, std::shared_ptr<RegionLocation>> PerTableLocationMap;
typedef std::unordered_map<hbase::pb::TableName, std::shared_ptr<PerTableLocationMap>,
                           TableNameHash, TableNameEquals>
    RegionLocationMap;

/**
 * Class that can look up and cache locations.
 */
class LocationCache : public AsyncRegionLocator {
 public:
  /**
   * Constructor.
   * @param conf Configuration instance to fetch Zookeeper Quorum and Zookeeper Znode.
   * @param cpu_executor executor used to run non network IO based
   * continuations.
   * @param io_executor executor used to talk to the network
   */
  LocationCache(std::shared_ptr<hbase::Configuration> conf,
                std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
                std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
                std::shared_ptr<ConnectionPool> cp);
  /**
   * Destructor.
   * This will clean up the zookeeper connections.
   */
  ~LocationCache();

  /**
   * Where is meta hosted.
   *
   * TODO: This should be a RegionLocation.
   */
  folly::Future<hbase::pb::ServerName> LocateMeta();

  /**
   * Go read meta and find out where a region is located. Most users should
   * never call this method directly and should use LocateRegion() instead.
   *
   * @param tn Table name of the table to look up. This object must live until
   * after the future is returned
   *
   * @param row of the table to look up. This object must live until after the
   * future is returned
   */
  folly::Future<std::shared_ptr<RegionLocation>> LocateFromMeta(const hbase::pb::TableName &tn,
                                                                const std::string &row);

  /**
   * The only method clients should use for meta lookups. If corresponding
   * location is cached, it's returned from the cache, otherwise lookup
   * in meta table is done, location is cached and then returned.
   * It's expected that tiny fraction of invocations incurs meta scan.
   * This method is to look up non-meta regions; use LocateMeta() to get the
   * location of hbase:meta region.
   *
   * @param tn Table name of the table to look up. This object must live until
   * after the future is returned
   *
   * @param row of the table to look up. This object must live until after the
   * future is returned
   */
  folly::Future<std::shared_ptr<RegionLocation>> LocateRegion(
      const hbase::pb::TableName &tn, const std::string &row,
      const RegionLocateType locate_type = RegionLocateType::kCurrent,
      const int64_t locate_ns = 0) override;

  /**
   * Remove the cached location of meta.
   */
  std::shared_ptr<folly::SharedPromise<hbase::pb::ServerName>> InvalidateMeta();

  /**
   * Return cached region location corresponding to this row,
   * nullptr if this location isn't cached.
   */
  std::shared_ptr<RegionLocation> GetCachedLocation(const hbase::pb::TableName &tn,
                                                    const std::string &row);

  /**
   * Add non-meta region location in the cache (location of meta itself
   * is cached separately).
   */
  void CacheLocation(const hbase::pb::TableName &tn, const std::shared_ptr<RegionLocation> loc);

  /**
   * Check if location corresponding to this row key is cached.
   */
  bool IsLocationCached(const hbase::pb::TableName &tn, const std::string &row);

  /**
   * Return cached location for all region of this table.
   */
  std::shared_ptr<PerTableLocationMap> GetTableLocations(const hbase::pb::TableName &tn);

  /**
   * Completely clear location cache.
   */
  void ClearCache();

  /**
   * Clear all cached locations for one table.
   */
  void ClearCachedLocations(const hbase::pb::TableName &tn);

  /**
   * Clear cached region location.
   */
  void ClearCachedLocation(const hbase::pb::TableName &tn, const std::string &row);

  /**
   * Update cached region location, possibly using the information from exception.
   */
  void UpdateCachedLocation(const RegionLocation &loc,
                            const folly::exception_wrapper &error) override;

  const std::string &zk_quorum() { return zk_quorum_; }

 private:
  void CloseZooKeeperConnection();
  void EnsureZooKeeperConnection();

 private:
  void RefreshMetaLocation();
  hbase::pb::ServerName ReadMetaLocation();
  std::shared_ptr<RegionLocation> CreateLocation(const Response &resp);
  std::shared_ptr<hbase::PerTableLocationMap> GetCachedTableLocations(
      const hbase::pb::TableName &tn);
  std::shared_ptr<hbase::PerTableLocationMap> GetNewTableLocations(const hbase::pb::TableName &tn);

  /* data */
  std::shared_ptr<hbase::Configuration> conf_;
  std::string zk_quorum_;
  std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor_;
  std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor_;
  std::shared_ptr<folly::SharedPromise<hbase::pb::ServerName>> meta_promise_;
  std::recursive_mutex meta_lock_;
  MetaUtil meta_util_;
  std::shared_ptr<ConnectionPool> cp_;

  // cached region locations
  RegionLocationMap cached_locations_;
  folly::SharedMutexWritePriority locations_lock_;

  // TODO: migrate this to a smart pointer with a deleter.
  zhandle_t *zk_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/append.cc<|end_filename|>


/*
 * 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.
 *
 */

#include "hbase/client/append.h"
#include <folly/Conv.h>
#include <algorithm>
#include <limits>
#include <stdexcept>
#include <utility>

namespace hbase {

/**
 *  @brief Append to the column from the specific family with the specified qualifier
 *  @param family family name
 *  @param qualifier column qualifier
 *  @param value value to append
 */
Append& Append::Add(const std::string& family, const std::string& qualifier,
                    const std::string& value) {
  family_map_[family].push_back(std::move(
      std::make_unique<Cell>(row_, family, qualifier, timestamp_, value, hbase::CellType::PUT)));
  return *this;
}
Append& Append::Add(std::unique_ptr<Cell> cell) {
  if (cell->Row() != row_) {
    throw std::runtime_error("The row in " + cell->DebugString() +
                             " doesn't match the original one " + row_);
  }

  family_map_[cell->Family()].push_back(std::move(cell));
  return *this;
}

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/get.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/get.h"
#include <algorithm>
#include <limits>
#include <stdexcept>

namespace hbase {

Get::~Get() {}

Get::Get(const std::string &row) : Row(row) {}

Get::Get(const Get &get) : Query(get) {
  row_ = get.row_;
  max_versions_ = get.max_versions_;
  cache_blocks_ = get.cache_blocks_;
  check_existence_only_ = get.check_existence_only_;
  consistency_ = get.consistency_;
  tr_.reset(new TimeRange(get.Timerange().MinTimeStamp(), get.Timerange().MaxTimeStamp()));
  family_map_.insert(get.family_map_.begin(), get.family_map_.end());
}

Get &Get::operator=(const Get &get) {
  Query::operator=(get);
  row_ = get.row_;
  max_versions_ = get.max_versions_;
  cache_blocks_ = get.cache_blocks_;
  check_existence_only_ = get.check_existence_only_;
  consistency_ = get.consistency_;
  tr_.reset(new TimeRange(get.Timerange().MinTimeStamp(), get.Timerange().MaxTimeStamp()));
  family_map_.insert(get.family_map_.begin(), get.family_map_.end());
  return *this;
}

Get &Get::AddFamily(const std::string &family) {
  const auto &it = family_map_.find(family);

  /**
   * Check if any qualifiers are already present or not.
   * Remove all existing qualifiers if the given family is already present in
   * the map
   */
  if (family_map_.end() != it) {
    it->second.clear();
  } else {
    family_map_[family];
  }
  return *this;
}

Get &Get::AddColumn(const std::string &family, const std::string &qualifier) {
  const auto &it = std::find(family_map_[family].begin(), family_map_[family].end(), qualifier);

  /**
   * Check if any qualifiers are already present or not.
   * Add only if qualifiers for a given family are not present
   */
  if (it == family_map_[family].end()) {
    family_map_[family].push_back(qualifier);
  }
  return *this;
}

hbase::pb::Consistency Get::Consistency() const { return consistency_; }

Get &Get::SetConsistency(hbase::pb::Consistency consistency) {
  consistency_ = consistency;
  return *this;
}

bool Get::HasFamilies() const { return !family_map_.empty(); }

const std::map<std::string, std::vector<std::string>> &Get::FamilyMap() const {
  return family_map_;
}

int Get::MaxVersions() const { return max_versions_; }

Get &Get::SetMaxVersions(int32_t max_versions) {
  if (0 == max_versions) throw std::runtime_error("max_versions must be positive");

  max_versions_ = max_versions;
  return *this;
}

bool Get::CacheBlocks() const { return cache_blocks_; }

Get &Get::SetCacheBlocks(bool cache_blocks) {
  cache_blocks_ = cache_blocks;
  return *this;
}

Get &Get::SetTimeRange(int64_t min_timestamp, int64_t max_timestamp) {
  tr_.reset(new TimeRange(min_timestamp, max_timestamp));
  return *this;
}

Get &Get::SetTimeStamp(int64_t timestamp) {
  tr_.reset(new TimeRange(timestamp, timestamp + 1));
  return *this;
}

const TimeRange &Get::Timerange() const { return *tr_; }

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/async-table-result-scanner.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/async-table-result-scanner.h"

#include <vector>

namespace hbase {
AsyncTableResultScanner::AsyncTableResultScanner(int64_t max_cache_size)
    : max_cache_size_(max_cache_size) {
  closed_ = false;
  cache_size_ = 0;
}

AsyncTableResultScanner::~AsyncTableResultScanner() { Close(); }

void AsyncTableResultScanner::Close() {
  std::unique_lock<std::mutex> mlock(mutex_);
  closed_ = true;
  while (!queue_.empty()) {
    queue_.pop();
  }
  cache_size_ = 0;
  if (resumer_ != nullptr) {
    resumer_->Resume();
  }
  cond_.notify_all();
}

std::shared_ptr<Result> AsyncTableResultScanner::Next() {
  VLOG(5) << "AsyncTableResultScanner: Next()";

  std::shared_ptr<Result> result = nullptr;
  std::shared_ptr<ScanResumer> local_resumer = nullptr;
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    while (queue_.empty()) {
      if (closed_) {
        return nullptr;
      }
      if (error_) {
        throw error_;
      }
      cond_.wait(mlock);
    }
    result = queue_.front();
    queue_.pop();

    cache_size_ -= EstimatedSizeWithSharedPtr(result);
    if (resumer_ != nullptr && cache_size_ <= max_cache_size_ / 2) {
      VLOG(1) << std::this_thread::get_id() << " resume scan prefetching";
      local_resumer = resumer_;
      resumer_ = nullptr;
    }
  }

  // Need to call ScanResumer::Resume() outside of the scope of the mutex. The reason is that
  // folly/wangle event loop might end up running the attached logic(.then()) at the Scan RPC
  // in the same event thread before returning from the previous call. This seems like the
  // wrong thing to do(™), but we cannot fix that now. Since the call back can end up calling
  // this::OnNext(), we should unlock the mutex.
  if (local_resumer != nullptr) {
    local_resumer->Resume();
  }
  return result;
}

void AsyncTableResultScanner::AddToCache(const std::vector<std::shared_ptr<Result>> &results) {
  VLOG(5) << "AsyncTableResultScanner: AddToCache()";
  for (const auto r : results) {
    queue_.push(r);
    cache_size_ += EstimatedSizeWithSharedPtr(r);
  }
}

template <typename T>
inline size_t AsyncTableResultScanner::EstimatedSizeWithSharedPtr(std::shared_ptr<T> t) {
  return t->EstimatedSize() + sizeof(std::shared_ptr<T>);
}

void AsyncTableResultScanner::OnNext(const std::vector<std::shared_ptr<Result>> &results,
                                     std::shared_ptr<ScanController> controller) {
  VLOG(5) << "AsyncTableResultScanner: OnNext()";
  {
    std::unique_lock<std::mutex> mlock(mutex_);
    if (closed_) {
      controller->Terminate();
      return;
    }
    AddToCache(results);

    if (cache_size_ >= max_cache_size_) {
      StopPrefetch(controller);
    }
  }
  cond_.notify_all();
}

void AsyncTableResultScanner::StopPrefetch(std::shared_ptr<ScanController> controller) {
  VLOG(1) << std::this_thread::get_id()
          << ": stop prefetching when scanning as the cache size " +
                 folly::to<std::string>(cache_size_) + " is greater than the max_cache_size " +
                 folly::to<std::string>(max_cache_size_);

  resumer_ = controller->Suspend();
  num_prefetch_stopped_++;
}

/**
 * Indicate that there is an heartbeat message but we have not cumulated enough cells to call
 * onNext.
 * <p>
 * This method give you a chance to terminate a slow scan operation.
 * @param controller used to suspend or terminate the scan. Notice that the {@code controller}
 *          instance is only valid within the scope of onHeartbeat method. You can only call its
 *          method in onHeartbeat, do NOT store it and call it later outside onHeartbeat.
 */
void AsyncTableResultScanner::OnHeartbeat(std::shared_ptr<ScanController> controller) {
  std::unique_lock<std::mutex> mlock(mutex_);
  if (closed_) {
    controller->Terminate();
  }
}

/**
 * Indicate that we hit an unrecoverable error and the scan operation is terminated.
 * <p>
 * We will not call {@link #onComplete()} after calling {@link #onError(Throwable)}.
 */
void AsyncTableResultScanner::OnError(const folly::exception_wrapper &error) {
  LOG(WARNING) << "Scanner received error" << error.what();
  std::unique_lock<std::mutex> mlock(mutex_);
  error_ = error;
  cond_.notify_all();
}

/**
 * Indicate that the scan operation is completed normally.
 */
void AsyncTableResultScanner::OnComplete() {
  std::unique_lock<std::mutex> mlock(mutex_);
  closed_ = true;
  cond_.notify_all();
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/row.h<|end_filename|>
/*
 * 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.
 *
 */

#include <limits>
#include <stdexcept>
#include <string>

#pragma once

namespace hbase {

class Row {
 public:
  Row() {}
  explicit Row(const std::string &row) : row_(row) { CheckRow(row_); }

  /**
   * @brief Returns the row for the Row interface.
   */
  const std::string &row() const { return row_; }
  virtual ~Row() {}

 private:
  /**
   * @brief Checks if the row for this Get operation is proper or not
   * @param row Row to check
   * @throws std::runtime_error if row is empty or greater than
   * MAX_ROW_LENGTH(i.e. std::numeric_limits<short>::max())
   */
  void CheckRow(const std::string &row) {
    const int16_t kMaxRowLength = std::numeric_limits<int16_t>::max();
    size_t row_length = row.size();
    if (0 == row_length) {
      throw std::runtime_error("Row length can't be 0");
    }
    if (row_length > kMaxRowLength) {
      throw std::runtime_error("Length of " + row + " is greater than max row size: " +
                               std::to_string(kMaxRowLength));
    }
  }

 protected:
  std::string row_ = "";
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/utils/time-util.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <chrono>
#include <string>

namespace hbase {

class TimeUtil {
 public:
  static inline int64_t ToMillis(const int64_t& nanos) {
    return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::nanoseconds(nanos))
        .count();
  }

  static inline std::chrono::milliseconds ToMillis(const std::chrono::nanoseconds& nanos) {
    return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::nanoseconds(nanos));
  }

  static inline std::chrono::nanoseconds ToNanos(const std::chrono::milliseconds& millis) {
    return std::chrono::duration_cast<std::chrono::nanoseconds>(millis);
  }

  static inline std::chrono::nanoseconds MillisToNanos(const int64_t& millis) {
    return std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::milliseconds(millis));
  }

  static inline std::chrono::nanoseconds SecondsToNanos(const int64_t& secs) {
    return std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::seconds(secs));
  }

  static inline std::string ToMillisStr(const std::chrono::nanoseconds& nanos) {
    return std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(nanos).count());
  }

  static inline int64_t GetNowNanos() {
    auto duration = std::chrono::high_resolution_clock::now().time_since_epoch();
    return std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
  }

  static inline int64_t ElapsedMillis(const int64_t& start_ns) {
    return std::chrono::duration_cast<std::chrono::milliseconds>(
               std::chrono::nanoseconds(GetNowNanos() - start_ns))
        .count();
  }

  static inline std::string ElapsedMillisStr(const int64_t& start_ns) {
    return std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
                              std::chrono::nanoseconds(GetNowNanos() - start_ns))
                              .count());
  }
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/query.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <memory>

#include "hbase/client/filter.h"

namespace hbase {

/**
 * Base class for read RPC calls (Get / Scan).
 */
class Query {
 public:
  Query() = default;
  Query(const Query &query) {
    // filter can be a custom subclass of Filter, so we do not do a deep copy here.
    filter_ = query.filter_;
  }

  Query &operator=(const Query &query) {
    filter_ = query.filter_;
    return *this;
  }

  virtual ~Query() {}

  void SetFilter(std::shared_ptr<Filter> filter) { filter_ = filter; }

  const std::shared_ptr<Filter> filter() const { return filter_; }

 protected:
  std::shared_ptr<Filter> filter_ = nullptr;
};

}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/exceptions/exception.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/ExceptionWrapper.h>
#include <folly/io/IOBuf.h>
#include <exception>
#include <memory>
#include <string>
#include <vector>

namespace hbase {

class ThrowableWithExtraContext {
 public:
  ThrowableWithExtraContext(folly::exception_wrapper cause, const int64_t& when)
      : cause_(cause), when_(when), extras_("") {}

  ThrowableWithExtraContext(folly::exception_wrapper cause, const int64_t& when,
                            const std::string& extras)
      : cause_(cause), when_(when), extras_(extras) {}

  virtual std::string ToString() {
    // TODO:
    // return new Date(this.when).toString() + ", " + extras + ", " + t.toString();
    return extras_ + ", " + cause_.what().toStdString();
  }

  virtual folly::exception_wrapper cause() { return cause_; }

 private:
  folly::exception_wrapper cause_;
  int64_t when_;
  std::string extras_;
};

class IOException : public std::logic_error {
 public:
  IOException() : logic_error(""), do_not_retry_(false) {}

  explicit IOException(const std::string& what) : logic_error(what), do_not_retry_(false) {}

  IOException(const std::string& what, bool do_not_retry)
      : logic_error(what), do_not_retry_(do_not_retry) {}

  IOException(const std::string& what, const folly::exception_wrapper& cause)
      : logic_error(what), cause_(cause), do_not_retry_(false) {}

  IOException(const std::string& what, folly::exception_wrapper cause, bool do_not_retry)
      : logic_error(what), cause_(cause), do_not_retry_(do_not_retry) {}

  virtual ~IOException() = default;

  virtual folly::exception_wrapper cause() const { return cause_; }

  bool do_not_retry() const { return do_not_retry_; }

  IOException* set_do_not_retry(bool value) {
    do_not_retry_ = value;
    return this;
  }

 private:
  folly::exception_wrapper cause_;
  // In case the exception is a RemoteException, do_not_retry information can come from
  // the PB field in the RPC response, or it can be deduced from the Java-exception
  // hierarchy in ExceptionUtil::ShouldRetry(). In case this is a client-side exception
  // raised from the C++ internals, set this field so that the retrying callers can
  // re-throw the exception without retrying.
  bool do_not_retry_;
};

class RetriesExhaustedException : public IOException {
 public:
  RetriesExhaustedException(const int& num_retries,
                            std::shared_ptr<std::vector<ThrowableWithExtraContext>> exceptions)
      : IOException(GetMessage(num_retries, exceptions),
                    exceptions->empty() ? folly::exception_wrapper{}
                                        : (*exceptions)[exceptions->size() - 1].cause()),
        num_retries_(num_retries) {}
  virtual ~RetriesExhaustedException() = default;

  int32_t num_retries() const { return num_retries_; }

 private:
  std::string GetMessage(const int& num_retries,
                         std::shared_ptr<std::vector<ThrowableWithExtraContext>> exceptions) {
    std::string buffer("Failed after attempts=");
    buffer.append(std::to_string(num_retries + 1));
    buffer.append(", exceptions:\n");
    for (auto it = exceptions->begin(); it != exceptions->end(); it++) {
      buffer.append(it->ToString());
      buffer.append("\n");
    }
    return buffer;
  }

 private:
  int32_t num_retries_;
};

class ConnectionException : public IOException {
 public:
  ConnectionException() {}

  ConnectionException(const std::string& what) : IOException(what) {}

  ConnectionException(const folly::exception_wrapper& cause) : IOException("", cause) {}

  ConnectionException(const std::string& what, const folly::exception_wrapper& cause)
      : IOException(what, cause) {}
};

class RemoteException : public IOException {
 public:
  RemoteException() : IOException(), port_(0) {}

  explicit RemoteException(const std::string& what) : IOException(what), port_(0) {}

  RemoteException(const std::string& what, folly::exception_wrapper cause)
      : IOException(what, cause), port_(0) {}

  virtual ~RemoteException() = default;

  std::string exception_class_name() const { return exception_class_name_; }

  RemoteException* set_exception_class_name(const std::string& value) {
    exception_class_name_ = value;
    return this;
  }

  std::string stack_trace() const { return stack_trace_; }

  RemoteException* set_stack_trace(const std::string& value) {
    stack_trace_ = value;
    return this;
  }

  std::string hostname() const { return hostname_; }

  RemoteException* set_hostname(const std::string& value) {
    hostname_ = value;
    return this;
  }

  int port() const { return port_; }

  RemoteException* set_port(int value) {
    port_ = value;
    return this;
  }

 private:
  std::string exception_class_name_;
  std::string stack_trace_;
  std::string hostname_;
  int port_;
};

/**
 * Raised from the client side if we cannot find the table (does not have anything to
 * do with the Java exception of the same name).
 */
class TableNotFoundException : public IOException {
 public:
  explicit TableNotFoundException(const std::string& table_name)
      : IOException("Table cannot be found:" + table_name, true) {}

  virtual ~TableNotFoundException() = default;
};

/**
 * List of known exceptions from Java side, and Java-specific exception logic
 */
class ExceptionUtil {
 private:
  // unknown scanner and sub-classes
  static constexpr const char* kUnknownScannerException =
      "org.apache.hadoop.hbase.UnknownScannerException";

  // not serving region and sub-classes
  static constexpr const char* kNotServingRegionException =
      "org.apache.hadoop.hbase.NotServingRegionException";
  static constexpr const char* kRegionInRecoveryException =
      "org.apache.hadoop.hbase.exceptions.RegionInRecoveryException";
  static constexpr const char* kRegionOpeningException =
      "org.apache.hadoop.hbase.exceptions.RegionOpeningException";
  static constexpr const char* kRegionMovedException =
      "org.apache.hadoop.hbase.exceptions.RegionMovedException";

  // Region server stopped and sub-classes
  static constexpr const char* kRegionServerStoppedException =
      "org.apache.hadoop.hbase.regionserver.RegionServerStoppedException";
  static constexpr const char* kRegionServerAbortedException =
      "org.apache.hadoop.hbase.regionserver.RegionServerAbortedException";

  // other scanner related exceptions
  static constexpr const char* kOutOfOrderScannerNextException =
      "org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException";
  static constexpr const char* kScannerResetException =
      "org.apache.hadoop.hbase.exceptions.ScannerResetException";

  // All other DoNotRetryIOExceptions
  static constexpr const char* kDoNotRetryIOException =
      "org.apache.hadoop.hbase.DoNotRetryIOException";
  static constexpr const char* kTableNotFoundException =
      "org.apache.hadoop.hbase.TableNotFoundException";
  static constexpr const char* kTableNotEnabledException =
      "org.apache.hadoop.hbase.TableNotEnabledException";
  static constexpr const char* kCoprocessorException =
      "org.apache.hadoop.hbase.coprocessor.CoprocessorException";
  static constexpr const char* kBypassCoprocessorException =
      "org.apache.hadoop.hbase.coprocessor.BypassCoprocessorException";
  static constexpr const char* kInvalidFamilyOperationException =
      "org.apache.hadoop.hbase.InvalidFamilyOperationException";
  static constexpr const char* kServerTooBusyException =
      "org.apache.hadoop.hbase.ipc.ServerTooBusyException";  // This should NOT be DNRIOE?
  static constexpr const char* kFailedSanityCheckException =
      "org.apache.hadoop.hbase.exceptions.FailedSanityCheckException";
  static constexpr const char* kCorruptHFileException =
      "org.apache.hadoop.hbase.io.hfile.CorruptHFileException";
  static constexpr const char* kLabelAlreadyExistsException =
      "org.apache.hadoop.hbase.security.visibility.LabelAlreadyExistsException";
  static constexpr const char* kFatalConnectionException =
      "org.apache.hadoop.hbase.ipc.FatalConnectionException";
  static constexpr const char* kUnsupportedCryptoException =
      "org.apache.hadoop.hbase.ipc.UnsupportedCryptoException";
  static constexpr const char* kUnsupportedCellCodecException =
      "org.apache.hadoop.hbase.ipc.UnsupportedCellCodecException";
  static constexpr const char* kEmptyServiceNameException =
      "org.apache.hadoop.hbase.ipc.EmptyServiceNameException";
  static constexpr const char* kUnknownServiceException =
      "org.apache.hadoop.hbase.ipc.UnknownServiceException";
  static constexpr const char* kWrongVersionException =
      "org.apache.hadoop.hbase.ipc.WrongVersionException";
  static constexpr const char* kBadAuthException = "org.apache.hadoop.hbase.ipc.BadAuthException";
  static constexpr const char* kUnsupportedCompressionCodecException =
      "org.apache.hadoop.hbase.ipc.UnsupportedCompressionCodecException";
  static constexpr const char* kDoNotRetryRegionException =
      "org.apache.hadoop.hbase.client.DoNotRetryRegionException";
  static constexpr const char* kRowTooBigException =
      "org.apache.hadoop.hbase.client.RowTooBigException";
  static constexpr const char* kRowTooBigExceptionDeprecated =
      "org.apache.hadoop.hbase.regionserver.RowTooBigException";
  static constexpr const char* kUnknownRegionException =
      "org.apache.hadoop.hbase.UnknownRegionException";
  static constexpr const char* kMergeRegionException =
      "org.apache.hadoop.hbase.exceptions.MergeRegionException";
  static constexpr const char* kNoServerForRegionException =
      "org.apache.hadoop.hbase.client.NoServerForRegionException";
  static constexpr const char* kQuotaExceededException =
      "org.apache.hadoop.hbase.quotas.QuotaExceededException";
  static constexpr const char* kSpaceLimitingException =
      "org.apache.hadoop.hbase.quotas.SpaceLimitingException";
  static constexpr const char* kThrottlingException =
      "org.apache.hadoop.hbase.quotas.ThrottlingException";
  static constexpr const char* kAccessDeniedException =
      "org.apache.hadoop.hbase.security.AccessDeniedException";
  static constexpr const char* kUnknownProtocolException =
      "org.apache.hadoop.hbase.exceptions.UnknownProtocolException";
  static constexpr const char* kRequestTooBigException =
      "org.apache.hadoop.hbase.exceptions.RequestTooBigException";
  static constexpr const char* kNotAllMetaRegionsOnlineException =
      "org.apache.hadoop.hbase.NotAllMetaRegionsOnlineException";
  static constexpr const char* kConstraintException =
      "org.apache.hadoop.hbase.constraint.ConstraintException";
  static constexpr const char* kNoSuchColumnFamilyException =
      "org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException";
  static constexpr const char* kLeaseException =
      "org.apache.hadoop.hbase.regionserver.LeaseException";
  static constexpr const char* kInvalidLabelException =
      "org.apache.hadoop.hbase.security.visibility.InvalidLabelException";

  // TODO:
  // These exceptions are not thrown in the regular read / write paths, although they are
  // DoNotRetryIOExceptions. Add these to the list below in case we start doing Admin/DDL ops
  // ReplicationPeerNotFoundException, XXXSnapshotException, NamespaceExistException,
  // NamespaceNotFoundException, TableExistsException, TableNotDisabledException,
  static const std::vector<const char*> kAllDoNotRetryIOExceptions;

 public:
  /**
   * Returns whether or not the exception should be retried by looking at the
   * client-side IOException, or RemoteException coming from server side.
   */
  static bool ShouldRetry(const folly::exception_wrapper& error);

  /**
   * Returns whether the java exception class extends DoNotRetryException.
   * In the java side, we just have a hierarchy of Exception classes that we use
   * both client side and server side. On the client side, we rethrow the server
   * side exception by un-wrapping the exception from a RemoteException or a ServiceException
   * (see ConnectionUtils.translateException() in Java).
   * Since this object-hierarchy info is not available in C++ side, we are doing a
   * very fragile catch-all list of all exception types in Java that extend the
   * DoNotRetryException class type.
   */
  static bool IsJavaDoNotRetryException(const std::string& java_class_name);

  /**
   * Returns whether the scanner is closed when the client received the
   * remote exception.
   * Since the object-hierarchy info is not available in C++ side, we are doing a
   * very fragile catch-all list of all exception types in Java that extend these
   * three base classes: UnknownScannerException, NotServingRegionException,
   * RegionServerStoppedException
   */
  static bool IsScannerClosed(const folly::exception_wrapper& exception);

  static bool IsScannerOutOfOrder(const folly::exception_wrapper& exception);
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/connection/client-dispatcher.cc<|end_filename|>
/*
 * 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.
 *
 */
#include "hbase/connection/client-dispatcher.h"

#include <folly/ExceptionWrapper.h>
#include <folly/Format.h>
#include <folly/io/async/AsyncSocketException.h>
#include <utility>

#include "hbase/connection/rpc-connection.h"
#include "hbase/exceptions/exception.h"

using std::unique_ptr;

namespace hbase {

ClientDispatcher::ClientDispatcher(const std::string &server)
    : current_call_id_(9), requests_(5000), server_(server), is_closed_(false) {}

void ClientDispatcher::read(Context *ctx, unique_ptr<Response> in) {
  VLOG(5) << "ClientDispatcher::read()";
  auto call_id = in->call_id();
  auto p = requests_.find_and_erase(call_id);

  VLOG(3) << folly::sformat("Read hbase::Response, call_id: {}, hasException: {}, what: {}",
                            in->call_id(), bool(in->exception()), in->exception().what());

  if (in->exception()) {
    p.setException(in->exception());
  } else {
    p.setValue(std::move(in));
  }
}

void ClientDispatcher::readException(Context *ctx, folly::exception_wrapper e) {
  VLOG(5) << "ClientDispatcher::readException()";
  CloseAndCleanUpCalls();
}

void ClientDispatcher::readEOF(Context *ctx) {
  VLOG(5) << "ClientDispatcher::readEOF()";
  CloseAndCleanUpCalls();
}

folly::Future<unique_ptr<Response>> ClientDispatcher::operator()(unique_ptr<Request> arg) {
  VLOG(5) << "ClientDispatcher::operator()";
  std::lock_guard<std::recursive_mutex> lock(mutex_);
  if (is_closed_) {
    throw ConnectionException("Connection closed already");
  }

  auto call_id = current_call_id_++;
  arg->set_call_id(call_id);

  // TODO: if the map is full (or we have more than hbase.client.perserver.requests.threshold)
  // then throw ServerTooBusyException so that upper layers will retry.
  auto &p = requests_[call_id];

  auto f = p.getFuture();
  p.setInterruptHandler([call_id, this](const folly::exception_wrapper &e) {
    LOG(ERROR) << "e = " << call_id;
    this->requests_.erase(call_id);
    // TODO: call Promise::SetException()?
  });

  try {
    this->pipeline_->write(std::move(arg));
  } catch (const folly::AsyncSocketException &e) {
    p.setException(folly::exception_wrapper{ConnectionException{folly::exception_wrapper{e}}});
    /* clear folly::Promise to avoid overflow. */
    requests_.erase(call_id);
  }

  return f;
}

void ClientDispatcher::CloseAndCleanUpCalls() {
  VLOG(5) << "ClientDispatcher::CloseAndCleanUpCalls()";
  std::lock_guard<std::recursive_mutex> lock(mutex_);
  if (is_closed_) {
    return;
  }
  for (auto &pair : requests_) {
    pair.second.setException(IOException{"Connection closed to server:" + server_});
  }
  requests_.clear();
  is_closed_ = true;
}

folly::Future<folly::Unit> ClientDispatcher::close() {
  CloseAndCleanUpCalls();
  return ClientDispatcherBase::close();
}

folly::Future<folly::Unit> ClientDispatcher::close(Context *ctx) {
  CloseAndCleanUpCalls();
  return ClientDispatcherBase::close(ctx);
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/serde/cell-outputstream.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <memory>

namespace hbase {

class Cell;

/**
 * @brief Encoder / Decoder for Cells.
 */
class CellOutputStream {
 public:
  virtual ~CellOutputStream() {}

  /**
   * Implementation must copy the entire state of the Cell. If the written Cell is modified
   * immediately after the write method returns, the modifications must have absolutely no effect
   * on the copy of the Cell that was added in the write.
   * @param cell Cell to write out
   * @throws IOException
   */
  virtual void Write(const Cell& cell) = 0;

  /**
   * Let the implementation decide what to do.  Usually means writing accumulated data into a
   * byte[] that can then be read from the implementation to be sent to disk, put in the block
   * cache, or sent over the network.
   * @throws IOException
   */
  virtual void Flush() = 0;
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/client/mutation.cc<|end_filename|>


/*
 * 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.
 *
 */

#include "hbase/client/mutation.h"
#include <algorithm>
#include <limits>
#include <stdexcept>

namespace hbase {

Mutation::Mutation(const std::string &row) : Row(row) {}
Mutation::Mutation(const std::string &row, int64_t timestamp) : Row(row), timestamp_(timestamp) {}

Mutation::Mutation(const Mutation &mutation) {
  row_ = mutation.row_;
  durability_ = mutation.durability_;
  timestamp_ = mutation.timestamp_;
  for (auto const &e : mutation.family_map_) {
    for (auto const &c : e.second) {
      family_map_[e.first].push_back(std::make_unique<Cell>(*c));
    }
  }
}

Mutation &Mutation::operator=(const Mutation &mutation) {
  row_ = mutation.row_;
  durability_ = mutation.durability_;
  timestamp_ = mutation.timestamp_;
  for (auto const &e : mutation.family_map_) {
    for (auto const &c : e.second) {
      family_map_[e.first].push_back(std::make_unique<Cell>(*c));
    }
  }
  return *this;
}

pb::MutationProto_Durability Mutation::Durability() const { return durability_; }

Mutation &Mutation::SetDurability(pb::MutationProto_Durability durability) {
  durability_ = durability;
  return *this;
}

bool Mutation::HasFamilies() const { return !family_map_.empty(); }

std::unique_ptr<Cell> Mutation::CreateCell(const std::string &family, const std::string &qualifier,
                                           int64_t timestamp, const std::string &value) {
  return std::make_unique<Cell>(row_, family, qualifier, timestamp, value, hbase::CellType::PUT);
}

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/region-result.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/region-result.h"
#include <glog/logging.h>
#include <stdexcept>

using hbase::pb::RegionLoadStats;

namespace hbase {

RegionResult::RegionResult() {}

RegionResult::~RegionResult() {}

void RegionResult::AddResultOrException(int32_t index, std::shared_ptr<hbase::Result> result,
                                        std::shared_ptr<folly::exception_wrapper> exc) {
  auto index_found = result_or_excption_.find(index);
  if (index_found == result_or_excption_.end()) {
    result_or_excption_[index] = std::make_tuple(result ? result : nullptr, exc ? exc : nullptr);
  } else {
    throw std::runtime_error("Index " + std::to_string(index) +
                             " already set with ResultOrException");
  }
}

void RegionResult::set_stat(std::shared_ptr<RegionLoadStats> stat) { stat_ = stat; }

int RegionResult::ResultOrExceptionSize() const { return result_or_excption_.size(); }

std::shared_ptr<ResultOrExceptionTuple> RegionResult::ResultOrException(int32_t index) const {
  return std::make_shared<ResultOrExceptionTuple>(result_or_excption_.at(index));
}

const std::shared_ptr<RegionLoadStats>& RegionResult::stat() const { return stat_; }

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/server-request.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <map>
#include <memory>
#include <stdexcept>
#include <string>
#include "hbase/client/action.h"
#include "hbase/client/region-location.h"
#include "hbase/client/region-request.h"

namespace hbase {

class ServerRequest {
 public:
  // Concurrent
  using ActionsByRegion = std::map<std::string, std::shared_ptr<RegionRequest>>;

  explicit ServerRequest(std::shared_ptr<RegionLocation> region_location) {
    auto region_name = region_location->region_name();
    auto region_request = std::make_shared<RegionRequest>(region_location);
    actions_by_region_[region_name] = region_request;
  }
  ~ServerRequest() {}

  void AddActionsByRegion(std::shared_ptr<RegionLocation> region_location,
                          std::shared_ptr<Action> action) {
    auto region_name = region_location->region_name();
    auto search = actions_by_region_.find(region_name);
    if (search == actions_by_region_.end()) {
      auto region_request = std::make_shared<RegionRequest>(region_location);
      actions_by_region_[region_name] = region_request;
      actions_by_region_[region_name]->AddAction(action);
    } else {
      search->second->AddAction(action);
    }
  }

  const ActionsByRegion &actions_by_region() const { return actions_by_region_; }

 private:
  ActionsByRegion actions_by_region_;
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/client/put.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <cstdint>
#include <map>
#include <memory>
#include <string>
#include <vector>
#include "hbase/client/cell.h"
#include "hbase/client/mutation.h"

namespace hbase {

class Put : public Mutation {
 public:
  /**
   * Constructors
   */
  explicit Put(const std::string& row) : Mutation(row) {}
  Put(const std::string& row, int64_t timestamp) : Mutation(row, timestamp) {}
  Put(const Put& cput) : Mutation(cput) {}
  Put& operator=(const Put& cput) {
    Mutation::operator=(cput);
    return *this;
  }

  ~Put() = default;

  /**
   *  @brief Add the specified column and value to this Put operation.
   *  @param family family name
   *  @param qualifier column qualifier
   *  @param value column value
   */
  Put& AddColumn(const std::string& family, const std::string& qualifier, const std::string& value);

  /**
   *  @brief Add the specified column and value to this Put operation.
   *  @param family family name
   *  @param qualifier column qualifier
   *  @param timestamp version timestamp
   *  @param value column value
   */
  Put& AddColumn(const std::string& family, const std::string& qualifier, int64_t timestamp,
                 const std::string& value);

  Put& Add(std::unique_ptr<Cell> cell);
};

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/utils/user-util.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/utils/user-util.h"

#include <folly/Logging.h>
#include <krb5/krb5.h>
#include <pwd.h>
#include <sys/types.h>
#include <unistd.h>

namespace hbase {

UserUtil::UserUtil() : once_flag_{} {}

std::string UserUtil::user_name(bool secure) {
  std::call_once(once_flag_, [this, secure]() { compute_user_name(secure); });
  return user_name_;
}

void UserUtil::compute_user_name(bool secure) {
  // According to the man page of getpwuid
  // this should never be free'd
  //
  // So yeah a raw pointer with no ownership....
  struct passwd *passwd = getpwuid(getuid());

  // make sure that we got something.
  if (passwd && passwd->pw_name) {
    user_name_ = std::string{passwd->pw_name};
  }
  if (!secure) return;
  krb5_context ctx;
  krb5_error_code ret = krb5_init_context(&ctx);
  if (ret != 0) {
    throw std::runtime_error("cannot init krb ctx " + std::to_string(ret));
  }
  krb5_ccache ccache;
  ret = krb5_cc_default(ctx, &ccache);
  if (ret != 0) {
    throw std::runtime_error("cannot get default cache " + std::to_string(ret));
  }
  // Here is sample principal: hbase/23a03935850c@EXAMPLE.<EMAIL>
  // There may be one (user) or two (user/host) components before the @ sign
  krb5_principal princ;
  ret = krb5_cc_get_principal(ctx, ccache, &princ);
  if (ret != 0) {
    throw std::runtime_error("cannot get default principal " + std::to_string(ret));
  }
  user_name_ = princ->data->data;
  if (krb5_princ_size(ctx, princ) >= 2) {
    user_name_ += "/";
    user_name_ += static_cast<char *>(princ->data[1].data);
  }
  user_name_ += "@";
  user_name_ += princ->realm.data;
  VLOG(1) << "user " << user_name_;
  krb5_free_principal(ctx, princ);
  krb5_free_context(ctx);
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/response.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/Conv.h>
#include <folly/ExceptionWrapper.h>

#include <cstdint>
#include <memory>
#include <string>
#include <utility>

#include "hbase/serde/cell-scanner.h"

// Forward
namespace google {
namespace protobuf {
class Message;
}
}

namespace hbase {

/**
 * @brief Class representing a rpc response
 *
 * This is the class sent to a service.
 */
class Response {
 public:
  /**
   * Constructor.
   * Initinalizes the call id to 0. 0 should never be a valid call id.
   */
  Response() : call_id_(0), resp_msg_(nullptr), cell_scanner_(nullptr), exception_(nullptr) {}

  /** Get the call_id */
  uint32_t call_id() { return call_id_; }

  /** Set the call_id */
  void set_call_id(uint32_t call_id) { call_id_ = call_id; }

  /**
   * Get the response message.
   * The caller is reponsible for knowing the type. In practice the call id is
   * used to figure out the type.
   */
  std::shared_ptr<google::protobuf::Message> resp_msg() const { return resp_msg_; }

  /** Set the response message. */
  void set_resp_msg(std::shared_ptr<google::protobuf::Message> response) {
    resp_msg_ = std::move(response);
  }

  void set_cell_scanner(std::shared_ptr<CellScanner> cell_scanner) { cell_scanner_ = cell_scanner; }

  const std::shared_ptr<CellScanner> cell_scanner() const { return cell_scanner_; }

  folly::exception_wrapper exception() { return exception_; }

  void set_exception(folly::exception_wrapper value) { exception_ = value; }

  std::string DebugString() const {
    std::string s{"call_id:"};
    s += folly::to<std::string>(call_id_);
    s += ", resp_msg:";
    s += resp_msg_->ShortDebugString();
    return s;
  }

 private:
  uint32_t call_id_;
  std::shared_ptr<google::protobuf::Message> resp_msg_;
  std::shared_ptr<CellScanner> cell_scanner_;
  folly::exception_wrapper exception_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/test/connection-pool-test.cc<|end_filename|>
/*
 * 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.
 *
 */

#include <folly/Logging.h>
#include <gmock/gmock.h>

#include "hbase/connection/connection-factory.h"
#include "hbase/connection/connection-id.h"
#include "hbase/connection/connection-pool.h"
#include "hbase/if/HBase.pb.h"
#include "hbase/serde/server-name.h"

using hbase::pb::ServerName;
using ::testing::Return;
using ::testing::_;
using hbase::ConnectionFactory;
using hbase::ConnectionPool;
using hbase::ConnectionId;
using hbase::HBaseService;
using hbase::Request;
using hbase::Response;
using hbase::RpcConnection;
using hbase::SerializePipeline;

class MockConnectionFactory : public ConnectionFactory {
 public:
  MockConnectionFactory() : ConnectionFactory(nullptr, nullptr, nullptr, nullptr) {}
  MOCK_METHOD0(MakeBootstrap, std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>>());
  MOCK_METHOD4(Connect, std::shared_ptr<HBaseService>(
                            std::shared_ptr<RpcConnection> rpc_connection,
                            std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>>,
                            const std::string &hostname, uint16_t port));
};

class MockBootstrap : public wangle::ClientBootstrap<SerializePipeline> {};

class MockService : public HBaseService {
 public:
  folly::Future<std::unique_ptr<Response>> operator()(std::unique_ptr<Request> req) override {
    return folly::makeFuture<std::unique_ptr<Response>>(
        std::make_unique<Response>(do_operation(req.get())));
  }
  MOCK_METHOD1(do_operation, Response(Request *));
};

TEST(TestConnectionPool, TestOnlyCreateOnce) {
  auto hostname = std::string{"hostname"};
  auto mock_boot = std::make_shared<MockBootstrap>();
  auto mock_service = std::make_shared<MockService>();
  auto mock_cf = std::make_shared<MockConnectionFactory>();
  uint32_t port{999};

  EXPECT_CALL((*mock_cf), Connect(_, _, _, _)).Times(1).WillRepeatedly(Return(mock_service));
  EXPECT_CALL((*mock_cf), MakeBootstrap()).Times(1).WillRepeatedly(Return(mock_boot));
  EXPECT_CALL((*mock_service), do_operation(_)).Times(1).WillRepeatedly(Return(Response{}));
  ConnectionPool cp{mock_cf};

  auto remote_id = std::make_shared<ConnectionId>(hostname, port);
  auto result = cp.GetConnection(remote_id);
  ASSERT_TRUE(result != nullptr);
  result = cp.GetConnection(remote_id);
  result->SendRequest(nullptr);
}

TEST(TestConnectionPool, TestOnlyCreateMultipleDispose) {
  std::string hostname_one{"hostname"};
  std::string hostname_two{"hostname_two"};
  uint32_t port{999};

  auto mock_boot = std::make_shared<MockBootstrap>();
  auto mock_service = std::make_shared<MockService>();
  auto mock_cf = std::make_shared<MockConnectionFactory>();

  EXPECT_CALL((*mock_cf), Connect(_, _, _, _)).Times(2).WillRepeatedly(Return(mock_service));
  EXPECT_CALL((*mock_cf), MakeBootstrap()).Times(2).WillRepeatedly(Return(mock_boot));
  EXPECT_CALL((*mock_service), do_operation(_)).Times(4).WillRepeatedly(Return(Response{}));
  ConnectionPool cp{mock_cf};

  {
    auto remote_id = std::make_shared<ConnectionId>(hostname_one, port);
    auto result_one = cp.GetConnection(remote_id);
    result_one->SendRequest(nullptr);
    auto remote_id2 = std::make_shared<ConnectionId>(hostname_two, port);
    auto result_two = cp.GetConnection(remote_id2);
    result_two->SendRequest(nullptr);
  }
  auto remote_id = std::make_shared<ConnectionId>(hostname_one, port);
  auto result_one = cp.GetConnection(remote_id);
  result_one->SendRequest(nullptr);
  auto remote_id2 = std::make_shared<ConnectionId>(hostname_two, port);
  auto result_two = cp.GetConnection(remote_id2);
  result_two->SendRequest(nullptr);
}

TEST(TestConnectionPool, TestCreateOneConnectionForOneService) {
  std::string hostname{"hostname"};
  uint32_t port{999};
  std::string service1{"service1"};
  std::string service2{"service2"};

  auto mock_boot = std::make_shared<MockBootstrap>();
  auto mock_service = std::make_shared<MockService>();
  auto mock_cf = std::make_shared<MockConnectionFactory>();

  EXPECT_CALL((*mock_cf), Connect(_, _, _, _)).Times(2).WillRepeatedly(Return(mock_service));
  EXPECT_CALL((*mock_cf), MakeBootstrap()).Times(2).WillRepeatedly(Return(mock_boot));
  EXPECT_CALL((*mock_service), do_operation(_)).Times(4).WillRepeatedly(Return(Response{}));
  ConnectionPool cp{mock_cf};

  {
    auto remote_id = std::make_shared<ConnectionId>(hostname, port, service1);
    auto result_one = cp.GetConnection(remote_id);
    result_one->SendRequest(nullptr);
    auto remote_id2 = std::make_shared<ConnectionId>(hostname, port, service2);
    auto result_two = cp.GetConnection(remote_id2);
    result_two->SendRequest(nullptr);
  }
  auto remote_id = std::make_shared<ConnectionId>(hostname, port, service1);
  auto result_one = cp.GetConnection(remote_id);
  result_one->SendRequest(nullptr);
  auto remote_id2 = std::make_shared<ConnectionId>(hostname, port, service2);
  auto result_two = cp.GetConnection(remote_id2);
  result_two->SendRequest(nullptr);
}


<|start_filename|>hbase-native-client/src/hbase/client/scan.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/scan.h"

#include <algorithm>
#include <iterator>
#include <limits>
#include <stdexcept>

namespace hbase {

Scan::Scan() {}

Scan::~Scan() {}

Scan::Scan(const std::string &start_row) : start_row_(start_row) { CheckRow(start_row_); }

Scan::Scan(const std::string &start_row, const std::string &stop_row)
    : start_row_(start_row), stop_row_(stop_row) {
  CheckRow(start_row_);
  CheckRow(stop_row_);
}

Scan::Scan(const Scan &scan) : Query(scan) {
  start_row_ = scan.start_row_;
  stop_row_ = scan.stop_row_;
  max_versions_ = scan.max_versions_;
  caching_ = scan.caching_;
  max_result_size_ = scan.max_result_size_;
  cache_blocks_ = scan.cache_blocks_;
  load_column_families_on_demand_ = scan.load_column_families_on_demand_;
  reversed_ = scan.reversed_;
  allow_partial_results_ = scan.allow_partial_results_;
  consistency_ = scan.consistency_;
  tr_.reset(new TimeRange(scan.tr_->MinTimeStamp(), scan.tr_->MaxTimeStamp()));
  family_map_.insert(scan.family_map_.begin(), scan.family_map_.end());
}

Scan &Scan::operator=(const Scan &scan) {
  Query::operator=(scan);
  start_row_ = scan.start_row_;
  stop_row_ = scan.stop_row_;
  max_versions_ = scan.max_versions_;
  caching_ = scan.caching_;
  max_result_size_ = scan.max_result_size_;
  cache_blocks_ = scan.cache_blocks_;
  load_column_families_on_demand_ = scan.load_column_families_on_demand_;
  reversed_ = scan.reversed_;
  allow_partial_results_ = scan.allow_partial_results_;
  consistency_ = scan.consistency_;
  tr_.reset(new TimeRange(scan.tr_->MinTimeStamp(), scan.tr_->MaxTimeStamp()));
  family_map_.insert(scan.family_map_.begin(), scan.family_map_.end());
  return *this;
}

Scan::Scan(const Get &get) {
  cache_blocks_ = get.CacheBlocks();
  max_versions_ = get.MaxVersions();
  tr_.reset(new TimeRange(get.Timerange().MinTimeStamp(), get.Timerange().MaxTimeStamp()));
  family_map_.insert(get.FamilyMap().begin(), get.FamilyMap().end());
}

Scan &Scan::AddFamily(const std::string &family) {
  const auto &it = family_map_.find(family);
  /**
   * Check if any qualifiers are already present or not.
   * Remove all existing qualifiers if the given family is already present in
   * the map
   */
  if (family_map_.end() != it) {
    it->second.clear();
  } else {
    family_map_[family];
  }
  return *this;
}

Scan &Scan::AddColumn(const std::string &family, const std::string &qualifier) {
  const auto &it = std::find(family_map_[family].begin(), family_map_[family].end(), qualifier);
  /**
   * Check if any qualifiers are already present or not.
   * Add only if qualifiers for a given family are not present
   */
  if (it == family_map_[family].end()) {
    family_map_[family].push_back(qualifier);
  }
  return *this;
}

void Scan::SetReversed(bool reversed) { reversed_ = reversed; }

bool Scan::IsReversed() const { return reversed_; }

void Scan::SetStartRow(const std::string &start_row) { start_row_ = start_row; }

const std::string &Scan::StartRow() const { return start_row_; }

void Scan::SetStopRow(const std::string &stop_row) { stop_row_ = stop_row; }

const std::string &Scan::StopRow() const { return stop_row_; }

void Scan::SetCaching(int caching) { caching_ = caching; }

int Scan::Caching() const { return caching_; }

Scan &Scan::SetConsistency(const hbase::pb::Consistency consistency) {
  consistency_ = consistency;
  return *this;
}

hbase::pb::Consistency Scan::Consistency() const { return consistency_; }

void Scan::SetCacheBlocks(bool cache_blocks) { cache_blocks_ = cache_blocks; }

bool Scan::CacheBlocks() const { return cache_blocks_; }

void Scan::SetAllowPartialResults(bool allow_partial_results) {
  allow_partial_results_ = allow_partial_results;
}

bool Scan::AllowPartialResults() const { return allow_partial_results_; }

void Scan::SetLoadColumnFamiliesOnDemand(bool load_column_families_on_demand) {
  load_column_families_on_demand_ = load_column_families_on_demand;
}

bool Scan::LoadColumnFamiliesOnDemand() const { return load_column_families_on_demand_; }

Scan &Scan::SetMaxVersions(uint32_t max_versions) {
  max_versions_ = max_versions;
  return *this;
}

int Scan::MaxVersions() const { return max_versions_; }

void Scan::SetMaxResultSize(int64_t max_result_size) { max_result_size_ = max_result_size; }

int64_t Scan::MaxResultSize() const { return max_result_size_; }

Scan &Scan::SetTimeRange(int64_t min_stamp, int64_t max_stamp) {
  tr_.reset(new TimeRange(min_stamp, max_stamp));
  return *this;
}

Scan &Scan::SetTimeStamp(int64_t timestamp) {
  tr_.reset(new TimeRange(timestamp, timestamp + 1));
  return *this;
}

const TimeRange &Scan::Timerange() const { return *tr_; }

void Scan::CheckRow(const std::string &row) {
  const int32_t kMaxRowLength = std::numeric_limits<int16_t>::max();
  int row_length = row.size();
  if (0 == row_length) {
    throw std::runtime_error("Row length can't be 0");
  }
  if (row_length > kMaxRowLength) {
    throw std::runtime_error("Length of " + row + " is greater than max row size: " +
                             std::to_string(kMaxRowLength));
  }
}

bool Scan::HasFamilies() const { return !family_map_.empty(); }

const std::map<std::string, std::vector<std::string>> &Scan::FamilyMap() const {
  return family_map_;
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/client/cell.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <cstdint>
#include <string>

namespace hbase {

enum class CellType {
  MINIMUM = 0,
  PUT = 4,
  DELETE = 8,
  DELETE_FAMILY_VERSION = 10,
  DELETE_COLUMN = 12,
  DELETE_FAMILY = 14,
  MAXIMUM = 255
};

class Cell {
 public:
  Cell(const std::string &row, const std::string &family, const std::string &qualifier,
       const int64_t timestamp, const std::string &value, const hbase::CellType &cell_type);
  Cell(const Cell &cell);
  Cell &operator=(const Cell &cell);
  virtual ~Cell();
  const std::string &Row() const;
  const std::string &Family() const;
  const std::string &Qualifier() const;
  int64_t Timestamp() const;
  const std::string &Value() const;
  CellType Type() const;
  int64_t SequenceId() const;
  std::string DebugString() const;
  /** Returns estimated size of the Cell object including deep heap space usage
    * of its data. Notice that this is a very rough estimate. */
  size_t EstimatedSize() const;

 private:
  std::string row_;
  std::string family_;
  std::string qualifier_;
  // Since java does not have unsigned, we are also using signed numerics here
  // so that we won't have surprises when large uint64's are treated as
  // negative values in the java server side
  int64_t timestamp_;
  hbase::CellType cell_type_;
  std::string value_;
  int64_t sequence_id_;

 private:
  static const char *TypeToString(CellType type);
};

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/async-client-scanner.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/async-client-scanner.h"

#include <algorithm>
#include <iterator>
#include <limits>
#include <stdexcept>

namespace hbase {

AsyncClientScanner::AsyncClientScanner(
    std::shared_ptr<AsyncConnection> conn, std::shared_ptr<Scan> scan,
    std::shared_ptr<pb::TableName> table_name, std::shared_ptr<RawScanResultConsumer> consumer,
    nanoseconds pause, uint32_t max_retries, nanoseconds scan_timeout_nanos,
    nanoseconds rpc_timeout_nanos, uint32_t start_log_errors_count)
    : conn_(conn),
      scan_(scan),
      table_name_(table_name),
      consumer_(consumer),
      pause_(pause),
      max_retries_(max_retries),
      scan_timeout_nanos_(scan_timeout_nanos),
      rpc_timeout_nanos_(rpc_timeout_nanos),
      start_log_errors_count_(start_log_errors_count) {
  results_cache_ = std::make_shared<ScanResultCache>();
  max_attempts_ = ConnectionUtils::Retries2Attempts(max_retries);
}

void AsyncClientScanner::Start() { OpenScanner(); }

folly::Future<std::shared_ptr<OpenScannerResponse>> AsyncClientScanner::CallOpenScanner(
    std::shared_ptr<hbase::RpcClient> rpc_client,
    std::shared_ptr<hbase::HBaseRpcController> controller,
    std::shared_ptr<hbase::RegionLocation> loc) {
  open_scanner_tries_++;

  auto preq = RequestConverter::ToScanRequest(*scan_, loc->region_name(), scan_->Caching(), false);

  auto self(shared_from_this());
  VLOG(5) << "Calling RPC Client to open the scanner for region:" << loc->DebugString();
  return rpc_client
      ->AsyncCall(loc->server_name().host_name(), loc->server_name().port(), std::move(preq),
                  security::User::defaultUser(), "ClientService")
      .then([self, loc, controller, rpc_client](const std::unique_ptr<Response>& presp) {
        VLOG(5) << "Scan Response:" << presp->DebugString();
        return std::make_shared<OpenScannerResponse>(rpc_client, presp, loc, controller);
      });
}

void AsyncClientScanner::OpenScanner() {
  auto self(shared_from_this());
  open_scanner_tries_ = 1;

  auto caller = conn_->caller_factory()
                    ->Single<std::shared_ptr<OpenScannerResponse>>()
                    ->table(table_name_)
                    ->row(scan_->StartRow())
                    ->locate_type(GetLocateType(*scan_))
                    ->rpc_timeout(rpc_timeout_nanos_)
                    ->operation_timeout(scan_timeout_nanos_)
                    ->pause(pause_)
                    ->max_retries(max_retries_)
                    ->start_log_errors_count(start_log_errors_count_)
                    ->action([&](std::shared_ptr<hbase::HBaseRpcController> controller,
                                 std::shared_ptr<hbase::RegionLocation> loc,
                                 std::shared_ptr<hbase::RpcClient> rpc_client)
                                 -> folly::Future<std::shared_ptr<OpenScannerResponse>> {
                                   return CallOpenScanner(rpc_client, controller, loc);
                                 })
                    ->Build();

  caller->Call()
      .then([this, self](std::shared_ptr<OpenScannerResponse> resp) {
        VLOG(3) << "Opened scanner with id:" << resp->scan_resp_->scanner_id()
                << ", region:" << resp->region_location_->DebugString() << ", starting scan";
        StartScan(resp);
      })
      .onError([this, self](const folly::exception_wrapper& e) {
        VLOG(3) << "Open scan request received error:" << e.what();
        consumer_->OnError(e);
      })
      .then([caller, self](const auto r) { return r; });
}

void AsyncClientScanner::StartScan(std::shared_ptr<OpenScannerResponse> resp) {
  auto self(shared_from_this());
  auto caller = conn_->caller_factory()
                    ->Scan()
                    ->scanner_id(resp->scan_resp_->scanner_id())
                    ->region_location(resp->region_location_)
                    ->scanner_lease_timeout(TimeUtil::MillisToNanos(resp->scan_resp_->ttl()))
                    ->scan(scan_)
                    ->rpc_client(resp->rpc_client_)
                    ->consumer(consumer_)
                    ->results_cache(results_cache_)
                    ->rpc_timeout(rpc_timeout_nanos_)
                    ->scan_timeout(scan_timeout_nanos_)
                    ->pause(pause_)
                    ->max_retries(max_retries_)
                    ->start_log_errors_count(start_log_errors_count_)
                    ->Build();

  caller->Start(resp->controller_, resp->scan_resp_, resp->cell_scanner_)
      .then([caller, self](const bool has_more) {
        if (has_more) {
          // open the next scanner on the next region.
          self->OpenScanner();
        } else {
          self->consumer_->OnComplete();
        }
      })
      .onError([caller, self](const folly::exception_wrapper& e) { self->consumer_->OnError(e); })
      .then([caller, self](const auto r) { return r; });
}

RegionLocateType AsyncClientScanner::GetLocateType(const Scan& scan) {
  // TODO: In C++, there is no Scan::IncludeStartRow() and Scan::IncludeStopRow().
  // When added, this method should be modified to return other RegionLocateTypes
  // (see ConnectionUtils.java #getLocateType())
  // TODO: When reversed scans are implemented, return other RegionLocateTypes
  return RegionLocateType::kCurrent;
}

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/connection/sasl-util.cc<|end_filename|>
/*
 * 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.
 *
 */
#include "hbase/connection/sasl-util.h"

#include <glog/logging.h>
#include <sasl/sasl.h>
#include <sasl/saslplug.h>
#include <sasl/saslutil.h>

#include <string>

int SaslUtil::GetPluginPath(void *context __attribute__((unused)), const char **path) {
  *path = getenv("SASL_PATH");

  if (*path == NULL) {
    *path = kDefaultPluginDir;
  }
  return SASL_OK;
}

void *SaslUtil::MutexNew(void) {
  auto m = new std::mutex();
  return m;
}

int SaslUtil::MutexLock(void *m) {
  (reinterpret_cast<std::mutex *>(m))->lock();
  return SASL_OK;
}

int SaslUtil::MutexUnlock(void *m) {
  (reinterpret_cast<std::mutex *>(m))->unlock();
  return SASL_OK;
}

void SaslUtil::MutexDispose(void *m) {
  std::mutex *mutex = reinterpret_cast<std::mutex *>(m);
  delete mutex;
}

std::once_flag SaslUtil::library_inited_;

void SaslUtil::InitializeSaslLib() {
  std::call_once(library_inited_, []() {
    sasl_set_mutex(reinterpret_cast<sasl_mutex_alloc_t *>(&SaslUtil::MutexNew),
                   reinterpret_cast<sasl_mutex_lock_t *>(&SaslUtil::MutexLock),
                   reinterpret_cast<sasl_mutex_unlock_t *>(&SaslUtil::MutexUnlock),
                   reinterpret_cast<sasl_mutex_free_t *>(&SaslUtil::MutexDispose));
    static sasl_callback_t callbacks[] = {
        {SASL_CB_GETPATH, (sasl_callback_ft)&SaslUtil::GetPluginPath, NULL},
        {SASL_CB_LIST_END, NULL, NULL}};
    int rc = sasl_client_init(callbacks);
    if (rc != SASL_OK) {
      throw std::runtime_error("Cannot initialize client " + std::to_string(rc));
    }
  });
}

std::string SaslUtil::ParseServiceName(std::shared_ptr<hbase::Configuration> conf, bool secure) {
  if (!secure) {
    return std::string();
  }
  std::string svrPrincipal = conf->Get(kServerPrincipalConfKey, "");
  // principal is of this form: hbase/23a03935850c@EXAMPLE.COM
  // where 23a03935850c is the host (optional)
  std::size_t pos = svrPrincipal.find("/");
  if (pos == std::string::npos && svrPrincipal.find("@") != std::string::npos) {
    pos = svrPrincipal.find("@");
  }
  if (pos == std::string::npos) {
    throw std::runtime_error("Couldn't retrieve service principal from conf");
  }
  VLOG(1) << "pos " << pos << " " << svrPrincipal;
  std::string service_name = svrPrincipal.substr(0, pos);
  return service_name;
}


<|start_filename|>hbase-native-client/src/hbase/client/response-converter.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/response-converter.h"
#include <glog/logging.h>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
#include "hbase/client/cell.h"
#include "hbase/client/multi-response.h"
#include "hbase/exceptions/exception.h"

using hbase::pb::GetResponse;
using hbase::pb::MutateResponse;
using hbase::pb::ScanResponse;
using hbase::pb::RegionLoadStats;

namespace hbase {

ResponseConverter::ResponseConverter() {}

ResponseConverter::~ResponseConverter() {}

// impl note: we are returning shared_ptr's instead of unique_ptr's because these
// go inside folly::Future's, making the move semantics extremely tricky.
std::shared_ptr<Result> ResponseConverter::FromGetResponse(const Response& resp) {
  auto get_resp = std::static_pointer_cast<GetResponse>(resp.resp_msg());
  VLOG(3) << "FromGetResponse:" << get_resp->ShortDebugString();
  return ToResult(get_resp->result(), resp.cell_scanner());
}

std::shared_ptr<Result> ResponseConverter::FromMutateResponse(const Response& resp) {
  auto mutate_resp = std::static_pointer_cast<MutateResponse>(resp.resp_msg());
  hbase::pb::Result result = mutate_resp->result();
  return ToResult(mutate_resp->result(), resp.cell_scanner());
}

bool ResponseConverter::BoolFromMutateResponse(const Response& resp) {
  auto mutate_resp = std::static_pointer_cast<MutateResponse>(resp.resp_msg());
  return mutate_resp->processed();
}

std::shared_ptr<Result> ResponseConverter::ToResult(
    const hbase::pb::Result& result, const std::shared_ptr<CellScanner> cell_scanner) {
  std::vector<std::shared_ptr<Cell>> vcells;
  for (auto cell : result.cell()) {
    std::shared_ptr<Cell> pcell =
        std::make_shared<Cell>(cell.row(), cell.family(), cell.qualifier(), cell.timestamp(),
                               cell.value(), static_cast<hbase::CellType>(cell.cell_type()));
    vcells.push_back(pcell);
  }

  // iterate over the cells coming from rpc codec
  if (cell_scanner != nullptr) {
    int cells_read = 0;
    while (cells_read != result.associated_cell_count()) {
      if (cell_scanner->Advance()) {
        vcells.push_back(cell_scanner->Current());
        cells_read += 1;
      } else {
        LOG(ERROR) << "CellScanner::Advance() returned false unexpectedly. Cells Read:- "
                   << cells_read << "; Expected Cell Count:- " << result.associated_cell_count();
        std::runtime_error("CellScanner::Advance() returned false unexpectedly");
      }
    }
  }
  return std::make_shared<Result>(vcells, result.exists(), result.stale(), result.partial());
}

std::vector<std::shared_ptr<Result>> ResponseConverter::FromScanResponse(const Response& resp) {
  auto scan_resp = std::static_pointer_cast<ScanResponse>(resp.resp_msg());
  return FromScanResponse(scan_resp, resp.cell_scanner());
}

std::vector<std::shared_ptr<Result>> ResponseConverter::FromScanResponse(
    const std::shared_ptr<ScanResponse> scan_resp, std::shared_ptr<CellScanner> cell_scanner) {
  VLOG(3) << "FromScanResponse:" << scan_resp->ShortDebugString()
          << " cell_scanner:" << (cell_scanner != nullptr);
  int num_results =
      cell_scanner != nullptr ? scan_resp->cells_per_result_size() : scan_resp->results_size();

  std::vector<std::shared_ptr<Result>> results{static_cast<size_t>(num_results)};
  for (int i = 0; i < num_results; i++) {
    if (cell_scanner != nullptr) {
      // Cells are out in cellblocks.  Group them up again as Results.  How many to read at a
      // time will be found in getCellsLength -- length here is how many Cells in the i'th Result
      int num_cells = scan_resp->cells_per_result(i);

      std::vector<std::shared_ptr<Cell>> vcells;
      for (int j = 0; j < num_cells; j++) {
        if (!cell_scanner->Advance()) {
          std::string msg = "Results sent from server=" + std::to_string(num_results) +
                            ". But only got " + std::to_string(i) +
                            " results completely at client. Resetting the scanner to scan again.";
          LOG(ERROR) << msg;
          throw std::runtime_error(msg);
        }
        vcells.push_back(cell_scanner->Current());
      }
      // TODO: handle partial results per Result by checking partial_flag_per_result
      results[i] = std::make_shared<Result>(vcells, false, scan_resp->stale(), false);
    } else {
      results[i] = ToResult(scan_resp->results(i), cell_scanner);
    }
  }

  return results;
}

std::unique_ptr<hbase::MultiResponse> ResponseConverter::GetResults(
    std::shared_ptr<Request> req, const Response& resp,
    const ServerRequest::ActionsByRegion& actions_by_region) {
  auto multi_req = std::static_pointer_cast<hbase::pb::MultiRequest>(req->req_msg());
  auto multi_resp = std::static_pointer_cast<hbase::pb::MultiResponse>(resp.resp_msg());
  VLOG(3) << "GetResults:" << multi_resp->ShortDebugString();
  int req_region_action_count = multi_req->regionaction_size();
  int res_region_action_count = multi_resp->regionactionresult_size();
  if (req_region_action_count != res_region_action_count) {
    throw std::runtime_error("Request mutation count=" + std::to_string(req_region_action_count) +
                             " does not match response mutation result count=" +
                             std::to_string(res_region_action_count));
  }
  auto multi_response = std::make_unique<hbase::MultiResponse>();
  for (int32_t num = 0; num < res_region_action_count; num++) {
    hbase::pb::RegionAction actions = multi_req->regionaction(num);
    hbase::pb::RegionActionResult action_result = multi_resp->regionactionresult(num);
    hbase::pb::RegionSpecifier rs = actions.region();
    if (rs.has_type() && rs.type() != hbase::pb::RegionSpecifier::REGION_NAME) {
      throw std::runtime_error("We support only encoded types for protobuf multi response.");
    }

    auto region_name = rs.value();
    if (action_result.has_exception()) {
      auto ew = ResponseConverter::GetRemoteException(action_result.exception());
      VLOG(8) << "Store Remote Region Exception:- " << ew->what().toStdString() << "; Region["
              << region_name << "];";
      multi_response->AddRegionException(region_name, ew);
      continue;
    }

    if (actions.action_size() != action_result.resultorexception_size()) {
      throw std::runtime_error("actions.action_size=" + std::to_string(actions.action_size()) +
                               ", action_result.resultorexception_size=" +
                               std::to_string(action_result.resultorexception_size()) +
                               " for region " + actions.region().value());
    }

    auto multi_actions = actions_by_region.at(region_name)->actions();
    uint64_t multi_actions_num = 0;
    for (hbase::pb::ResultOrException roe : action_result.resultorexception()) {
      std::shared_ptr<Result> result;
      std::shared_ptr<folly::exception_wrapper> ew;
      if (roe.has_exception()) {
        auto ew = ResponseConverter::GetRemoteException(roe.exception());
        VLOG(8) << "Store Remote Region Exception:- " << ew->what().toStdString() << "; Region["
                << region_name << "];";
        multi_response->AddRegionException(region_name, ew);
      } else if (roe.has_result()) {
        result = ToResult(roe.result(), resp.cell_scanner());
      } else if (roe.has_service_result()) {
        // TODO Not processing Coprocessor Service Result;
      } else {
        // Sometimes, the response is just "it was processed". Generally, this occurs for things
        // like mutateRows where either we get back 'processed' (or not) and optionally some
        // statistics about the regions we touched.
        std::vector<std::shared_ptr<Cell>> empty_cells;
        result = std::make_shared<Result>(empty_cells, multi_resp->processed() ? true : false,
                                          false, false);
      }
      // We add the original index of the multi-action so that when populating the response back we
      // do it as per the action index
      multi_response->AddRegionResult(
          region_name, multi_actions[multi_actions_num]->original_index(), std::move(result), ew);
      multi_actions_num++;
    }
  }

  if (multi_resp->has_regionstatistics()) {
    hbase::pb::MultiRegionLoadStats stats = multi_resp->regionstatistics();
    for (int i = 0; i < stats.region_size(); i++) {
      multi_response->AddStatistic(stats.region(i).value(),
                                   std::make_shared<RegionLoadStats>(stats.stat(i)));
    }
  }
  return multi_response;
}

std::shared_ptr<folly::exception_wrapper> ResponseConverter::GetRemoteException(
    const hbase::pb::NameBytesPair& exc_resp) {
  std::string what;
  std::string exception_class_name = exc_resp.has_name() ? exc_resp.name() : "";
  std::string stack_trace = exc_resp.has_value() ? exc_resp.value() : "";

  what.append(exception_class_name).append(stack_trace);
  auto remote_exception = std::make_unique<RemoteException>(what);
  remote_exception->set_exception_class_name(exception_class_name)
      ->set_stack_trace(stack_trace)
      ->set_hostname("")
      ->set_port(0);

  return std::make_shared<folly::exception_wrapper>(
      folly::make_exception_wrapper<RemoteException>(*remote_exception));
}
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/serde/zk.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

namespace google {
namespace protobuf {
class Message;
}
}
namespace folly {
class IOBuf;
}

namespace hbase {

/** @brief A class to convert data from ZooKeeper to other formats.
 *
 * This class will convert data to and from Zookeeper into protobuf objects.
 *
 */
class ZkDeserializer {
 public:
  /**
   * Merge the data from a buffer into a given message.
   *
   * @param buf Naked pointer to iobuf containing data read from zookeeper.
   * @param out Naked pointer into which the data will be merged. The message
   * should be the correct type.
   * @return returns true if the parsing was successful.
   */
  bool Parse(folly::IOBuf *buf, google::protobuf::Message *out);
};
}  // namespace hbase


<|start_filename|>hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/model/TestModelBase.java<|end_filename|>
/*
 * 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.hadoop.hbase.rest.model;

import junit.framework.TestCase;
import org.apache.hadoop.hbase.testclassification.SmallTests;
import org.apache.hadoop.hbase.rest.ProtobufMessageHandler;
import org.apache.hadoop.hbase.rest.provider.JAXBContextResolver;
import org.apache.hadoop.hbase.util.Base64;
import org.codehaus.jackson.jaxrs.JacksonJaxbJsonProvider;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.node.ObjectNode;
import org.junit.experimental.categories.Category;

import javax.ws.rs.core.MediaType;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import java.io.IOException;
import java.io.StringReader;
import java.io.StringWriter;

@Category(SmallTests.class)
public abstract class TestModelBase<T> extends TestCase {

  protected String AS_XML;

  protected String AS_PB;

  protected String AS_JSON;

  protected JAXBContext context;

  protected Class<?> clazz;

  protected ObjectMapper mapper;

  protected TestModelBase(Class<?> clazz) throws Exception {
    super();
    this.clazz = clazz;
    context = new JAXBContextResolver().getContext(clazz);
    mapper = new JacksonJaxbJsonProvider().locateMapper(clazz,
        MediaType.APPLICATION_JSON_TYPE);
  }

  protected abstract T buildTestModel();

  @SuppressWarnings("unused")
  protected String toXML(T model) throws JAXBException {
    StringWriter writer = new StringWriter();
    context.createMarshaller().marshal(model, writer);
    return writer.toString();
  }

  protected String toJSON(T model) throws JAXBException, IOException {
    StringWriter writer = new StringWriter();
    mapper.writeValue(writer, model);
//  original marshaller, uncomment this and comment mapper to verify backward compatibility
//  ((JSONJAXBContext)context).createJSONMarshaller().marshallToJSON(model, writer);
    return writer.toString();
  }

  public T fromJSON(String json) throws JAXBException, IOException {
    return (T)
      mapper.readValue(json, clazz);
  }

  public T fromXML(String xml) throws JAXBException {
    return (T)
      context.createUnmarshaller().unmarshal(new StringReader(xml));
  }

  @SuppressWarnings("unused")
  protected byte[] toPB(ProtobufMessageHandler model) {
    return model.createProtobufOutput();
  }

  protected T fromPB(String pb) throws
      Exception {
    return (T)clazz.getMethod("getObjectFromMessage", byte[].class).invoke(
        clazz.newInstance(),
        Base64.decode(AS_PB));
  }

  protected abstract  void checkModel(T model);

  public void testBuildModel() throws Exception {
    checkModel(buildTestModel());
  }

  public void testFromPB() throws Exception {
    checkModel(fromPB(AS_PB));
  }

  public void testFromXML() throws Exception {
    checkModel(fromXML(AS_XML));
  }

  public void testToXML() throws Exception {
    // Uses fromXML to check model because XML element ordering can be random.
    checkModel(fromXML(toXML(buildTestModel())));
  }

  public void testToJSON() throws Exception {
    try {
      ObjectNode expObj = mapper.readValue(AS_JSON, ObjectNode.class);
      ObjectNode actObj = mapper.readValue(toJSON(buildTestModel()), ObjectNode.class);
      assertEquals(expObj, actObj);
    } catch(Exception e) {
      assertEquals(AS_JSON, toJSON(buildTestModel()));
    }
  }

  public void testFromJSON() throws Exception {
    checkModel(fromJSON(AS_JSON));
  }
}



<|start_filename|>hbase-native-client/src/hbase/client/client.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/client.h"

#include <glog/logging.h>
#include <chrono>
#include <exception>
#include <memory>
#include <utility>

using hbase::pb::TableName;

namespace hbase {

Client::Client() {
  HBaseConfigurationLoader loader;
  auto conf = loader.LoadDefaultResources();
  if (!conf) {
    LOG(ERROR) << "Unable to create default Configuration object. Either hbase-default.xml or "
                  "hbase-site.xml is absent in the search path or problems in XML parsing";
    throw std::runtime_error("Configuration object not present.");
  }
  Init(conf.value());
}

Client::Client(const Configuration &conf) { Init(conf); }

void Client::Init(const Configuration &conf) {
  auto conf_ = std::make_shared<Configuration>(conf);
  async_connection_ = AsyncConnectionImpl::Create(conf_);
}

std::unique_ptr<Table> Client::Table(const TableName &table_name) {
  return std::make_unique<hbase::Table>(table_name, async_connection_);
}

void Client::Close() { async_connection_->Close(); }
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/zk-util.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/zk-util.h"

#include <folly/Conv.h>
#include <boost/algorithm/string.hpp>

#include <vector>

namespace hbase {

/**
 * Returns a "proper" zookeeper quorum string, from hbase's broken quorum string formats. In
 * hbase.zookeeper.quorum, the ports are not listed explicitly per server (eg. s1,s2,s3),
 * however ZooKeeper expects the string of the format s1:2181,s2:2181,s3:2181. This code
 * appends the "clientPort" to each node in the quorum string if not there.
 */
std::string ZKUtil::ParseZooKeeperQuorum(const hbase::Configuration& conf) {
  auto zk_quorum = conf.Get(kHBaseZookeeperQuorum_, kDefHBaseZookeeperQuorum_);
  auto zk_port = conf.GetInt(kHBaseZookeeperClientPort_, kDefHBaseZookeeperClientPort_);

  std::vector<std::string> zk_quorum_parts;
  boost::split(zk_quorum_parts, zk_quorum, boost::is_any_of(","), boost::token_compress_on);
  std::vector<std::string> servers;
  for (auto server : zk_quorum_parts) {
    if (boost::contains(server, ":")) {
      servers.push_back(server);
    } else {
      servers.push_back(server + ":" + folly::to<std::string>(zk_port));
    }
  }
  return boost::join(servers, ",");
}

std::string ZKUtil::MetaZNode(const hbase::Configuration& conf) {
  std::string zk_node = conf.Get(kHBaseZnodeParent_, kDefHBaseZnodeParent_) + "/";
  zk_node += kHBaseMetaRegionServer_;
  return zk_node;
}

int32_t ZKUtil::SessionTimeout(const hbase::Configuration& conf) {
  return conf.GetInt(kHBaseZookeeperSessionTimeout_, kDefHBaseZookeeperSessionTimeout_);
}

}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/sasl-util.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <memory>
#include <mutex>
#include <string>

#include "hbase/client/configuration.h"

class SaslUtil {
 public:
  void InitializeSaslLib(void);
  static std::string ParseServiceName(std::shared_ptr<hbase::Configuration> conf, bool secure);

 private:
  static constexpr const char *kDefaultPluginDir = "/usr/lib/sasl2";
  // for now the sasl handler is hardcoded to work against the regionservers only. In the future, if
  // we
  // need the master rpc to work, we could have a map of service names to principals to use (similar
  // to the Java implementation)
  static constexpr const char *kServerPrincipalConfKey = "hbase.regionserver.kerberos.principal";

  static int GetPluginPath(void *context, const char **path);
  static void *MutexNew(void);
  static int MutexLock(void *m);
  static int MutexUnlock(void *m);
  static void MutexDispose(void *m);
  static std::once_flag library_inited_;
};


<|start_filename|>hbase-native-client/include/hbase/connection/rpc-client.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <google/protobuf/service.h>

#include <folly/ExceptionWrapper.h>
#include <chrono>
#include <memory>
#include <string>
#include <utility>

#include "hbase/connection/connection-id.h"
#include "hbase/connection/connection-pool.h"
#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/security/user.h"

namespace hbase {

class RpcClient {
 public:
  RpcClient(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
            std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
            std::shared_ptr<Codec> codec, std::shared_ptr<Configuration> conf,
            std::chrono::nanoseconds connect_timeout = std::chrono::nanoseconds(0));

  virtual ~RpcClient() { Close(); }

  virtual std::unique_ptr<Response> SyncCall(const std::string &host, uint16_t port,
                                             std::unique_ptr<Request> req,
                                             std::shared_ptr<security::User> ticket);

  virtual std::unique_ptr<Response> SyncCall(const std::string &host, uint16_t port,
                                             std::unique_ptr<Request> req,
                                             std::shared_ptr<security::User> ticket,
                                             const std::string &service_name);

  virtual folly::Future<std::unique_ptr<Response>> AsyncCall(
      const std::string &host, uint16_t port, std::unique_ptr<Request> req,
      std::shared_ptr<security::User> ticket);

  virtual folly::Future<std::unique_ptr<Response>> AsyncCall(const std::string &host, uint16_t port,
                                                             std::unique_ptr<Request> req,
                                                             std::shared_ptr<security::User> ticket,
                                                             const std::string &service_name);

  virtual void Close();

  std::shared_ptr<ConnectionPool> connection_pool() const { return cp_; }

 private:
  std::shared_ptr<RpcConnection> GetConnection(std::shared_ptr<ConnectionId> remote_id);
  folly::Future<std::unique_ptr<Response>> SendRequest(std::shared_ptr<ConnectionId> remote_id,
                                                       std::unique_ptr<Request> req);
  template <typename EXCEPTION>
  folly::Future<std::unique_ptr<Response>> GetFutureWithException(const EXCEPTION &e);

  folly::Future<std::unique_ptr<Response>> GetFutureWithException(
      const folly::exception_wrapper &ew);

 private:
  std::shared_ptr<ConnectionPool> cp_;
  std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor_;
  std::shared_ptr<Configuration> conf_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/utils/concurrent-map.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <map>
#include <memory>
#include <shared_mutex>
#include <unordered_map>
#include <utility>

namespace hbase {

/**
 * A concurrent version of std::unordered_map where we acquire a shared or exclusive
 * lock for operations. This is NOT a highly-concurrent and scalable implementation
 * since there is only one lock object.
 * Replace this with tbb::concurrent_unordered_map or similar.
 *
 * Concurrency here is different than in Java. For example, the iterators returned from
 * find() will not copy the key, value pairs.
 */
template <typename K, typename V>
class concurrent_map {
 public:
  typedef K key_type;
  typedef V mapped_type;
  typedef std::pair<const key_type, mapped_type> value_type;
  typedef typename std::unordered_map<K, V>::iterator iterator;
  typedef typename std::unordered_map<K, V>::const_iterator const_iterator;

  concurrent_map() : map_(), mutex_() {}
  explicit concurrent_map(int32_t n) : map_(n), mutex_() {}

  void insert(const value_type& value) {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    map_.insert(value);
  }

  /**
   * Return the mapped object for this key. Be careful to not use the return reference
   * to do assignment. I think it won't be thread safe
   */
  mapped_type& at(const key_type& key) {
    std::shared_lock<std::shared_timed_mutex> lock(mutex_);
    iterator where = map_.find(key);
    if (where == end()) {
      std::runtime_error("Key not found");
    }
    return where->second;
  }

  mapped_type& operator[](const key_type& key) {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    iterator where = map_.find(key);
    if (where == end()) {
      return map_[key];
    }
    return where->second;
  }

  /**
   * Atomically finds the entry and removes it from the map, returning
   * the previously associated value.
   */
  mapped_type find_and_erase(const K& key) {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    auto search = map_.find(key);
    // It's an error if it's not there.
    CHECK(search != end());
    auto val = std::move(search->second);
    map_.erase(key);
    return val;
  }

  void erase(const K& key) {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    map_.erase(key);
  }

  iterator begin() { return map_.begin(); }

  const_iterator begin() const { return map_.begin(); }

  const_iterator cbegin() const { return map_.begin(); }

  iterator end() { return map_.end(); }

  const_iterator end() const { return map_.end(); }

  const_iterator cend() const { return map_.end(); }

  iterator find(const K& key) {
    std::shared_lock<std::shared_timed_mutex> lock(mutex_);
    return map_.find(key);
  }

  // TODO: find(), at() returning const_iterator

  bool empty() const {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    return map_.empty();
  }

  void clear() {
    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
    map_.clear();
  }

 private:
  std::shared_timed_mutex mutex_;
  std::unordered_map<K, V> map_;
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/connection/rpc-test-server-handler.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <wangle/channel/Handler.h>

#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/serde/rpc-serde.h"

using namespace hbase;

namespace hbase {
// A real rpc server would probably use generated client/server stubs
class RpcTestServerSerializeHandler
    : public wangle::Handler<std::unique_ptr<folly::IOBuf>, std::unique_ptr<Request>,
                             std::unique_ptr<Response>, std::unique_ptr<folly::IOBuf>> {
 public:
  RpcTestServerSerializeHandler() : serde_() {}

  void read(Context* ctx, std::unique_ptr<folly::IOBuf> buf) override;

  folly::Future<folly::Unit> write(Context* ctx, std::unique_ptr<Response> resp) override;

 private:
  std::unique_ptr<Request> CreateReceivedRequest(const std::string& method_name);

 private:
  hbase::RpcSerde serde_;
};
}  // end of namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/connection-pool.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/SharedMutex.h>
#include <boost/functional/hash.hpp>
#include <chrono>
#include <memory>
#include <mutex>
#include <unordered_map>

#include "hbase/connection/connection-factory.h"
#include "hbase/connection/connection-id.h"
#include "hbase/connection/rpc-connection.h"
#include "hbase/connection/service.h"
#include "hbase/if/HBase.pb.h"

namespace hbase {

/**
 * @brief Connection pooling for HBase rpc connection.
 *
 * This is a thread safe connection pool. It allows getting
 * a shared rpc connection to HBase servers by connection id.
 */
class ConnectionPool {
 public:
  /** Create connection pool wit default connection factory */
  ConnectionPool(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
                 std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
                 std::shared_ptr<Codec> codec, std::shared_ptr<Configuration> conf,
                 std::chrono::nanoseconds connect_timeout = std::chrono::nanoseconds(0));

  /**
   * Constructor that allows specifiying the connetion factory.
   * This is useful for testing.
   */
  explicit ConnectionPool(std::shared_ptr<ConnectionFactory> cf);

  /**
   * Destructor.
   * All connections will be close.
   * All connections will be released
   */
  ~ConnectionPool();

  /**
   * Get a connection to the server name. Start time is ignored.
   * This can be a blocking operation for a short time.
   */
  std::shared_ptr<RpcConnection> GetConnection(std::shared_ptr<ConnectionId> remote_id);

  /**
   * Close/remove a connection.
   */
  void Close(std::shared_ptr<ConnectionId> remote_id);

  /**
   * Close the Connection Pool
   */
  void Close();

 private:
  std::shared_ptr<RpcConnection> GetCachedConnection(std::shared_ptr<ConnectionId> remote_id);
  std::shared_ptr<RpcConnection> GetNewConnection(std::shared_ptr<ConnectionId> remote_id);
  std::unordered_map<std::shared_ptr<ConnectionId>, std::shared_ptr<RpcConnection>,
                     ConnectionIdHash, ConnectionIdEquals>
      connections_;
  folly::SharedMutexWritePriority map_mutex_;
  std::shared_ptr<ConnectionFactory> cf_;
  std::shared_ptr<Configuration> conf_;
};

}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/delete.cc<|end_filename|>


/*
 * 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.
 *
 */

#include "hbase/client/delete.h"
#include <folly/Conv.h>
#include <algorithm>
#include <limits>
#include <stdexcept>
#include <utility>

namespace hbase {

/**
 * @brief Add the specified column to this Delete operation.
 * This is an expensive call in that on the server-side, it first does a
 * get to find the latest versions timestamp.  Then it adds a delete using
 * the fetched cells timestamp.
 *  @param family family name
 *  @param qualifier column qualifier
 */
Delete& Delete::AddColumn(const std::string& family, const std::string& qualifier) {
  return AddColumn(family, qualifier, timestamp_);
}

/**
 *  @brief Add the specified column to this Delete operation.
 *  @param family family name
 *  @param qualifier column qualifier
 *  @param timestamp version timestamp
 */
Delete& Delete::AddColumn(const std::string& family, const std::string& qualifier,
                          int64_t timestamp) {
  if (timestamp < 0) {
    throw std::runtime_error("Timestamp cannot be negative. ts=" +
                             folly::to<std::string>(timestamp));
  }

  return Add(
      std::make_unique<Cell>(row_, family, qualifier, timestamp, "", hbase::CellType::DELETE));
}
/**
 * Delete all versions of the specified column.
 * @param family family name
 * @param qualifier column qualifier
 */
Delete& Delete::AddColumns(const std::string& family, const std::string& qualifier) {
  return AddColumns(family, qualifier, timestamp_);
}
/**
 * Delete all versions of the specified column with a timestamp less than
 * or equal to the specified timestamp.
 * @param family family name
 * @param qualifier column qualifier
 * @param timestamp maximum version timestamp
 */
Delete& Delete::AddColumns(const std::string& family, const std::string& qualifier,
                           int64_t timestamp) {
  if (timestamp < 0) {
    throw std::runtime_error("Timestamp cannot be negative. ts=" +
                             folly::to<std::string>(timestamp));
  }

  return Add(std::make_unique<Cell>(row_, family, qualifier, timestamp, "",
                                    hbase::CellType::DELETE_COLUMN));
}
/**
 * Delete all versions of all columns of the specified family.
 * <p>
 * Overrides previous calls to deleteColumn and deleteColumns for the
 * specified family.
 * @param family family name
 */
Delete& Delete::AddFamily(const std::string& family) { return AddFamily(family, timestamp_); }

/**
 * Delete all columns of the specified family with a timestamp less than
 * or equal to the specified timestamp.
 * <p>
 * Overrides previous calls to deleteColumn and deleteColumns for the
 * specified family.
 * @param family family name
 * @param timestamp maximum version timestamp
 */
Delete& Delete::AddFamily(const std::string& family, int64_t timestamp) {
  const auto& it = family_map_.find(family);
  if (family_map_.end() != it) {
    it->second.clear();
  } else {
    family_map_[family];
  }
  return Add(
      std::make_unique<Cell>(row_, family, "", timestamp, "", hbase::CellType::DELETE_FAMILY));
}
/**
 * Delete all columns of the specified family with a timestamp equal to
 * the specified timestamp.
 * @param family family name
 * @param timestamp version timestamp
 */
Delete& Delete::AddFamilyVersion(const std::string& family, int64_t timestamp) {
  return Add(std::make_unique<Cell>(row_, family, "", timestamp, "",
                                    hbase::CellType::DELETE_FAMILY_VERSION));
}
Delete& Delete::Add(std::unique_ptr<Cell> cell) {
  if (cell->Row() != row_) {
    throw std::runtime_error("The row in " + cell->DebugString() +
                             " doesn't match the original one " + row_);
  }

  family_map_[cell->Family()].push_back(std::move(cell));
  return *this;
}
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/connection/pipeline.cc<|end_filename|>
/*
 * 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.
 *
 */
#include "hbase/connection/pipeline.h"

#include <folly/Logging.h>
#include <wangle/channel/AsyncSocketHandler.h>
#include <wangle/channel/EventBaseHandler.h>
#include <wangle/channel/OutputBufferingHandler.h>
#include <wangle/codec/LengthFieldBasedFrameDecoder.h>

#include "hbase/connection/client-handler.h"
#include "hbase/connection/sasl-handler.h"

namespace hbase {

RpcPipelineFactory::RpcPipelineFactory(std::shared_ptr<Codec> codec,
                                       std::shared_ptr<Configuration> conf)
    : user_util_(), codec_(codec), conf_(conf) {}
SerializePipeline::Ptr RpcPipelineFactory::newPipeline(
    std::shared_ptr<folly::AsyncTransportWrapper> sock) {
  folly::SocketAddress addr;  // for logging
  sock->getPeerAddress(&addr);

  auto pipeline = SerializePipeline::create();
  pipeline->addBack(wangle::AsyncSocketHandler{sock});
  pipeline->addBack(wangle::EventBaseHandler{});
  bool secure = false;
  /* for RPC test, there's no need to setup Sasl */
  if (!conf_->GetBool(RpcSerde::HBASE_CLIENT_RPC_TEST_MODE,
                      RpcSerde::DEFAULT_HBASE_CLIENT_RPC_TEST_MODE)) {
    secure = security::User::IsSecurityEnabled(*conf_);
    pipeline->addBack(SaslHandler{user_util_.user_name(secure), conf_});
  }
  pipeline->addBack(wangle::LengthFieldBasedFrameDecoder{});
  pipeline->addBack(ClientHandler{user_util_.user_name(secure), codec_, conf_, addr.describe()});
  pipeline->finalize();
  return pipeline;
}
}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/pipeline.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <folly/io/IOBufQueue.h>
#include <wangle/service/Service.h>

#include <memory>

#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/client/configuration.h"
#include "hbase/serde/codec.h"
#include "hbase/utils/user-util.h"

namespace hbase {

/** Pipeline to turn IOBuf into requests */
using SerializePipeline = wangle::Pipeline<folly::IOBufQueue &, std::unique_ptr<Request>>;

/**
 * Factory to create new pipelines for HBase RPC's.
 */
class RpcPipelineFactory : public wangle::PipelineFactory<SerializePipeline> {
 public:
  /**
   * Constructor. This will create user util.
   */
  explicit RpcPipelineFactory(std::shared_ptr<Codec> codec, std::shared_ptr<Configuration> conf);

  /**
   * Create a new pipeline.
   * The pipeline will be:
   *
   * - Async Socke Handler
   * - Event Base Handler
   * - Length Field Based Frame Decoder
   * - Client Handler
   */
  SerializePipeline::Ptr newPipeline(std::shared_ptr<folly::AsyncTransportWrapper> sock) override;

 private:
  UserUtil user_util_;
  std::shared_ptr<Codec> codec_;
  std::shared_ptr<Configuration> conf_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/connection/connection-pool.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/connection/connection-pool.h"

#include <folly/Conv.h>
#include <folly/Logging.h>
#include <wangle/service/Service.h>

#include <memory>
#include <string>
#include <utility>

using std::chrono::nanoseconds;

namespace hbase {

ConnectionPool::ConnectionPool(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
                               std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
                               std::shared_ptr<Codec> codec, std::shared_ptr<Configuration> conf,
                               nanoseconds connect_timeout)
    : cf_(std::make_shared<ConnectionFactory>(io_executor, cpu_executor, codec, conf,
                                              connect_timeout)),
      connections_(),
      map_mutex_(),
      conf_(conf) {}
ConnectionPool::ConnectionPool(std::shared_ptr<ConnectionFactory> cf)
    : cf_(cf), connections_(), map_mutex_() {}

ConnectionPool::~ConnectionPool() {}

std::shared_ptr<RpcConnection> ConnectionPool::GetConnection(
    std::shared_ptr<ConnectionId> remote_id) {
  // Try and get th cached connection.
  auto found_ptr = GetCachedConnection(remote_id);

  // If there's no connection then create it.
  if (found_ptr == nullptr) {
    found_ptr = GetNewConnection(remote_id);
  }
  return found_ptr;
}

std::shared_ptr<RpcConnection> ConnectionPool::GetCachedConnection(
    std::shared_ptr<ConnectionId> remote_id) {
  folly::SharedMutexWritePriority::ReadHolder holder(map_mutex_);
  auto found = connections_.find(remote_id);
  if (found == connections_.end()) {
    return nullptr;
  }
  return found->second;
}

std::shared_ptr<RpcConnection> ConnectionPool::GetNewConnection(
    std::shared_ptr<ConnectionId> remote_id) {
  // Grab the upgrade lock. While we are double checking other readers can
  // continue on
  folly::SharedMutexWritePriority::UpgradeHolder u_holder{map_mutex_};

  // Now check if someone else created the connection before we got the lock
  // This is safe since we hold the upgrade lock.
  // upgrade lock is more power than the reader lock.
  auto found = connections_.find(remote_id);
  if (found != connections_.end() && found->second != nullptr) {
    return found->second;
  } else {
    // Yeah it looks a lot like there's no connection
    folly::SharedMutexWritePriority::WriteHolder w_holder{std::move(u_holder)};

    // Make double sure there are not stale connections hanging around.
    connections_.erase(remote_id);

    /* create new connection */
    auto connection = std::make_shared<RpcConnection>(remote_id, cf_);

    connections_.insert(std::make_pair(remote_id, connection));

    return connection;
  }
}

void ConnectionPool::Close(std::shared_ptr<ConnectionId> remote_id) {
  folly::SharedMutexWritePriority::WriteHolder holder{map_mutex_};
  DLOG(INFO) << "Closing RPC Connection to host:" << remote_id->host()
             << ", port:" << folly::to<std::string>(remote_id->port());

  auto found = connections_.find(remote_id);
  if (found == connections_.end() || found->second == nullptr) {
    return;
  }
  found->second->Close();
  connections_.erase(found);
}

void ConnectionPool::Close() {
  folly::SharedMutexWritePriority::WriteHolder holder{map_mutex_};
  for (auto &item : connections_) {
    auto &con = item.second;
    con->Close();
  }
  connections_.clear();
}
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/keyvalue-codec.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/keyvalue-codec.h"

#include <string>

namespace hbase {

KeyValueCodec::KVDecoder::KVDecoder(std::unique_ptr<folly::IOBuf> cell_block, uint32_t offset,
                                    uint32_t length)
    : cell_block_(std::move(cell_block)), offset_(offset), length_(length) {}

KeyValueCodec::KVDecoder::~KVDecoder() {}

std::shared_ptr<Cell> KeyValueCodec::KVDecoder::Decode(folly::io::Cursor &cursor) {
  uint32_t key_length = cursor.readBE<uint32_t>();
  uint32_t value_length = cursor.readBE<uint32_t>();
  uint16_t row_length = cursor.readBE<uint16_t>();
  std::string row = cursor.readFixedString(row_length);
  uint8_t column_family_length = cursor.readBE<uint8_t>();
  std::string column_family = cursor.readFixedString(column_family_length);
  int qualifier_length =
      key_length - (row_length + column_family_length + kHBaseSizeOfKeyInfrastructure_);
  std::string column_qualifier = cursor.readFixedString(qualifier_length);
  uint64_t timestamp = cursor.readBE<uint64_t>();
  uint8_t key_type = cursor.readBE<uint8_t>();
  std::string value = cursor.readFixedString(value_length);

  return std::make_shared<Cell>(row, column_family, column_qualifier, timestamp, value,
                                static_cast<hbase::CellType>(key_type));
}

bool KeyValueCodec::KVDecoder::Advance() {
  if (end_of_cell_block_) {
    return false;
  }

  if (cur_pos_ == length_) {
    end_of_cell_block_ = true;
    return false;
  }

  folly::io::Cursor cursor(cell_block_.get());
  cursor.skip(offset_ + cur_pos_);
  uint32_t current_cell_size = cursor.readBE<uint32_t>();
  current_cell_ = Decode(cursor);
  cur_pos_ += kHBaseSizeOfInt_ + current_cell_size;
  return true;
}

uint32_t KeyValueCodec::KVDecoder::CellBlockLength() const { return length_; }
} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/client/async-scan-rpc-retrying-caller.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/async-scan-rpc-retrying-caller.h"

namespace hbase {

ScanResumerImpl::ScanResumerImpl(std::shared_ptr<AsyncScanRpcRetryingCaller> caller)
    : caller_(caller), mutex_() {}

void ScanResumerImpl::Resume() {
  // just used to fix findbugs warnings. In fact, if resume is called before prepare, then we
  // just return at the first if condition without loading the resp and numValidResuls field. If
  // resume is called after suspend, then it is also safe to just reference resp and
  // numValidResults after the synchronized block as no one will change it anymore.
  std::shared_ptr<pb::ScanResponse> local_resp;
  int64_t local_num_complete_rows;

  {
    std::unique_lock<std::mutex> mlock{mutex_};
    if (state_ == ScanResumerState::kInitialized) {
      // user calls this method before we call prepare, so just set the state to
      // RESUMED, the implementation will just go on.
      state_ = ScanResumerState::kResumed;
      return;
    }
    if (state_ == ScanResumerState::kResumed) {
      // already resumed, give up.
      return;
    }
    state_ = ScanResumerState::kResumed;
    local_resp = resp_;
    local_num_complete_rows = num_complete_rows_;
  }

  caller_->CompleteOrNext(local_resp);
}

bool ScanResumerImpl::Prepare(std::shared_ptr<pb::ScanResponse> resp, int num_complete_rows) {
  std::unique_lock<std::mutex> mlock(mutex_);
  if (state_ == ScanResumerState::kResumed) {
    // user calls resume before we actually suspend the scan, just continue;
    return false;
  }
  state_ = ScanResumerState::kSuspended;
  resp_ = resp;
  num_complete_rows_ = num_complete_rows;

  return true;
}

ScanControllerImpl::ScanControllerImpl(std::shared_ptr<AsyncScanRpcRetryingCaller> caller)
    : caller_(caller) {}

std::shared_ptr<ScanResumer> ScanControllerImpl::Suspend() {
  PreCheck();
  state_ = ScanControllerState::kSuspended;
  resumer_ = std::make_shared<ScanResumerImpl>(caller_);
  return resumer_;
}

void ScanControllerImpl::Terminate() {
  PreCheck();
  state_ = ScanControllerState::kTerminated;
}

// return the current state, and set the state to DESTROYED.
ScanControllerState ScanControllerImpl::Destroy() {
  ScanControllerState state = state_;
  state_ = ScanControllerState::kDestroyed;
  return state;
}

void ScanControllerImpl::PreCheck() {
  CHECK(std::this_thread::get_id() == caller_thread_id_)
      << "The current thread is" << std::this_thread::get_id() << ", expected thread is "
      << caller_thread_id_ << ", you should not call this method outside OnNext or OnHeartbeat";

  CHECK(state_ == ScanControllerState::kInitialized) << "Invalid Stopper state "
                                                     << DebugString(state_);
}

std::string ScanControllerImpl::DebugString(ScanControllerState state) {
  switch (state) {
    case ScanControllerState::kInitialized:
      return "kInitialized";
    case ScanControllerState::kSuspended:
      return "kSuspended";
    case ScanControllerState::kTerminated:
      return "kTerminated";
    case ScanControllerState::kDestroyed:
      return "kDestroyed";
    default:
      return "UNKNOWN";
  }
}

std::string ScanControllerImpl::DebugString(ScanResumerState state) {
  switch (state) {
    case ScanResumerState::kInitialized:
      return "kInitialized";
    case ScanResumerState::kSuspended:
      return "kSuspended";
    case ScanResumerState::kResumed:
      return "kResumed";
    default:
      return "UNKNOWN";
  }
}

AsyncScanRpcRetryingCaller::AsyncScanRpcRetryingCaller(
    std::shared_ptr<AsyncConnection> conn, std::shared_ptr<folly::HHWheelTimer> retry_timer,
    std::shared_ptr<hbase::RpcClient> rpc_client, std::shared_ptr<Scan> scan, int64_t scanner_id,
    std::shared_ptr<ScanResultCache> results_cache, std::shared_ptr<RawScanResultConsumer> consumer,
    std::shared_ptr<RegionLocation> region_location, nanoseconds scanner_lease_timeout_nanos,
    nanoseconds pause, uint32_t max_retries, nanoseconds scan_timeout_nanos,
    nanoseconds rpc_timeout_nanos, uint32_t start_log_errors_count)
    : conn_(conn),
      retry_timer_(retry_timer),
      rpc_client_(rpc_client),
      scan_(scan),
      scanner_id_(scanner_id),
      results_cache_(results_cache),
      consumer_(consumer),
      region_location_(region_location),
      scanner_lease_timeout_nanos_(scanner_lease_timeout_nanos),
      pause_(pause),
      max_retries_(max_retries),
      scan_timeout_nanos_(scan_timeout_nanos),
      rpc_timeout_nanos_(rpc_timeout_nanos),
      start_log_errors_count_(start_log_errors_count),
      promise_(std::make_shared<folly::Promise<bool>>()),
      tries_(1) {
  controller_ = conn_->CreateRpcController();
  start_ns_ = TimeUtil::GetNowNanos();
  max_attempts_ = ConnectionUtils::Retries2Attempts(max_retries);
  exceptions_ = std::make_shared<std::vector<ThrowableWithExtraContext>>();
}

folly::Future<bool> AsyncScanRpcRetryingCaller::Start(
    std::shared_ptr<HBaseRpcController> controller,
    std::shared_ptr<pb::ScanResponse> open_scan_resp,
    const std::shared_ptr<CellScanner> cell_scanner) {
  OnComplete(controller, open_scan_resp, cell_scanner);
  return promise_->getFuture();
}

int64_t AsyncScanRpcRetryingCaller::RemainingTimeNs() {
  return scan_timeout_nanos_.count() - (TimeUtil::GetNowNanos() - start_ns_);
}

void AsyncScanRpcRetryingCaller::OnComplete(std::shared_ptr<HBaseRpcController> controller,
                                            std::shared_ptr<pb::ScanResponse> resp,
                                            const std::shared_ptr<CellScanner> cell_scanner) {
  VLOG(5) << "Scan: OnComplete, scanner_id:" << scanner_id_;

  if (controller->Failed()) {
    OnError(controller->exception());
    return;
  }

  bool is_heartbeat = resp->has_heartbeat_message() && resp->heartbeat_message();

  int64_t num_complete_rows_before = results_cache_->num_complete_rows();
  try {
    auto raw_results = ResponseConverter::FromScanResponse(resp, cell_scanner);

    auto results = results_cache_->AddAndGet(raw_results, is_heartbeat);

    auto scan_controller = std::make_shared<ScanControllerImpl>(shared_from_this());

    if (results.size() > 0) {
      UpdateNextStartRowWhenError(*results[results.size() - 1]);
      VLOG(5) << "Calling consumer->OnNext()";
      consumer_->OnNext(results, scan_controller);
    } else if (is_heartbeat) {
      consumer_->OnHeartbeat(scan_controller);
    }

    ScanControllerState state = scan_controller->Destroy();
    if (state == ScanControllerState::kTerminated) {
      if (resp->has_more_results_in_region() && !resp->more_results_in_region()) {
        // we have more results in region but user request to stop the scan, so we need to close the
        // scanner explicitly.
        CloseScanner();
      }
      CompleteNoMoreResults();
      return;
    }

    int64_t num_complete_rows = results_cache_->num_complete_rows() - num_complete_rows_before;
    if (state == ScanControllerState::kSuspended) {
      if (scan_controller->resumer()->Prepare(resp, num_complete_rows)) {
        return;
      }
    }
  } catch (const std::runtime_error& e) {
    // We can not retry here. The server has responded normally and the call sequence has been
    // increased so a new scan with the same call sequence will cause an
    // OutOfOrderScannerNextException. Let the upper layer open a new scanner.
    LOG(WARNING) << "Received exception in reading the scan response:" << e.what();
    CompleteWhenError(true);
    return;
  }

  CompleteOrNext(resp);
}

void AsyncScanRpcRetryingCaller::CompleteOrNext(std::shared_ptr<pb::ScanResponse> resp) {
  VLOG(5) << "Scan: CompleteOrNext, scanner_id" << scanner_id_
          << ", response:" << resp->ShortDebugString();

  if (resp->has_more_results() && !resp->more_results()) {
    // RS tells us there is no more data for the whole scan
    CompleteNoMoreResults();
    return;
  }
  // TODO: Implement Scan::limit(), and check the limit here

  if (resp->has_more_results_in_region() && !resp->more_results_in_region()) {
    // TODO: check whether Scan is reversed here
    CompleteWhenNoMoreResultsInRegion();
    return;
  }
  Next();
}

void AsyncScanRpcRetryingCaller::CompleteExceptionally(bool close_scanner) {
  VLOG(5) << "Scan: CompleteExceptionally";
  results_cache_->Clear();
  if (close_scanner) {
    CloseScanner();
  }
  this->promise_->setException(RetriesExhaustedException(tries_ - 1, exceptions_));
}

void AsyncScanRpcRetryingCaller::CompleteNoMoreResults() {
  // In master code, scanners auto-close if we have exhausted the region. It may not be the case
  // in branch-1 code. If this is backported, make sure that the scanner is closed.
  VLOG(5) << "Scan: CompleteNoMoreResults, scanner_id:" << scanner_id_;
  promise_->setValue(false);
}

void AsyncScanRpcRetryingCaller::CompleteWhenNoMoreResultsInRegion() {
  VLOG(5) << "Scan: CompleteWhenNoMoreResultsInRegion, scanner_id:" << scanner_id_;
  // In master code, scanners auto-close if we have exhausted the region. It may not be the case
  // in branch-1 code. If this is backported, make sure that the scanner is closed.
  if (NoMoreResultsForScan(*scan_, region_location_->region_info())) {
    CompleteNoMoreResults();
  } else {
    CompleteWithNextStartRow(region_location_->region_info().end_key(), true);
  }
}

void AsyncScanRpcRetryingCaller::CompleteWithNextStartRow(std::string row, bool inclusive) {
  VLOG(5) << "Scan: CompleteWithNextStartRow: region scan is complete, move to next region";
  scan_->SetStartRow(row);
  // TODO: set inclusive if it is reverse scans
  promise_->setValue(true);
}

void AsyncScanRpcRetryingCaller::UpdateNextStartRowWhenError(const Result& result) {
  next_start_row_when_error_ = optional<std::string>(result.Row());
  include_next_start_row_when_error_ = result.Partial();
}

void AsyncScanRpcRetryingCaller::CompleteWhenError(bool close_scanner) {
  VLOG(5) << "Scan: CompleteWhenError, scanner_id:" << scanner_id_;
  results_cache_->Clear();
  if (close_scanner) {
    CloseScanner();
  }
  if (next_start_row_when_error_) {
    // TODO: HBASE-17583 adds include start / stop row to the Scan. Once we rebase and implement
    // those options in Scan , we can start using that here.
    scan_->SetStartRow(include_next_start_row_when_error_
                           ? *next_start_row_when_error_
                           : BytesUtil::CreateClosestRowAfter(*next_start_row_when_error_));
  }
  promise_->setValue(true);
}

void AsyncScanRpcRetryingCaller::OnError(const folly::exception_wrapper& error) {
  VLOG(5) << "Scan: OnError, scanner_id:" << scanner_id_;
  if (tries_ > start_log_errors_count_ || VLOG_IS_ON(5)) {
    LOG(WARNING) << "Call to " << region_location_->server_name().ShortDebugString()
                 << " for scanner id = " << scanner_id_ << " for "
                 << region_location_->region_info().ShortDebugString()
                 << " failed, , tries = " << tries_ << ", maxAttempts = " << max_attempts_
                 << ", timeout = " << TimeUtil::ToMillis(scan_timeout_nanos_).count()
                 << " ms, time elapsed = " << TimeUtil::ElapsedMillis(start_ns_) << " ms"
                 << error.what().toStdString();
  }

  bool scanner_closed = ExceptionUtil::IsScannerClosed(error);
  ThrowableWithExtraContext twec(error, TimeUtil::GetNowNanos());
  exceptions_->push_back(twec);
  if (tries_ >= max_retries_) {
    CompleteExceptionally(!scanner_closed);
    return;
  }

  int64_t delay_ns;
  if (scan_timeout_nanos_.count() > 0) {
    int64_t max_delay_ns = RemainingTimeNs() - ConnectionUtils::kSleepDeltaNs;
    if (max_delay_ns <= 0) {
      CompleteExceptionally(!scanner_closed);
      return;
    }
    delay_ns = std::min(max_delay_ns, ConnectionUtils::GetPauseTime(pause_.count(), tries_ - 1));
  } else {
    delay_ns = ConnectionUtils::GetPauseTime(pause_.count(), tries_ - 1);
  }

  if (scanner_closed) {
    CompleteWhenError(false);
    return;
  }

  if (ExceptionUtil::IsScannerOutOfOrder(error)) {
    CompleteWhenError(true);
    return;
  }
  if (!ExceptionUtil::ShouldRetry(error)) {
    CompleteExceptionally(true);
    return;
  }
  tries_++;

  auto self(shared_from_this());
  conn_->retry_executor()->add([&]() {
    retry_timer_->scheduleTimeoutFn(
        [self]() { self->conn_->cpu_executor()->add([&]() { self->Call(); }); },
        std::chrono::milliseconds(TimeUtil::ToMillis(delay_ns)));
  });
}

bool AsyncScanRpcRetryingCaller::NoMoreResultsForScan(const Scan& scan,
                                                      const pb::RegionInfo& info) {
  if (BytesUtil::IsEmptyStopRow(info.end_key())) {
    return true;
  }
  if (BytesUtil::IsEmptyStopRow(scan.StopRow())) {
    return false;
  }
  int32_t c = BytesUtil::CompareTo(info.end_key(), scan.StopRow());
  // 1. if our stop row is less than the endKey of the region
  // 2. if our stop row is equal to the endKey of the region and we do not include the stop row
  // for scan.
  return c > 0 ||
         (c == 0 /* && !scan.IncludeStopRow()*/);  // TODO: Scans always exclude StopRow for now.
}

void AsyncScanRpcRetryingCaller::Next() {
  VLOG(5) << "Scan: Next";
  next_call_seq_++;
  tries_ = 1;
  exceptions_->clear();
  start_ns_ = TimeUtil::GetNowNanos();
  Call();
}

void AsyncScanRpcRetryingCaller::Call() {
  VLOG(5) << "Scan: Call";
  auto self(shared_from_this());
  // As we have a call sequence for scan, it is useless to have a different rpc timeout which is
  // less than the scan timeout. If the server does not respond in time(usually this will not
  // happen as we have heartbeat now), we will get an OutOfOrderScannerNextException when
  // resending the next request and the only way to fix this is to close the scanner and open a
  // new one.
  int64_t call_timeout_nanos;
  if (scan_timeout_nanos_.count() > 0) {
    int64_t remaining_nanos = scan_timeout_nanos_.count() - (TimeUtil::GetNowNanos() - start_ns_);
    if (remaining_nanos <= 0) {
      CompleteExceptionally(true);
      return;
    }
    call_timeout_nanos = remaining_nanos;
  } else {
    call_timeout_nanos = 0L;
  }

  ResetController(controller_, call_timeout_nanos);

  auto req =
      RequestConverter::ToScanRequest(scanner_id_, scan_->Caching(), false, next_call_seq_, false);

  // do the RPC call
  rpc_client_
      ->AsyncCall(region_location_->server_name().host_name(),
                  region_location_->server_name().port(), std::move(req),
                  security::User::defaultUser(), "ClientService")
      .via(conn_->cpu_executor().get())
      .then([self, this](const std::unique_ptr<Response>& resp) {
        auto scan_resp = std::static_pointer_cast<pb::ScanResponse>(resp->resp_msg());
        return OnComplete(controller_, scan_resp, resp->cell_scanner());
      })
      .onError([self, this](const folly::exception_wrapper& e) { OnError(e); });
}

void AsyncScanRpcRetryingCaller::CloseScanner() {
  auto self(shared_from_this());
  ResetController(controller_, rpc_timeout_nanos_.count());

  VLOG(5) << "Closing scanner with scanner_id:" << folly::to<std::string>(scanner_id_);

  // Do a close scanner RPC. Fire and forget.
  auto req = RequestConverter::ToScanRequest(scanner_id_, 0, true);
  rpc_client_
      ->AsyncCall(region_location_->server_name().host_name(),
                  region_location_->server_name().port(), std::move(req),
                  security::User::defaultUser(), "ClientService")
      .onError([self, this](const folly::exception_wrapper& e) -> std::unique_ptr<Response> {
        LOG(WARNING) << "Call to " + region_location_->server_name().ShortDebugString() +
                            " for closing scanner_id = " + folly::to<std::string>(scanner_id_) +
                            " for " + region_location_->region_info().ShortDebugString() +
                            " failed, ignore, probably already closed. Exception:" +
                            e.what().toStdString();
        return nullptr;
      });
}

void AsyncScanRpcRetryingCaller::ResetController(std::shared_ptr<HBaseRpcController> controller,
                                                 const int64_t& timeout_nanos) {
  controller->Reset();
  if (timeout_nanos >= 0) {
    controller->set_call_timeout(
        milliseconds(std::min(static_cast<int64_t>(INT_MAX), TimeUtil::ToMillis(timeout_nanos))));
  }
}

}  // namespace hbase


<|start_filename|>hbase-native-client/include/hbase/connection/connection-factory.h<|end_filename|>
/*
 * 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.
 *
 */
#pragma once

#include <wangle/concurrent/CPUThreadPoolExecutor.h>
#include <wangle/concurrent/IOThreadPoolExecutor.h>
#include <wangle/service/Service.h>

#include <chrono>
#include <memory>
#include <string>

#include "hbase/connection/pipeline.h"
#include "hbase/connection/request.h"
#include "hbase/connection/response.h"
#include "hbase/connection/service.h"
#include "hbase/security/user.h"

namespace hbase {

class RpcConnection;

/**
 * Class to create a ClientBootstrap and turn it into a connected
 * pipeline.
 */
class ConnectionFactory {
 public:
  /**
   * Constructor.
   * There should only be one ConnectionFactory per client.
   */
  ConnectionFactory(std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor,
                    std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor,
                    std::shared_ptr<Codec> codec, std::shared_ptr<Configuration> conf,
                    std::chrono::nanoseconds connect_timeout = std::chrono::nanoseconds(0));

  /** Default Destructor */
  virtual ~ConnectionFactory() = default;

  /**
   * Create a BootStrap from which a connection can be made.
   */
  virtual std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>> MakeBootstrap();

  /**
   * Connect a ClientBootstrap to a server and return the pipeline.
   *
   * This is mostly visible so that mocks can override socket connections.
   */
  virtual std::shared_ptr<HBaseService> Connect(
      std::shared_ptr<RpcConnection> rpc_connection,
      std::shared_ptr<wangle::ClientBootstrap<SerializePipeline>> client_bootstrap,
      const std::string &hostname, uint16_t port);

  std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor() { return io_executor_; }

  std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor() { return cpu_executor_; }

 private:
  std::chrono::nanoseconds connect_timeout_;
  std::shared_ptr<Configuration> conf_;
  std::shared_ptr<wangle::IOThreadPoolExecutor> io_executor_;
  std::shared_ptr<wangle::CPUThreadPoolExecutor> cpu_executor_;
  std::shared_ptr<RpcPipelineFactory> pipeline_factory_;
};
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/connection/sasl-handler.cc<|end_filename|>
/*
 * 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.
 *
 */
#include "hbase/connection/sasl-handler.h"

#include <glog/logging.h>
#include <sasl/sasl.h>
#include <sasl/saslplug.h>
#include <sasl/saslutil.h>

#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl.h>
#include <wangle/channel/Handler.h>

#include <condition_variable>
#include <memory>
#include <mutex>
#include <string>
#include <utility>

#include "hbase/connection/service.h"
#include "hbase/security/user.h"
using hbase::security::User;

using std::chrono::nanoseconds;
using namespace folly;
using namespace wangle;
using namespace hbase;

SaslHandler::SaslHandler(std::string user_name, std::shared_ptr<Configuration> conf)
    : user_name_(user_name) {
  host_name_.clear();
  secure_ = User::IsSecurityEnabled(*conf);
  service_name_ = SaslUtil::ParseServiceName(conf, secure_);
  sasl_connection_setup_started_.clear();
  sasl_connection_setup_in_progress_.store(true);
}

SaslHandler::SaslHandler(const SaslHandler &hdlr) {
  user_name_ = hdlr.user_name_;
  service_name_ = hdlr.service_name_;
  secure_ = hdlr.secure_;
  host_name_ = hdlr.host_name_;
  // copy-constructor sets the flags below to their initial state as opposed to getting them
  // from the object this class is constructed from. That way, this instance is ready to do
  // sasl stuff without issues, right from the SaslInit. Sharing a sasl session is not useful
  // between two handler instances.
  sasl_connection_setup_started_.clear();
  sasl_connection_setup_in_progress_.store(true);
  sconn_ = nullptr;
}

SaslHandler::~SaslHandler() {
  if (nullptr != sconn_) {
    sasl_dispose(&sconn_);
  }
  sconn_ = nullptr;
}

void SaslHandler::transportActive(Context *ctx) {
  // assign hostname; needed for the sasl handshake if secure
  folly::SocketAddress address;
  ctx->getTransport()->getPeerAddress(&address);
  host_name_ = address.getHostStr();

  // now init the sasl library; this is once per process
  if (secure_) {
    sasl_util_.InitializeSaslLib();
  }
  // write the preamble to kick off the RPC handshake
  VLOG(3) << "Writing RPC connection Preamble to server: " << host_name_;
  auto preamble = RpcSerde::Preamble(secure_);
  ctx->fireWrite(std::move(preamble));
  ctx->fireTransportActive();
}

void SaslHandler::read(Context *ctx, folly::IOBufQueue &buf) {
  // if security is not on, or in case of security-on, if secure connection setup not in progress,
  // pass it up without touching
  if (!secure_ || !sasl_connection_setup_in_progress_.load()) {
    ctx->fireRead(buf);
  } else {
    // message is for this handler; process it appropriately
    ContinueSaslNegotiation(ctx, &buf);
  }
}

folly::Future<folly::Unit> SaslHandler::write(Context *ctx, std::unique_ptr<folly::IOBuf> buf) {
  // if security is on, and if secure connection setup in progress,
  // this message is for this handler to process and respond
  if (secure_ && sasl_connection_setup_in_progress_.load()) {
    // store IOBuf which is to be sent to server after SASL handshake
    iobuf_.push_back(std::move(buf));
    if (!sasl_connection_setup_started_.test_and_set()) {
      // for the first incoming RPC from the higher layer, trigger sasl initialization
      return SaslInit(ctx);
    } else {
      // for the subsequent incoming RPCs from the higher layer, just return empty future
      folly::Promise<folly::Unit> p_;
      return p_.getFuture();
    }
  }
  // pass the bytes recieved down without touching it
  return ctx->fireWrite(std::move(buf));
}

folly::Future<folly::Unit> SaslHandler::WriteSaslOutput(Context *ctx, const char *out,
                                                        unsigned int outlen) {
  int buffer_size = outlen + 4;
  auto iob = IOBuf::create(buffer_size);
  iob->append(buffer_size);
  // Create the array output stream.
  google::protobuf::io::ArrayOutputStream aos{iob->writableData(), buffer_size};
  std::unique_ptr<google::protobuf::io::CodedOutputStream> coded_output =
      std::make_unique<google::protobuf::io::CodedOutputStream>(&aos);
  uint32_t total_size = outlen;
  total_size = ntohl(total_size);
  coded_output->WriteRaw(&total_size, 4);
  coded_output->WriteRaw(out, outlen);
  return ctx->fireWrite(std::move(iob));
}

void SaslHandler::FinishAuth(Context *ctx, folly::IOBufQueue *bufQueue) {
  std::unique_ptr<folly::IOBuf> iob;
  if (!bufQueue->empty()) {
    iob = bufQueue->pop_front();
    throw std::runtime_error("Error in the final step of handshake " +
                             std::string(reinterpret_cast<const char *>(iob->data())));
  } else {
    sasl_connection_setup_in_progress_.store(false);
    // write what we buffered
    for (size_t i = 0; i < iobuf_.size(); i++) {
      iob = std::move(iobuf_.at(i));
      ctx->fireWrite(std::move(iob));
    }
  }
}

folly::Future<folly::Unit> SaslHandler::SaslInit(Context *ctx) {
  int rc;
  const char *mechusing, *mechlist = "GSSAPI";
  const char *out;
  unsigned int outlen;

  rc = sasl_client_new(service_name_.c_str(), /* The service we are using*/
                       host_name_.c_str(), NULL,
                       NULL, /* Local and remote IP address strings
                                   (NULL disables mechanisms which require this info)*/
                       NULL, /*connection-specific callbacks*/
                       0 /*security flags*/, &sconn_);
  if (rc != SASL_OK) {
    LOG(FATAL) << "Cannot create client (" << rc << ") ";
    throw std::runtime_error("Cannot create client");
  }
  int curr_rc;
  do {
    curr_rc = sasl_client_start(sconn_,   /* the same context from above */
                                mechlist, /* the list of mechanisms from the server */
                                NULL,     /* filled in if an interaction is needed */
                                &out,     /* filled in on success */
                                &outlen,  /* filled in on success */
                                &mechusing);
  } while (curr_rc == SASL_INTERACT); /* the mechanism may ask us to fill
     in things many times. result is SASL_CONTINUE on success */
  if (curr_rc != SASL_CONTINUE) {
    throw std::runtime_error("Cannot start client (" + std::to_string(curr_rc) + ")");
  }
  folly::Future<folly::Unit> fut = WriteSaslOutput(ctx, out, outlen);
  return fut;
}

void SaslHandler::ContinueSaslNegotiation(Context *ctx, folly::IOBufQueue *bufQueue) {
  const char *out;
  unsigned int outlen;

  int bytes_sent = 0;
  int bytes_received = 0;

  std::unique_ptr<folly::IOBuf> iob = bufQueue->pop_front();
  bytes_received = iob->length();
  if (bytes_received == 0) {
    throw std::runtime_error("Error in sasl handshake");
  }
  folly::io::RWPrivateCursor c(iob.get());
  std::uint32_t status = c.readBE<std::uint32_t>();
  std::uint32_t sz = c.readBE<std::uint32_t>();

  if (status != 0 /*Status 0 is success*/) {
    // Assumption here is that the response from server is not more than 8 * 1024
    throw std::runtime_error("Error in sasl handshake " +
                             std::string(reinterpret_cast<char *>(c.writableData())));
  }
  out = nullptr;
  outlen = 0;

  int curr_rc =
      sasl_client_step(sconn_,                                     /* our context */
                       reinterpret_cast<char *>(c.writableData()), /* the data from the server */
                       sz,                                         /* its length */
                       NULL,     /* this should be unallocated and NULL */
                       &out,     /* filled in on success */
                       &outlen); /* filled in on success */

  if (curr_rc == SASL_OK || curr_rc == SASL_CONTINUE) {
    WriteSaslOutput(ctx, out, outlen);
  }
  if (curr_rc == SASL_OK) {
    FinishAuth(ctx, bufQueue);
  }
}


<|start_filename|>hbase-native-client/include/hbase/client/action.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <memory>
#include "hbase/client/row.h"

namespace hbase {
class Action {
 public:
  Action(std::shared_ptr<hbase::Row> action, int32_t original_index)
      : action_(action), original_index_(original_index) {}
  ~Action() {}

  int32_t original_index() const { return original_index_; }

  std::shared_ptr<hbase::Row> action() const { return action_; }

 private:
  std::shared_ptr<hbase::Row> action_;
  int32_t original_index_;
  int64_t nonce_ = -1;
  int32_t replica_id_ = -1;
};

} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/client/scan-result-cache.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/scan-result-cache.h"
#include <algorithm>
#include <iterator>
#include <limits>
#include <stdexcept>

namespace hbase {
/**
 * Add the given results to cache and get valid results back.
 * @param results the results of a scan next. Must not be null.
 * @param is_hearthbeat indicate whether the results is gotten from a heartbeat response.
 * @return valid results, never null.
 */
std::vector<std::shared_ptr<Result>> ScanResultCache::AddAndGet(
    const std::vector<std::shared_ptr<Result>> &results, bool is_hearthbeat) {
  // If no results were returned it indicates that either we have the all the partial results
  // necessary to construct the complete result or the server had to send a heartbeat message
  // to the client to keep the client-server connection alive
  if (results.empty()) {
    // If this response was an empty heartbeat message, then we have not exhausted the region
    // and thus there may be more partials server side that still need to be added to the partial
    // list before we form the complete Result
    if (!partial_results_.empty() && !is_hearthbeat) {
      return UpdateNumberOfCompleteResultsAndReturn(
          std::vector<std::shared_ptr<Result>>{Combine()});
    }
    return std::vector<std::shared_ptr<Result>>{};
  }
  // In every RPC response there should be at most a single partial result. Furthermore, if
  // there is a partial result, it is guaranteed to be in the last position of the array.
  auto last = results[results.size() - 1];
  if (last->Partial()) {
    if (partial_results_.empty()) {
      partial_results_.push_back(last);
      std::vector<std::shared_ptr<Result>> new_results;
      std::copy_n(results.begin(), results.size() - 1, std::back_inserter(new_results));
      return UpdateNumberOfCompleteResultsAndReturn(new_results);
    }
    // We have only one result and it is partial
    if (results.size() == 1) {
      // check if there is a row change
      if (partial_results_.at(0)->Row() == last->Row()) {
        partial_results_.push_back(last);
        return std::vector<std::shared_ptr<Result>>{};
      }
      auto complete_result = Combine();
      partial_results_.push_back(last);
      return UpdateNumberOfCompleteResultsAndReturn(complete_result);
    }
    // We have some complete results
    auto results_to_return = PrependCombined(results, results.size() - 1);
    partial_results_.push_back(last);
    return UpdateNumberOfCompleteResultsAndReturn(results_to_return);
  }
  if (!partial_results_.empty()) {
    return UpdateNumberOfCompleteResultsAndReturn(PrependCombined(results, results.size()));
  }
  return UpdateNumberOfCompleteResultsAndReturn(results);
}

void ScanResultCache::Clear() { partial_results_.clear(); }

std::shared_ptr<Result> ScanResultCache::CreateCompleteResult(
    const std::vector<std::shared_ptr<Result>> &partial_results) {
  if (partial_results.empty()) {
    return std::make_shared<Result>(std::vector<std::shared_ptr<Cell>>{}, false, false, false);
  }
  std::vector<std::shared_ptr<Cell>> cells{};
  bool stale = false;
  std::string prev_row = "";
  std::string current_row = "";
  size_t i = 0;
  for (const auto &r : partial_results) {
    current_row = r->Row();
    if (i != 0 && prev_row != current_row) {
      throw new std::runtime_error(
          "Cannot form complete result. Rows of partial results do not match.");
    }
    // Ensure that all Results except the last one are marked as partials. The last result
    // may not be marked as a partial because Results are only marked as partials when
    // the scan on the server side must be stopped due to reaching the maxResultSize.
    // Visualizing it makes it easier to understand:
    // maxResultSize: 2 cells
    // (-x-) represents cell number x in a row
    // Example: row1: -1- -2- -3- -4- -5- (5 cells total)
    // How row1 will be returned by the server as partial Results:
    // Result1: -1- -2- (2 cells, size limit reached, mark as partial)
    // Result2: -3- -4- (2 cells, size limit reached, mark as partial)
    // Result3: -5- (1 cell, size limit NOT reached, NOT marked as partial)
    if (i != partial_results.size() - 1 && !r->Partial()) {
      throw new std::runtime_error("Cannot form complete result. Result is missing partial flag.");
    }
    prev_row = current_row;
    stale = stale || r->Stale();
    for (const auto &c : r->Cells()) {
      cells.push_back(c);
    }
    i++;
  }

  return std::make_shared<Result>(cells, false, stale, false);
}

std::shared_ptr<Result> ScanResultCache::Combine() {
  auto result = CreateCompleteResult(partial_results_);
  partial_results_.clear();
  return result;
}

std::vector<std::shared_ptr<Result>> ScanResultCache::PrependCombined(
    const std::vector<std::shared_ptr<Result>> &results, int length) {
  if (length == 0) {
    return std::vector<std::shared_ptr<Result>>{Combine()};
  }
  // the last part of a partial result may not be marked as partial so here we need to check if
  // there is a row change.
  size_t start;
  if (partial_results_[0]->Row() == results[0]->Row()) {
    partial_results_.push_back(results[0]);
    start = 1;
    length--;
  } else {
    start = 0;
  }
  std::vector<std::shared_ptr<Result>> prepend_results{};
  prepend_results.push_back(Combine());
  std::copy_n(results.begin() + start, length, std::back_inserter(prepend_results));
  return prepend_results;
}

std::vector<std::shared_ptr<Result>> ScanResultCache::UpdateNumberOfCompleteResultsAndReturn(
    const std::shared_ptr<Result> &result) {
  return UpdateNumberOfCompleteResultsAndReturn(std::vector<std::shared_ptr<Result>>{result});
}

std::vector<std::shared_ptr<Result>> ScanResultCache::UpdateNumberOfCompleteResultsAndReturn(
    const std::vector<std::shared_ptr<Result>> &results) {
  num_complete_rows_ += results.size();
  return results;
}
}  // namespace hbase


<|start_filename|>hbase-native-client/src/hbase/client/multi-response.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/multi-response.h"
#include <glog/logging.h>
#include "hbase/client/region-result.h"

using hbase::pb::RegionLoadStats;

namespace hbase {

MultiResponse::MultiResponse() {}

int MultiResponse::Size() const {
  int size = 0;
  for (const auto& result : results_) {
    size += result.second->ResultOrExceptionSize();
  }
  return size;
}

void MultiResponse::AddRegionResult(const std::string& region_name, int32_t original_index,
                                    std::shared_ptr<Result> result,
                                    std::shared_ptr<folly::exception_wrapper> exc) {
  auto itr = results_.find(region_name);
  if (itr == results_.end()) {
    auto region_result = std::make_shared<RegionResult>();
    region_result->AddResultOrException(original_index, result, exc);
    results_[region_name] = region_result;
  } else {
    itr->second->AddResultOrException(original_index, result, exc);
  }
}

void MultiResponse::AddRegionException(const std::string& region_name,
                                       std::shared_ptr<folly::exception_wrapper> exception) {
  VLOG(8) << "Store Region Exception:- " << exception->what() << "; Region[" << region_name << "];";
  bool region_found = false;
  auto itr = exceptions_.find(region_name);
  if (itr == exceptions_.end()) {
    auto region_result = std::make_shared<folly::exception_wrapper>();
    exceptions_[region_name] = exception;
  } else {
    itr->second = exception;
  }
}

std::shared_ptr<folly::exception_wrapper> MultiResponse::RegionException(
    const std::string& region_name) const {
  auto find = exceptions_.at(region_name);
  return find;
}

const std::map<std::string, std::shared_ptr<folly::exception_wrapper> >&
MultiResponse::RegionExceptions() const {
  return exceptions_;
}

void MultiResponse::AddStatistic(const std::string& region_name,
                                 std::shared_ptr<RegionLoadStats> stat) {
  results_[region_name]->set_stat(stat);
}

const std::map<std::string, std::shared_ptr<RegionResult> >& MultiResponse::RegionResults() const {
  return results_;
}

MultiResponse::~MultiResponse() {}

} /* namespace hbase */


<|start_filename|>hbase-native-client/include/hbase/utils/connection-util.h<|end_filename|>
/*
 * 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.
 *
 */

#pragma once

#include <algorithm>
#include <climits>
#include <cstdlib>
#include <memory>
#include <vector>
#include "hbase/utils/time-util.h"

namespace hbase {
class ConnectionUtils {
 public:
  static int Retries2Attempts(const int& retries) {
    return std::max(1, retries == INT_MAX ? INT_MAX : retries + 1);
  }

  /* Add a delta to avoid timeout immediately after a retry sleeping. */
  static const uint64_t kSleepDeltaNs = 1000000;

  static const std::vector<uint32_t> kRetryBackoff;
  /**
   * Calculate pause time. Built on {@link kRetryBackoff}.
   * @param pause time to pause
   * @param tries amount of tries
   * @return How long to wait after <code>tries</code> retries
   */
  static int64_t GetPauseTime(const int64_t& pause, const int32_t& tries) {
    int32_t ntries = tries;
    if (static_cast<size_t>(ntries) >= kRetryBackoff.size()) {
      ntries = kRetryBackoff.size() - 1;
    }
    if (ntries < 0) {
      ntries = 0;
    }

    int64_t normal_pause = pause * kRetryBackoff[ntries];
    // 1% possible jitter
    float r = static_cast<float>(std::rand()) / static_cast<float>(RAND_MAX);
    int64_t jitter = (int64_t)(normal_pause * r * 0.01f);
    return normal_pause + jitter;
  }
};
} /* namespace hbase */


<|start_filename|>hbase-native-client/src/hbase/client/result.cc<|end_filename|>
/*
 * 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.
 *
 */

#include "hbase/client/result.h"

namespace hbase {

Result::~Result() {}

Result::Result(const std::vector<std::shared_ptr<Cell> > &cells, bool exists, bool stale,
               bool partial)
    : exists_(exists), stale_(stale), partial_(partial), cells_(cells) {
  row_ = (cells_.size() == 0 ? "" : cells_[0]->Row());
}

Result::Result(const Result &result) {
  exists_ = result.exists_;
  stale_ = result.stale_;
  partial_ = result.partial_;
  row_ = result.row_;
  if (!result.cells_.empty()) {
    for (const auto &cell : result.cells_) {
      cells_.push_back(cell);
    }
  }
}

const std::vector<std::shared_ptr<Cell> > &Result::Cells() const { return cells_; }

std::vector<std::shared_ptr<Cell> > Result::ColumnCells(const std::string &family,
                                                        const std::string &qualifier) const {
  std::vector<std::shared_ptr<Cell> > column_cells;
  // TODO implement a BinarySearch here ?
  for (const auto &cell : cells_) {
    if (cell->Family() == family && cell->Qualifier() == qualifier) {
      column_cells.push_back(cell);
    }
  }
  return column_cells;
}

const std::shared_ptr<Cell> Result::ColumnLatestCell(const std::string &family,
                                                     const std::string &qualifier) const {
  // TODO implement a BinarySearch here ?
  for (const auto &cell : cells_) {
    // We find the latest(first) occurrence of the Cell for a given column and
    // qualifier and break
    if (cell->Family() == family && cell->Qualifier() == qualifier) {
      return cell;
    }
  }
  return nullptr;
}

optional<std::string> Result::Value(const std::string &family, const std::string &qualifier) const {
  std::shared_ptr<Cell> latest_cell(ColumnLatestCell(family, qualifier));
  if (latest_cell.get()) {
    return optional<std::string>(latest_cell->Value());
  }
  return optional<std::string>();
}

bool Result::IsEmpty() const { return cells_.empty(); }

const std::string &Result::Row() const { return row_; }

int Result::Size() const { return cells_.size(); }

ResultMap Result::Map() const {
  ResultMap result_map;
  for (const auto &cell : cells_) {
    result_map[cell->Family()][cell->Qualifier()][cell->Timestamp()] = cell->Value();
  }
  return result_map;
}

std::map<std::string, std::string> Result::FamilyMap(const std::string &family) const {
  std::map<std::string, std::string> family_map;
  if (!IsEmpty()) {
    auto result_map = Map();
    auto itr = result_map.find(family);
    if (itr == result_map.end()) {
      return family_map;
    }

    for (auto qitr = itr->second.begin(); qitr != itr->second.end(); ++qitr) {
      for (auto vitr = qitr->second.begin(); vitr != qitr->second.end(); ++vitr) {
        // We break after inserting the first value. Result.java takes only
        // the first value
        family_map[qitr->first] = vitr->second;
        break;
      }
    }
  }

  return family_map;
}

std::string Result::DebugString() const {
  std::string ret{"keyvalues="};
  if (IsEmpty()) {
    ret += "NONE";
    return ret;
  }
  ret += "{";
  bool is_first = true;
  for (const auto &cell : cells_) {
    if (is_first) {
      is_first = false;
    } else {
      ret += ", ";
    }
    ret += cell->DebugString();
  }
  ret += "}";

  return ret;
}

size_t Result::EstimatedSize() const {
  size_t s = sizeof(Result);
  s += row_.capacity();
  for (const auto c : cells_) {
    s += sizeof(std::shared_ptr<Cell>);
    s + c->EstimatedSize();
  }
  return s;
}

} /* namespace hbase */
