You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2016/06/10 01:38:13 UTC
incubator-kudu git commit: KUDU-1353: remove per-tablet replica
locations cache
Repository: incubator-kudu
Updated Branches:
refs/heads/master f180051a8 -> 5f7823fe7
KUDU-1353: remove per-tablet replica locations cache
This is the simpler alternative to rebuilding the caches on metadata load.
Instead of using the caches to retrieve tserver RPC addresses, we'll look
the addresses up on the spot using the TSDescriptor cache. The two caches
should be equally coherent, as both were populated as a result of the master
receiving TS heartbeats.
I've also done away with the concept of a 'stale' locations response. The
distinction can be real, but it's confusing, and more importantly, clients
never took advantage of it. And if we're going to persist TSDescriptor data
in master state as has been recently discussed, the distinction becomes even
less useful.
Change-Id: I6376b5307f75f5d505b33a5ff4262da619cd1d8d
Reviewed-on: http://gerrit.cloudera.org:8080/2887
Tested-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Jean-Daniel Cryans
Reviewed-by: Mike Percy <mp...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/5f7823fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/5f7823fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/5f7823fe
Branch: refs/heads/master
Commit: 5f7823fe7f94edeb0f8dbb2b9d7a2201614e5e16
Parents: f180051
Author: Adar Dembo <ad...@cloudera.com>
Authored: Wed Apr 27 18:28:47 2016 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Fri Jun 10 01:28:25 2016 +0000
----------------------------------------------------------------------
.../client/GetMasterRegistrationReceived.java | 1 -
.../org/kududb/client/TestAsyncKuduClient.java | 1 -
src/kudu/client/client-test.cc | 39 ---
src/kudu/integration-tests/mini_cluster.cc | 3 +-
src/kudu/master/catalog_manager.cc | 267 ++++++++-----------
src/kudu/master/catalog_manager.h | 56 +---
src/kudu/master/master-path-handlers.cc | 70 +++--
src/kudu/master/master-path-handlers.h | 12 +-
src/kudu/master/master-test-util.h | 9 +-
src/kudu/master/master.proto | 7 +-
src/kudu/master/ts_manager.cc | 4 +-
src/kudu/master/ts_manager.h | 4 +-
src/kudu/tools/create-demo-table.cc | 3 +
13 files changed, 172 insertions(+), 304 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/java/kudu-client/src/main/java/org/kududb/client/GetMasterRegistrationReceived.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/kududb/client/GetMasterRegistrationReceived.java b/java/kudu-client/src/main/java/org/kududb/client/GetMasterRegistrationReceived.java
index 3e17f38..c7159b3 100644
--- a/java/kudu-client/src/main/java/org/kududb/client/GetMasterRegistrationReceived.java
+++ b/java/kudu-client/src/main/java/org/kududb/client/GetMasterRegistrationReceived.java
@@ -165,7 +165,6 @@ final class GetMasterRegistrationReceived {
.setPartitionKeyEnd(ByteString.EMPTY));
locationBuilder.setTabletId(
ByteString.copyFromUtf8(AsyncKuduClient.MASTER_TABLE_NAME_PLACEHOLDER));
- locationBuilder.setStale(false);
locationBuilder.addReplicas(replicaBuilder);
// No one else has called this before us.
if (responseDCalled.compareAndSet(false, true)) {
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/java/kudu-client/src/test/java/org/kududb/client/TestAsyncKuduClient.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/kududb/client/TestAsyncKuduClient.java b/java/kudu-client/src/test/java/org/kududb/client/TestAsyncKuduClient.java
index ad6e2df..e66a531 100644
--- a/java/kudu-client/src/test/java/org/kududb/client/TestAsyncKuduClient.java
+++ b/java/kudu-client/src/test/java/org/kududb/client/TestAsyncKuduClient.java
@@ -127,7 +127,6 @@ public class TestAsyncKuduClient extends BaseKuduTest {
partition.setPartitionKeyStart(ByteString.copyFrom("a" + i, Charsets.UTF_8.name()));
partition.setPartitionKeyEnd(ByteString.copyFrom("b" + i, Charsets.UTF_8.name()));
tabletPb.setPartition(partition);
- tabletPb.setStale(false);
tabletPb.setTabletId(ByteString.copyFromUtf8("some id " + i));
Master.TSInfoPB.Builder tsInfoBuilder = Master.TSInfoPB.newBuilder();
Common.HostPortPB.Builder hostBuilder = Common.HostPortPB.newBuilder();
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/client/client-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/client-test.cc b/src/kudu/client/client-test.cc
index 94f709b..9d26b8b 100644
--- a/src/kudu/client/client-test.cc
+++ b/src/kudu/client/client-test.cc
@@ -2339,45 +2339,6 @@ TEST_F(ClientTest, TestGetTableSchema) {
ASSERT_STR_CONTAINS(s.ToString(), "The table does not exist");
}
-TEST_F(ClientTest, TestStaleLocations) {
- string tablet_id = GetFirstTabletId(client_table2_.get());
-
- // The Tablet is up and running the location should not be stale
- master::TabletLocationsPB locs_pb;
- ASSERT_OK(cluster_->mini_master()->master()->catalog_manager()->GetTabletLocations(
- tablet_id, &locs_pb));
- ASSERT_FALSE(locs_pb.stale());
-
- // On Master restart and no tablet report we expect the locations to be stale
- cluster_->mini_tablet_server(0)->Shutdown();
- ASSERT_OK(cluster_->mini_master()->Restart());
- ASSERT_OK(cluster_->mini_master()->master()->
- WaitUntilCatalogManagerIsLeaderAndReadyForTests(MonoDelta::FromSeconds(5)));
- ASSERT_OK(cluster_->mini_master()->master()->catalog_manager()->GetTabletLocations(
- tablet_id, &locs_pb));
- ASSERT_TRUE(locs_pb.stale());
-
- // Restart the TS and Wait for the tablets to be reported to the master.
- ASSERT_OK(cluster_->mini_tablet_server(0)->Start());
- ASSERT_OK(cluster_->WaitForTabletServerCount(1));
- ASSERT_OK(cluster_->mini_master()->master()->catalog_manager()->GetTabletLocations(
- tablet_id, &locs_pb));
-
- // It may take a while to bootstrap the tablet and send the location report
- // so spin until we get a non-stale location.
- int wait_time = 1000;
- for (int i = 0; i < 80; ++i) {
- ASSERT_OK(cluster_->mini_master()->master()->catalog_manager()->GetTabletLocations(
- tablet_id, &locs_pb));
- if (!locs_pb.stale()) {
- break;
- }
- SleepFor(MonoDelta::FromMicroseconds(wait_time));
- wait_time = std::min(wait_time * 5 / 4, 1000000);
- }
- ASSERT_FALSE(locs_pb.stale());
-}
-
// Test creating and accessing a table which has multiple tablets,
// each of which is replicated.
//
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/integration-tests/mini_cluster.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/mini_cluster.cc b/src/kudu/integration-tests/mini_cluster.cc
index 584822f..a32287e 100644
--- a/src/kudu/integration-tests/mini_cluster.cc
+++ b/src/kudu/integration-tests/mini_cluster.cc
@@ -252,8 +252,7 @@ Status MiniCluster::WaitForReplicaCount(const string& tablet_id,
while (sw.elapsed().wall_seconds() < kTabletReportWaitTimeSeconds) {
Status s =
leader_mini_master()->master()->catalog_manager()->GetTabletLocations(tablet_id, locations);
- if (s.ok() && ((locations->stale() && expected_count == 0) ||
- (!locations->stale() && locations->replicas_size() == expected_count))) {
+ if (s.ok() && locations->replicas_size() == expected_count) {
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/catalog_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/catalog_manager.cc b/src/kudu/master/catalog_manager.cc
index 30f0cc2..7619759 100644
--- a/src/kudu/master/catalog_manager.cc
+++ b/src/kudu/master/catalog_manager.cc
@@ -1508,13 +1508,15 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
if (!tablet) {
LOG(INFO) << "Got report from unknown tablet " << report.tablet_id()
<< ": Sending delete request for this orphan tablet";
- SendDeleteReplicaRequest(report.tablet_id(), TABLET_DATA_DELETED, boost::none, nullptr, ts_desc,
+ SendDeleteReplicaRequest(report.tablet_id(), TABLET_DATA_DELETED,
+ boost::none, nullptr, ts_desc->permanent_uuid(),
"Report from unknown tablet");
return Status::OK();
}
if (!tablet->table()) {
LOG(INFO) << "Got report from an orphaned tablet " << report.tablet_id();
- SendDeleteReplicaRequest(report.tablet_id(), TABLET_DATA_DELETED, boost::none, nullptr, ts_desc,
+ SendDeleteReplicaRequest(report.tablet_id(), TABLET_DATA_DELETED,
+ boost::none, nullptr, ts_desc->permanent_uuid(),
"Report from an orphaned tablet");
return Status::OK();
}
@@ -1535,8 +1537,9 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
<< " (" << msg << "): Sending delete request for this tablet";
// TODO: Cancel tablet creation, instead of deleting, in cases where
// that might be possible (tablet creation timeout & replacement).
- SendDeleteReplicaRequest(tablet->tablet_id(), TABLET_DATA_DELETED, boost::none,
- tablet->table(), ts_desc, msg);
+ SendDeleteReplicaRequest(tablet->tablet_id(), TABLET_DATA_DELETED,
+ boost::none, tablet->table(),
+ ts_desc->permanent_uuid(), msg);
return Status::OK();
}
@@ -1599,7 +1602,8 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
cstate.config().opid_index() < prev_cstate.config().opid_index() &&
!IsRaftConfigMember(ts_desc->permanent_uuid(), prev_cstate.config())) {
SendDeleteReplicaRequest(report.tablet_id(), TABLET_DATA_TOMBSTONED,
- prev_cstate.config().opid_index(), tablet->table(), ts_desc,
+ prev_cstate.config().opid_index(),
+ tablet->table(), ts_desc->permanent_uuid(),
Substitute("Replica from old config with index $0 (latest is $1)",
cstate.config().opid_index(),
prev_cstate.config().opid_index()));
@@ -1658,9 +1662,8 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
}
}
- // If a replica is reporting a new consensus configuration, reset the tablet's replicas.
- // Note that we leave out replicas who live in tablet servers who have not heartbeated to
- // master yet.
+ // If a replica is reporting a new consensus configuration, update the
+ // master's copy of that configuration.
LOG(INFO) << "Tablet: " << tablet->tablet_id() << " reported consensus state change."
<< " New consensus state: " << cstate.ShortDebugString();
@@ -1674,24 +1677,17 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
final_report = &updated_report;
}
- VLOG(2) << "Resetting replicas for tablet " << final_report->tablet_id()
+ VLOG(2) << "Updating consensus configuration for tablet "
+ << final_report->tablet_id()
<< " from config reported by " << ts_desc->permanent_uuid()
<< " to that committed in log index "
<< final_report->committed_consensus_state().config().opid_index()
<< " with leader state from term "
<< final_report->committed_consensus_state().current_term();
- RETURN_NOT_OK(ResetTabletReplicasFromReportedConfig(*final_report, tablet,
- &tablet_lock, &table_lock));
+ RETURN_NOT_OK(HandleRaftConfigChanged(*final_report, tablet,
+ &tablet_lock, &table_lock));
- } else {
- // Report opid_index is equal to the previous opid_index. If some
- // replica is reporting the same consensus configuration we already know about and hasn't
- // been added as replica, add it.
- DVLOG(2) << "Peer " << ts_desc->permanent_uuid() << " sent full tablet report"
- << " with data we have already received. Ensuring replica is being tracked."
- << " Replica consensus state: " << cstate.ShortDebugString();
- AddReplicaToTabletIfNotFound(ts_desc, report, tablet);
}
}
@@ -1704,7 +1700,6 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
if (!s.ok()) {
LOG(WARNING) << "Error updating tablets: " << s.ToString() << ". Tablet report was: "
<< report.ShortDebugString();
- // TODO: we should undo the in-memory tablet replica locations changes made above.
return s;
}
tablet_lock.Commit();
@@ -1721,7 +1716,7 @@ Status CatalogManager::HandleReportedTablet(TSDescriptor* ts_desc,
return Status::OK();
}
-Status CatalogManager::ResetTabletReplicasFromReportedConfig(
+Status CatalogManager::HandleRaftConfigChanged(
const ReportedTabletPB& report,
const scoped_refptr<TabletInfo>& tablet,
TabletMetadataLock* tablet_lock,
@@ -1732,25 +1727,6 @@ Status CatalogManager::ResetTabletReplicasFromReportedConfig(
const ConsensusStatePB& cstate = report.committed_consensus_state();
*tablet_lock->mutable_data()->pb.mutable_committed_consensus_state() = cstate;
- TabletInfo::ReplicaMap replica_locations;
- for (const consensus::RaftPeerPB& peer : cstate.config().peers()) {
- shared_ptr<TSDescriptor> ts_desc;
- if (!peer.has_permanent_uuid()) {
- return Status::InvalidArgument("Missing UUID for peer", peer.ShortDebugString());
- }
- if (!master_->ts_manager()->LookupTSByUUID(peer.permanent_uuid(), &ts_desc)) {
- LOG_WITH_PREFIX(WARNING) << "Tablet server has never reported in. "
- << "Not including in replica locations map yet. Peer: " << peer.ShortDebugString()
- << "; Tablet: " << tablet->ToString();
- continue;
- }
-
- TabletReplica replica;
- NewReplica(ts_desc.get(), report, &replica);
- InsertOrDie(&replica_locations, replica.ts_desc->permanent_uuid(), replica);
- }
- tablet->SetReplicaLocations(replica_locations);
-
if (FLAGS_master_tombstone_evicted_tablet_replicas) {
unordered_set<string> current_member_uuids;
for (const consensus::RaftPeerPB& peer : cstate.config().peers()) {
@@ -1760,10 +1736,8 @@ Status CatalogManager::ResetTabletReplicasFromReportedConfig(
for (const consensus::RaftPeerPB& prev_peer : prev_cstate.config().peers()) {
const string& peer_uuid = prev_peer.permanent_uuid();
if (!ContainsKey(current_member_uuids, peer_uuid)) {
- shared_ptr<TSDescriptor> ts_desc;
- if (!master_->ts_manager()->LookupTSByUUID(peer_uuid, &ts_desc)) continue;
SendDeleteReplicaRequest(report.tablet_id(), TABLET_DATA_TOMBSTONED,
- prev_cstate.config().opid_index(), tablet->table(), ts_desc.get(),
+ prev_cstate.config().opid_index(), tablet->table(), peer_uuid,
Substitute("TS $0 not found in new config with opid_index $1",
peer_uuid, cstate.config().opid_index()));
}
@@ -1779,24 +1753,6 @@ Status CatalogManager::ResetTabletReplicasFromReportedConfig(
return Status::OK();
}
-void CatalogManager::AddReplicaToTabletIfNotFound(TSDescriptor* ts_desc,
- const ReportedTabletPB& report,
- const scoped_refptr<TabletInfo>& tablet) {
- TabletReplica replica;
- NewReplica(ts_desc, report, &replica);
- // Only inserts if a replica with a matching UUID was not already present.
- ignore_result(tablet->AddToReplicaLocations(replica));
-}
-
-void CatalogManager::NewReplica(TSDescriptor* ts_desc,
- const ReportedTabletPB& report,
- TabletReplica* replica) {
- CHECK(report.has_committed_consensus_state()) << "No cstate: " << report.ShortDebugString();
- replica->state = report.state();
- replica->role = GetConsensusRole(ts_desc->permanent_uuid(), report.committed_consensus_state());
- replica->ts_desc = ts_desc;
-}
-
Status CatalogManager::GetTabletPeer(const string& tablet_id,
scoped_refptr<TabletPeer>* tablet_peer) const {
// Note: CatalogManager has only one table, 'sys_catalog', with only
@@ -1829,11 +1785,8 @@ class TSPicker {
TSPicker() {}
virtual ~TSPicker() {}
- // Sets *ts_desc to the tablet server to contact for the next RPC.
- //
- // This assumes that TSDescriptors are never deleted by the master,
- // so the caller does not take ownership of the returned pointer.
- virtual Status PickReplica(TSDescriptor** ts_desc) = 0;
+ // Sets *ts_uuid to the uuid of the tserver to contact for the next RPC.
+ virtual Status PickReplica(string* ts_uuid) = 0;
private:
DISALLOW_COPY_AND_ASSIGN(TSPicker);
@@ -1843,20 +1796,16 @@ class TSPicker {
// identified by its UUID.
class PickSpecificUUID : public TSPicker {
public:
- PickSpecificUUID(Master* master, string ts_uuid)
- : master_(master), ts_uuid_(std::move(ts_uuid)) {}
+ explicit PickSpecificUUID(string ts_uuid)
+ : ts_uuid_(std::move(ts_uuid)) {}
- virtual Status PickReplica(TSDescriptor** ts_desc) OVERRIDE {
- shared_ptr<TSDescriptor> ts;
- if (!master_->ts_manager()->LookupTSByUUID(ts_uuid_, &ts)) {
- return Status::NotFound("unknown tablet server ID", ts_uuid_);
- }
- *ts_desc = ts.get();
+ virtual Status PickReplica(string* ts_uuid) OVERRIDE {
+ // Just a straight passthrough.
+ *ts_uuid = ts_uuid_;
return Status::OK();
}
private:
- Master* const master_;
const string ts_uuid_;
DISALLOW_COPY_AND_ASSIGN(PickSpecificUUID);
@@ -1870,16 +1819,23 @@ class PickLeaderReplica : public TSPicker {
tablet_(tablet) {
}
- virtual Status PickReplica(TSDescriptor** ts_desc) OVERRIDE {
- TabletInfo::ReplicaMap replica_locations;
- tablet_->GetReplicaLocations(&replica_locations);
- for (const TabletInfo::ReplicaMap::value_type& r : replica_locations) {
- if (r.second.role == consensus::RaftPeerPB::LEADER) {
- *ts_desc = r.second.ts_desc;
- return Status::OK();
- }
+ virtual Status PickReplica(string* ts_uuid) OVERRIDE {
+ TabletMetadataLock l(tablet_.get(), TabletMetadataLock::READ);
+
+ string err_msg;
+ if (!l.data().pb.has_committed_consensus_state()) {
+ // The tablet is still in the PREPARING state and has no replicas.
+ err_msg = Substitute("Tablet $0 has no consensus state",
+ tablet_->tablet_id());
+ } else if (!l.data().pb.committed_consensus_state().has_leader_uuid()) {
+ // The tablet may be in the midst of a leader election.
+ err_msg = Substitute("Tablet $0 consensus state has no leader",
+ tablet_->tablet_id());
+ } else {
+ *ts_uuid = l.data().pb.committed_consensus_state().leader_uuid();
+ return Status::OK();
}
- return Status::NotFound("no leader");
+ return Status::NotFound("No leader found", err_msg);
}
private:
@@ -2091,7 +2047,17 @@ class RetryingTSRpcTask : public MonitoredTask {
Status ResetTSProxy() {
// TODO: if there is no replica available, should we still keep the task running?
- RETURN_NOT_OK(replica_picker_->PickReplica(&target_ts_desc_));
+ string ts_uuid;
+ RETURN_NOT_OK(replica_picker_->PickReplica(&ts_uuid));
+ shared_ptr<TSDescriptor> ts_desc;
+ if (!master_->ts_manager()->LookupTSByUUID(ts_uuid, &ts_desc)) {
+ return Status::NotFound(Substitute("Could not find TS for UUID $0",
+ ts_uuid));
+ }
+
+ // This assumes that TSDescriptors are never deleted by the master,
+ // so the task need not take ownership of the returned pointer.
+ target_ts_desc_ = ts_desc.get();
shared_ptr<tserver::TabletServerAdminServiceProxy> ts_proxy;
RETURN_NOT_OK(target_ts_desc_->GetTSAdminProxy(master_->messenger(), &ts_proxy));
@@ -2119,7 +2085,7 @@ class RetrySpecificTSRpcTask : public RetryingTSRpcTask {
const scoped_refptr<TableInfo>& table)
: RetryingTSRpcTask(master,
callback_pool,
- gscoped_ptr<TSPicker>(new PickSpecificUUID(master, permanent_uuid)),
+ gscoped_ptr<TSPicker>(new PickSpecificUUID(permanent_uuid)),
table),
permanent_uuid_(permanent_uuid) {
}
@@ -2542,14 +2508,22 @@ void CatalogManager::SendDeleteTableRequest(const scoped_refptr<TableInfo>& tabl
void CatalogManager::SendDeleteTabletRequest(const scoped_refptr<TabletInfo>& tablet,
const string& deletion_msg) {
- TabletInfo::ReplicaMap locations;
- tablet->GetReplicaLocations(&locations);
-
- LOG(INFO) << "Sending DeleteTablet for " << locations.size()
+ TabletMetadataLock l(tablet.get(), TabletMetadataLock::READ);
+ if (!l.data().pb.has_committed_consensus_state()) {
+ // We could end up here if we're deleting a tablet that never made it to
+ // the CREATING state. That would mean no replicas were ever assigned, so
+ // there's nothing to delete.
+ LOG(INFO) << "Not sending DeleteTablet requests; no consensus state for tablet "
+ << tablet->tablet_id();
+ return;
+ }
+ const ConsensusStatePB& cstate = l.data().pb.committed_consensus_state();
+ LOG(INFO) << "Sending DeleteTablet for " << cstate.config().peers().size()
<< " replicas of tablet " << tablet->tablet_id();
- for (const TabletInfo::ReplicaMap::value_type& r : locations) {
+ for (const auto& peer : cstate.config().peers()) {
SendDeleteReplicaRequest(tablet->tablet_id(), TABLET_DATA_DELETED,
- boost::none, tablet->table(), r.second.ts_desc, deletion_msg);
+ boost::none, tablet->table(),
+ peer.permanent_uuid(), deletion_msg);
}
}
@@ -2558,15 +2532,15 @@ void CatalogManager::SendDeleteReplicaRequest(
TabletDataState delete_type,
const boost::optional<int64_t>& cas_config_opid_index_less_or_equal,
const scoped_refptr<TableInfo>& table,
- TSDescriptor* ts_desc,
+ const string& ts_uuid,
const string& reason) {
LOG_WITH_PREFIX(INFO) << Substitute("Deleting tablet $0 on peer $1 "
"with delete type $2 ($3)",
- tablet_id, ts_desc->permanent_uuid(),
+ tablet_id, ts_uuid,
TabletDataState_Name(delete_type),
reason);
AsyncDeleteReplica* call =
- new AsyncDeleteReplica(master_, worker_pool_.get(), ts_desc->permanent_uuid(), table,
+ new AsyncDeleteReplica(master_, worker_pool_.get(), ts_uuid, table,
tablet_id, delete_type, cas_config_opid_index_less_or_equal,
reason);
if (table != nullptr) {
@@ -2643,7 +2617,7 @@ void CatalogManager::HandleAssignCreatingTablet(TabletInfo* tablet,
DeferredAssignmentActions* deferred,
vector<scoped_refptr<TabletInfo> >* new_tablets) {
MonoDelta time_since_updated =
- MonoTime::Now(MonoTime::FINE).GetDeltaSince(tablet->last_update_time());
+ MonoTime::Now(MonoTime::FINE).GetDeltaSince(tablet->last_create_tablet_time());
int64_t remaining_timeout_ms =
FLAGS_tablet_creation_timeout_ms - time_since_updated.ToMilliseconds();
@@ -2871,7 +2845,7 @@ void CatalogManager::SendCreateTabletRequests(const vector<TabletInfo*>& tablets
for (TabletInfo *tablet : tablets) {
const consensus::RaftConfigPB& config =
tablet->metadata().state().pb.committed_consensus_state().config();
- tablet->set_last_update_time(MonoTime::Now(MonoTime::FINE));
+ tablet->set_last_create_tablet_time(MonoTime::Now(MonoTime::FINE));
for (const RaftPeerPB& peer : config.peers()) {
AsyncCreateReplica* task = new AsyncCreateReplica(master_, worker_pool_.get(),
peer.permanent_uuid(), tablet);
@@ -2994,59 +2968,47 @@ void CatalogManager::SelectReplicas(const TSDescriptorVector& ts_descs,
Status CatalogManager::BuildLocationsForTablet(const scoped_refptr<TabletInfo>& tablet,
TabletLocationsPB* locs_pb) {
- TSRegistrationPB reg;
-
- TabletInfo::ReplicaMap locs;
- consensus::ConsensusStatePB cstate;
- {
- TabletMetadataLock l_tablet(tablet.get(), TabletMetadataLock::READ);
- if (PREDICT_FALSE(l_tablet.data().is_deleted())) {
- return Status::NotFound("Tablet deleted", l_tablet.data().pb.state_msg());
- }
-
- if (PREDICT_FALSE(!l_tablet.data().is_running())) {
- return Status::ServiceUnavailable("Tablet not running");
- }
-
- tablet->GetReplicaLocations(&locs);
- if (locs.empty() && l_tablet.data().pb.has_committed_consensus_state()) {
- cstate = l_tablet.data().pb.committed_consensus_state();
- }
+ TabletMetadataLock l_tablet(tablet.get(), TabletMetadataLock::READ);
+ if (PREDICT_FALSE(l_tablet.data().is_deleted())) {
+ return Status::NotFound("Tablet deleted", l_tablet.data().pb.state_msg());
+ }
- locs_pb->mutable_partition()->CopyFrom(tablet->metadata().state().pb.partition());
+ if (PREDICT_FALSE(!l_tablet.data().is_running())) {
+ return Status::ServiceUnavailable("Tablet not running");
}
- locs_pb->set_tablet_id(tablet->tablet_id());
- locs_pb->set_stale(locs.empty());
+ // Guaranteed because the tablet is RUNNING.
+ DCHECK(l_tablet.data().pb.has_committed_consensus_state());
- // If the locations are cached.
- if (!locs.empty()) {
- for (const TabletInfo::ReplicaMap::value_type& replica : locs) {
- TabletLocationsPB_ReplicaPB* replica_pb = locs_pb->add_replicas();
- replica_pb->set_role(replica.second.role);
+ const ConsensusStatePB& cstate = l_tablet.data().pb.committed_consensus_state();
+ for (const consensus::RaftPeerPB& peer : cstate.config().peers()) {
+ // TODO: GetConsensusRole() iterates over all of the peers, making this an
+ // O(n^2) loop. If replication counts get high, it should be optimized.
+ TabletLocationsPB_ReplicaPB* replica_pb = locs_pb->add_replicas();
+ replica_pb->set_role(GetConsensusRole(peer.permanent_uuid(), cstate));
- TSInfoPB* tsinfo_pb = replica_pb->mutable_ts_info();
- tsinfo_pb->set_permanent_uuid(replica.second.ts_desc->permanent_uuid());
+ TSInfoPB* tsinfo_pb = replica_pb->mutable_ts_info();
+ tsinfo_pb->set_permanent_uuid(peer.permanent_uuid());
- replica.second.ts_desc->GetRegistration(®);
+ shared_ptr<TSDescriptor> ts_desc;
+ if (master_->ts_manager()->LookupTSByUUID(peer.permanent_uuid(), &ts_desc)) {
+ TSRegistrationPB reg;
+ ts_desc->GetRegistration(®);
tsinfo_pb->mutable_rpc_addresses()->Swap(reg.mutable_rpc_addresses());
+ } else {
+ // If we've never received a heartbeat from the tserver, we'll fall back
+ // to the last known RPC address in the RaftPeerPB.
+ //
+ // TODO: We should track these RPC addresses in the master table itself.
+ tsinfo_pb->add_rpc_addresses()->CopyFrom(peer.last_known_addr());
}
- return Status::OK();
}
- // If the locations were not cached.
- // TODO: Why would this ever happen? See KUDU-759.
- if (cstate.IsInitialized()) {
- for (const consensus::RaftPeerPB& peer : cstate.config().peers()) {
- TabletLocationsPB_ReplicaPB* replica_pb = locs_pb->add_replicas();
- CHECK(peer.has_permanent_uuid()) << "Missing UUID: " << peer.ShortDebugString();
- replica_pb->set_role(GetConsensusRole(peer.permanent_uuid(), cstate));
+ locs_pb->mutable_partition()->CopyFrom(tablet->metadata().state().pb.partition());
+ locs_pb->set_tablet_id(tablet->tablet_id());
- TSInfoPB* tsinfo_pb = replica_pb->mutable_ts_info();
- tsinfo_pb->set_permanent_uuid(peer.permanent_uuid());
- tsinfo_pb->add_rpc_addresses()->CopyFrom(peer.last_known_addr());
- }
- }
+ // No longer used; always set to false.
+ locs_pb->set_deprecated_stale(false);
return Status::OK();
}
@@ -3097,7 +3059,6 @@ Status CatalogManager::GetTableLocations(const GetTableLocationsRequestPB* req,
table->GetTabletsInRange(req, &tablets_in_range);
TSRegistrationPB reg;
- vector<TabletReplica> locs;
for (const scoped_refptr<TabletInfo>& tablet : tablets_in_range) {
Status s = BuildLocationsForTablet(tablet, resp->add_tablet_locations());
if (s.ok()) {
@@ -3196,36 +3157,20 @@ TabletInfo::TabletInfo(const scoped_refptr<TableInfo>& table,
std::string tablet_id)
: tablet_id_(std::move(tablet_id)),
table_(table),
- last_update_time_(MonoTime::Now(MonoTime::FINE)),
+ last_create_tablet_time_(MonoTime::Now(MonoTime::FINE)),
reported_schema_version_(0) {}
TabletInfo::~TabletInfo() {
}
-void TabletInfo::SetReplicaLocations(const ReplicaMap& replica_locations) {
- std::lock_guard<simple_spinlock> l(lock_);
- last_update_time_ = MonoTime::Now(MonoTime::FINE);
- replica_locations_ = replica_locations;
-}
-
-void TabletInfo::GetReplicaLocations(ReplicaMap* replica_locations) const {
- std::lock_guard<simple_spinlock> l(lock_);
- *replica_locations = replica_locations_;
-}
-
-bool TabletInfo::AddToReplicaLocations(const TabletReplica& replica) {
- std::lock_guard<simple_spinlock> l(lock_);
- return InsertIfNotPresent(&replica_locations_, replica.ts_desc->permanent_uuid(), replica);
-}
-
-void TabletInfo::set_last_update_time(const MonoTime& ts) {
+void TabletInfo::set_last_create_tablet_time(const MonoTime& ts) {
std::lock_guard<simple_spinlock> l(lock_);
- last_update_time_ = ts;
+ last_create_tablet_time_ = ts;
}
-MonoTime TabletInfo::last_update_time() const {
+MonoTime TabletInfo::last_create_tablet_time() const {
std::lock_guard<simple_spinlock> l(lock_);
- return last_update_time_;
+ return last_create_tablet_time_;
}
bool TabletInfo::set_reported_schema_version(uint32_t version) {
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/catalog_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/master/catalog_manager.h b/src/kudu/master/catalog_manager.h
index daf1feb..586be9e 100644
--- a/src/kudu/master/catalog_manager.h
+++ b/src/kudu/master/catalog_manager.h
@@ -80,16 +80,7 @@ struct PersistentTabletInfo {
SysTabletsEntryPB pb;
};
-// Information on a current replica of a tablet.
-// This is copyable so that no locking is needed.
-struct TabletReplica {
- TSDescriptor* ts_desc;
- tablet::TabletStatePB state;
- consensus::RaftPeerPB::Role role;
-};
-
// The information about a single tablet which exists in the cluster,
-// including its state and locations.
//
// This object uses copy-on-write for the portions of data which are persisted
// on disk. This allows the mutated data to be staged and written to disk
@@ -108,7 +99,6 @@ struct TabletReplica {
class TabletInfo : public RefCountedThreadSafe<TabletInfo> {
public:
typedef PersistentTabletInfo cow_state;
- typedef std::unordered_map<std::string, TabletReplica> ReplicaMap;
TabletInfo(const scoped_refptr<TableInfo>& table, std::string tablet_id);
@@ -120,19 +110,9 @@ class TabletInfo : public RefCountedThreadSafe<TabletInfo> {
const CowObject<PersistentTabletInfo>& metadata() const { return metadata_; }
CowObject<PersistentTabletInfo>* mutable_metadata() { return &metadata_; }
- // Accessors for the latest known tablet replica locations.
- // These locations include only the members of the latest-reported Raft
- // configuration whose tablet servers have ever heartbeated to this Master.
- void SetReplicaLocations(const ReplicaMap& replica_locations);
- void GetReplicaLocations(ReplicaMap* replica_locations) const;
-
- // Adds the given replica to the replica_locations_ map.
- // Returns true iff the replica was inserted.
- bool AddToReplicaLocations(const TabletReplica& replica);
-
- // Accessors for the last time the replica locations were updated.
- void set_last_update_time(const MonoTime& ts);
- MonoTime last_update_time() const;
+ // Accessors for the last time create tablet RPCs were sent for this tablet.
+ void set_last_create_tablet_time(const MonoTime& ts);
+ MonoTime last_create_tablet_time() const;
// Accessors for the last reported schema version
bool set_reported_schema_version(uint32_t version);
@@ -154,13 +134,8 @@ class TabletInfo : public RefCountedThreadSafe<TabletInfo> {
// This doesn't protect metadata_ (the on-disk portion).
mutable simple_spinlock lock_;
- // The last time the replica locations were updated.
- // Also set when the Master first attempts to create the tablet.
- MonoTime last_update_time_;
-
- // The locations in the latest raft config where this tablet has been
- // reported. The map is keyed by tablet server UUID.
- ReplicaMap replica_locations_;
+ // The last time the master sent create tablet RPCs for the tablet.
+ MonoTime last_create_tablet_time_;
// Reported schema version (in-memory only).
uint32_t reported_schema_version_;
@@ -495,21 +470,10 @@ class CatalogManager : public tserver::TabletPeerLookupIf {
const ReportedTabletPB& report,
ReportedTabletUpdatesPB *report_updates);
- Status ResetTabletReplicasFromReportedConfig(const ReportedTabletPB& report,
- const scoped_refptr<TabletInfo>& tablet,
- TabletMetadataLock* tablet_lock,
- TableMetadataLock* table_lock);
-
- // Register a tablet server whenever it heartbeats with a consensus configuration. This is
- // needed because we have logic in the Master that states that if a tablet
- // server that is part of a consensus configuration has not heartbeated to the Master yet, we
- // leave it out of the consensus configuration reported to clients.
- // TODO: See if we can remove this logic, as it seems confusing.
- void AddReplicaToTabletIfNotFound(TSDescriptor* ts_desc,
- const ReportedTabletPB& report,
- const scoped_refptr<TabletInfo>& tablet);
-
- void NewReplica(TSDescriptor* ts_desc, const ReportedTabletPB& report, TabletReplica* replica);
+ Status HandleRaftConfigChanged(const ReportedTabletPB& report,
+ const scoped_refptr<TabletInfo>& tablet,
+ TabletMetadataLock* tablet_lock,
+ TableMetadataLock* table_lock);
// Extract the set of tablets that must be processed because not running yet.
void ExtractTabletsToProcess(std::vector<scoped_refptr<TabletInfo>>* tablets_to_process);
@@ -592,7 +556,7 @@ class CatalogManager : public tserver::TabletPeerLookupIf {
tablet::TabletDataState delete_type,
const boost::optional<int64_t>& cas_config_opid_index_less_or_equal,
const scoped_refptr<TableInfo>& table,
- TSDescriptor* ts_desc,
+ const std::string& ts_uuid,
const std::string& reason);
// Start a task to change the config to add an additional voter because the
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/master-path-handlers.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/master-path-handlers.cc b/src/kudu/master/master-path-handlers.cc
index de68cc1..e286bb2 100644
--- a/src/kudu/master/master-path-handlers.cc
+++ b/src/kudu/master/master-path-handlers.cc
@@ -20,12 +20,15 @@
#include <algorithm>
#include <boost/bind.hpp>
#include <map>
+#include <memory>
#include <string>
+#include <utility>
#include <vector>
#include "kudu/common/partition.h"
#include "kudu/common/schema.h"
#include "kudu/common/wire_protocol.h"
+#include "kudu/consensus/quorum_util.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stringprintf.h"
@@ -43,10 +46,13 @@
namespace kudu {
+using consensus::ConsensusStatePB;
using consensus::RaftPeerPB;
-using std::vector;
+using std::pair;
+using std::shared_ptr;
using std::string;
using std::stringstream;
+using std::vector;
using strings::Substitute;
namespace master {
@@ -108,8 +114,9 @@ void MasterPathHandlers::HandleCatalogManager(const Webserver::WebRequest& req,
namespace {
-bool CompareByRole(const TabletReplica& a, const TabletReplica& b) {
- return a.role < b.role;
+bool CompareByRole(const pair<string, RaftPeerPB::Role>& a,
+ const pair<string, RaftPeerPB::Role>& b) {
+ return a.second < b.second;
}
} // anonymous namespace
@@ -167,13 +174,38 @@ void MasterPathHandlers::HandleTablePage(const Webserver::WebRequest& req,
*output << " <tr><th>Tablet ID</th><th>Partition</th><th>State</th>"
"<th>Message</th><th>RaftConfig</th></tr>\n";
for (const scoped_refptr<TabletInfo>& tablet : tablets) {
- TabletInfo::ReplicaMap locations;
- tablet->GetReplicaLocations(&locations);
- vector<TabletReplica> sorted_locations;
- AppendValuesFromMap(locations, &sorted_locations);
- std::sort(sorted_locations.begin(), sorted_locations.end(), &CompareByRole);
-
+ vector<pair<string, RaftPeerPB::Role>> sorted_replicas;
TabletMetadataLock l(tablet.get(), TabletMetadataLock::READ);
+ if (l.data().pb.has_committed_consensus_state()) {
+ const ConsensusStatePB& cstate = l.data().pb.committed_consensus_state();
+ for (const auto& peer : cstate.config().peers()) {
+ RaftPeerPB::Role role = GetConsensusRole(peer.permanent_uuid(), cstate);
+ string html;
+ string location_html;
+ shared_ptr<TSDescriptor> ts_desc;
+ if (master_->ts_manager()->LookupTSByUUID(peer.permanent_uuid(), &ts_desc)) {
+ location_html = TSDescriptorToHtml(*ts_desc.get(), tablet->tablet_id());
+ } else {
+ location_html = EscapeForHtmlToString(peer.permanent_uuid());
+ }
+ if (role == RaftPeerPB::LEADER) {
+ html = Substitute(" <li><b>LEADER: $0</b></li>\n", location_html);
+ } else {
+ html = Substitute(" <li>$0: $1</li>\n",
+ RaftPeerPB_Role_Name(role), location_html);
+ }
+ sorted_replicas.emplace_back(html, role);
+ }
+ }
+ std::sort(sorted_replicas.begin(), sorted_replicas.end(), &CompareByRole);
+
+ // Generate the RaftConfig table cell.
+ stringstream raft_config_html;
+ raft_config_html << "<ul>\n";
+ for (const auto& e : sorted_replicas) {
+ raft_config_html << e.first;
+ }
+ raft_config_html << "</ul>\n";
Partition partition;
Partition::FromPB(l.data().pb.partition(), &partition);
@@ -186,7 +218,7 @@ void MasterPathHandlers::HandleTablePage(const Webserver::WebRequest& req,
EscapeForHtmlToString(partition_schema.PartitionDebugString(partition, schema)),
state,
EscapeForHtmlToString(l.data().pb.state_msg()),
- RaftConfigToHtml(sorted_locations, tablet->tablet_id()));
+ raft_config_html.str());
}
*output << "</table>\n";
@@ -410,24 +442,6 @@ Status MasterPathHandlers::Register(Webserver* server) {
return Status::OK();
}
-string MasterPathHandlers::RaftConfigToHtml(const std::vector<TabletReplica>& locations,
- const std::string& tablet_id) const {
- stringstream html;
-
- html << "<ul>\n";
- for (const TabletReplica& location : locations) {
- string location_html = TSDescriptorToHtml(*location.ts_desc, tablet_id);
- if (location.role == RaftPeerPB::LEADER) {
- html << Substitute(" <li><b>LEADER: $0</b></li>\n", location_html);
- } else {
- html << Substitute(" <li>$0: $1</li>\n",
- RaftPeerPB_Role_Name(location.role), location_html);
- }
- }
- html << "</ul>\n";
- return html.str();
-}
-
string MasterPathHandlers::TSDescriptorToHtml(const TSDescriptor& desc,
const std::string& tablet_id) const {
TSRegistrationPB reg;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/master-path-handlers.h
----------------------------------------------------------------------
diff --git a/src/kudu/master/master-path-handlers.h b/src/kudu/master/master-path-handlers.h
index 149eac5..4406757 100644
--- a/src/kudu/master/master-path-handlers.h
+++ b/src/kudu/master/master-path-handlers.h
@@ -17,13 +17,13 @@
#ifndef KUDU_MASTER_MASTER_PATH_HANDLERS_H
#define KUDU_MASTER_MASTER_PATH_HANDLERS_H
-#include "kudu/gutil/macros.h"
-#include "kudu/server/webserver.h"
-
#include <string>
#include <sstream>
#include <vector>
+#include "kudu/gutil/macros.h"
+#include "kudu/server/webserver.h"
+
namespace kudu {
class Schema;
@@ -58,12 +58,6 @@ class MasterPathHandlers {
void HandleDumpEntities(const Webserver::WebRequest& req,
std::stringstream* output);
- // Convert location of peers to HTML, indicating the roles
- // of each tablet server in a consensus configuration.
- // This method will display 'locations' in the order given.
- std::string RaftConfigToHtml(const std::vector<TabletReplica>& locations,
- const std::string& tablet_id) const;
-
// Convert the specified TSDescriptor to HTML, adding a link to the
// tablet server's own webserver if specified in 'desc'.
std::string TSDescriptorToHtml(const TSDescriptor& desc,
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/master-test-util.h
----------------------------------------------------------------------
diff --git a/src/kudu/master/master-test-util.h b/src/kudu/master/master-test-util.h
index 1ee264c..da5861b 100644
--- a/src/kudu/master/master-test-util.h
+++ b/src/kudu/master/master-test-util.h
@@ -48,14 +48,7 @@ Status WaitForRunningTabletCount(MiniMaster* mini_master,
req.set_max_returned_locations(expected_count);
RETURN_NOT_OK(mini_master->master()->catalog_manager()->GetTableLocations(&req, resp));
if (resp->tablet_locations_size() >= expected_count) {
- bool is_stale = false;
- for (const TabletLocationsPB& loc : resp->tablet_locations()) {
- is_stale |= loc.stale();
- }
-
- if (!is_stale) {
- return Status::OK();
- }
+ return Status::OK();
}
LOG(INFO) << "Waiting for " << expected_count << " tablets for table "
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/master.proto
----------------------------------------------------------------------
diff --git a/src/kudu/master/master.proto b/src/kudu/master/master.proto
index fbe3506..f575e2e 100644
--- a/src/kudu/master/master.proto
+++ b/src/kudu/master/master.proto
@@ -299,11 +299,8 @@ message TabletLocationsPB {
repeated ReplicaPB replicas = 4;
- // true if the tablet was running but no tablet server has reported it yet.
- // The set of replicas will be the last one that was hosting the tablet.
- // This should happen on Master restart when the request is issued before
- // the TS has the time to notify the Master about the tablets that is hosting.
- required bool stale = 5;
+ // DEPRECATED. Still set by servers, but should be ignored by clients.
+ optional bool DEPRECATED_stale = 5;
}
// Info about a single tablet server, returned to the client as part
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/ts_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/ts_manager.cc b/src/kudu/master/ts_manager.cc
index ee8761f..76d03fb 100644
--- a/src/kudu/master/ts_manager.cc
+++ b/src/kudu/master/ts_manager.cc
@@ -46,7 +46,7 @@ TSManager::~TSManager() {
}
Status TSManager::LookupTS(const NodeInstancePB& instance,
- shared_ptr<TSDescriptor>* ts_desc) {
+ shared_ptr<TSDescriptor>* ts_desc) const {
boost::shared_lock<rw_spinlock> l(lock_);
const shared_ptr<TSDescriptor>* found_ptr =
FindOrNull(servers_by_id_, instance.permanent_uuid());
@@ -64,7 +64,7 @@ Status TSManager::LookupTS(const NodeInstancePB& instance,
}
bool TSManager::LookupTSByUUID(const string& uuid,
- std::shared_ptr<TSDescriptor>* ts_desc) {
+ std::shared_ptr<TSDescriptor>* ts_desc) const {
boost::shared_lock<rw_spinlock> l(lock_);
return FindCopy(servers_by_id_, uuid, ts_desc);
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/master/ts_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/master/ts_manager.h b/src/kudu/master/ts_manager.h
index b2df8ae..52d0b24 100644
--- a/src/kudu/master/ts_manager.h
+++ b/src/kudu/master/ts_manager.h
@@ -58,13 +58,13 @@ class TSManager {
// current instance ID for the TS, then a NotFound status is returned.
// Otherwise, *desc is set and OK is returned.
Status LookupTS(const NodeInstancePB& instance,
- std::shared_ptr<TSDescriptor>* desc);
+ std::shared_ptr<TSDescriptor>* desc) const;
// Lookup the tablet server descriptor for the given UUID.
// Returns false if the TS has never registered.
// Otherwise, *desc is set and returns true.
bool LookupTSByUUID(const std::string& uuid,
- std::shared_ptr<TSDescriptor>* desc);
+ std::shared_ptr<TSDescriptor>* desc) const;
// Register or re-register a tablet server with the manager.
//
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5f7823fe/src/kudu/tools/create-demo-table.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tools/create-demo-table.cc b/src/kudu/tools/create-demo-table.cc
index d43da3f..a6c50e0 100644
--- a/src/kudu/tools/create-demo-table.cc
+++ b/src/kudu/tools/create-demo-table.cc
@@ -47,6 +47,8 @@ using std::vector;
DEFINE_string(master_address, "localhost",
"Comma separated list of master addresses to run against.");
+DEFINE_int32(num_replicas, 3,
+ "Replication count for the created table.");
static const char* const kTwitterTabletId = "twitter";
static const char* const kTPCH1TabletId = "tpch1";
@@ -109,6 +111,7 @@ static int CreateDemoTable(int argc, char** argv) {
gscoped_ptr<KuduTableCreator> table_creator(client->NewTableCreator());
CHECK_OK(table_creator->table_name(table_name)
.schema(&schema)
+ .num_replicas(FLAGS_num_replicas)
.Create());
return 0;
}