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 2017/11/29 07:21:21 UTC
[1/3] kudu git commit: java build: Add script to verify output of
generated JARs
Repository: kudu
Updated Branches:
refs/heads/master d74b98fbb -> 9120cdd1d
java build: Add script to verify output of generated JARs
This script reports any files included in packaging JARs that do not
match a set of patterns in the script. This allows us to define certain
dependencies that we will ship (shaded or otherwise) and, if this is
incorporated into the build process, will immediately alert us as things
change over time.
The script runs automatically as part of the regular CI build.
Change-Id: Ie16bca12dfe2d0267e559c91abe1178d906e3122
Reviewed-on: http://gerrit.cloudera.org:8080/8662
Tested-by: Kudu Jenkins
Reviewed-by: Dan Burkert <da...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/3f658629
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/3f658629
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/3f658629
Branch: refs/heads/master
Commit: 3f658629b49d9ccc04fa8bd7a2e4acc0542994d8
Parents: d74b98f
Author: Mike Percy <mp...@apache.org>
Authored: Mon Nov 27 19:10:30 2017 -0800
Committer: Mike Percy <mp...@apache.org>
Committed: Wed Nov 29 07:19:59 2017 +0000
----------------------------------------------------------------------
build-support/jenkins/build-and-test.sh | 4 +
build-support/verify_jars.pl | 105 +++++++++++++++++++++++++++
2 files changed, 109 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/3f658629/build-support/jenkins/build-and-test.sh
----------------------------------------------------------------------
diff --git a/build-support/jenkins/build-and-test.sh b/build-support/jenkins/build-and-test.sh
index c2b0717..e642657 100755
--- a/build-support/jenkins/build-and-test.sh
+++ b/build-support/jenkins/build-and-test.sh
@@ -395,6 +395,10 @@ if [ "$BUILD_JAVA" == "1" ]; then
fi
fi
+ # Run a script to verify the contents of the JARs to ensure the shading and
+ # packaging is correct.
+ $SOURCE_ROOT/build-support/verify_jars.pl .
+
set +x
popd
fi
http://git-wip-us.apache.org/repos/asf/kudu/blob/3f658629/build-support/verify_jars.pl
----------------------------------------------------------------------
diff --git a/build-support/verify_jars.pl b/build-support/verify_jars.pl
new file mode 100755
index 0000000..803db85
--- /dev/null
+++ b/build-support/verify_jars.pl
@@ -0,0 +1,105 @@
+#!/usr/bin/perl
+################################################################################
+# 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.
+################################################################################
+# This script verifies that the dependencies shipped in the Kudu jars do not
+# change over time without us noticing.
+################################################################################
+use strict;
+use warnings;
+
+# Prefix for shaded classes.
+my $pat_kudu_shaded_prefix = qr{^org/apache/kudu/shaded/};
+
+# Allowed filenames of non-Java files in JARs.
+my $pat_allow_non_java =
+ qr{(?:\.(?:txt|xml|properties|proto|MF)|
+ LICENSE|NOTICE|DEPENDENCIES)$}x;
+
+# Allowed filenames of shaded dependencies in JARs.
+my $pat_allow_kudu_shaded =
+ qr{^org/apache/kudu/shaded/
+ (?:org/jboss/netty|com/google/(?:common|protobuf|thirdparty/publicsuffix)|
+ com/sangupta/murmur)
+ }x;
+
+# Allowed paths of unshaded Kudu dependencies in JARs.
+# Currently, there is no restriction imposed for org.apache.kudu classes.
+my $pat_allow_kudu_unshaded = qr{^org/apache/kudu/.*};
+
+# Allowed paths of unshaded non-Kudu dependencies in JARs.
+my $pat_allow_nonkudu_unshaded = qr{^(?:com/databricks/spark/avro|
+ org/apache/parquet/|
+ org/apache/yetus/|
+ com/stumbleupon/async/)}x;
+
+if (scalar @ARGV != 1) {
+ print STDERR "Usage: $0 <dest_dir>\n";
+ exit 1;
+}
+
+my $dest_dir = $ARGV[0];
+chdir($dest_dir) or die "cannot chdir to destination directory $dest_dir: $!";
+print "Checking jars in directory: " . `pwd`;
+
+chomp(my @jars = `find . -type f -name \*.jar | grep -v build/ | grep -v "tests\.jar" |
+ grep -v original | grep -v sources\.jar | grep -v javadoc\.jar`);
+
+my $num_errors = 0;
+
+foreach my $jar (@jars) {
+ print "> $jar\n";
+ chomp(my @files = `jar tf $jar`);
+ foreach my $file (@files) {
+ # In each case, ensure the files match the expected patterns.
+ if ($file =~ qr{/$}) {
+ # A directory. Skip.
+ } elsif ($file !~ qr{\.class$}) {
+ # Non-Java stuff.
+ if ($file !~ $pat_allow_non_java) {
+ $num_errors++;
+ print "NON_JAVA $file\n";
+ }
+ } elsif ($file =~ $pat_kudu_shaded_prefix) {
+ # Shaded Kudu classes.
+ if ($file !~ $pat_allow_kudu_shaded) {
+ $num_errors++;
+ print "KUDU_SHADED $file\n";
+ }
+ } elsif ($file =~ qr{^org/apache/kudu}) {
+ # Unshaded Kudu classes.
+ if ($file !~ $pat_allow_kudu_unshaded) {
+ $num_errors++;
+ print "KUDU_UNSHADED $file\n";
+ }
+ } else {
+ # Non-Kudu classes.
+ if ($file !~ $pat_allow_nonkudu_unshaded) {
+ $num_errors++;
+ print "NON_KUDU $file\n";
+ }
+ }
+ }
+}
+
+if ($num_errors != 0) {
+ print "Found $num_errors errors.\n";
+ exit 1;
+}
+print "OK.\n";
+exit 0;
[3/3] kudu git commit: KUDU-1097 (patch 5b): kudu tablet
config_change move should use 3-4-3
Posted by mp...@apache.org.
KUDU-1097 (patch 5b): kudu tablet config_change move should use 3-4-3
This patch adds support for 3-4-3 re-replication to the "replica move"
tool.
Change-Id: I259aeaf24f60a05b88c5c3e0e11d603aa75646a5
Reviewed-on: http://gerrit.cloudera.org:8080/8645
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/9120cdd1
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/9120cdd1
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/9120cdd1
Branch: refs/heads/master
Commit: 9120cdd1d288ef6b3e03e2cd8445436712a7d4a9
Parents: eb0718e
Author: Mike Percy <mp...@apache.org>
Authored: Mon Nov 27 02:46:24 2017 -0800
Committer: Mike Percy <mp...@apache.org>
Committed: Wed Nov 29 07:20:56 2017 +0000
----------------------------------------------------------------------
.../integration-tests/cluster_itest_util.cc | 22 +++-
src/kudu/integration-tests/cluster_itest_util.h | 11 +-
.../raft_config_change-itest.cc | 2 +-
src/kudu/tools/kudu-admin-test.cc | 52 +++++++-
src/kudu/tools/tool_action_tablet.cc | 127 ++++++++++++++++---
5 files changed, 185 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/9120cdd1/src/kudu/integration-tests/cluster_itest_util.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/cluster_itest_util.cc b/src/kudu/integration-tests/cluster_itest_util.cc
index 757f6d0..089dba1 100644
--- a/src/kudu/integration-tests/cluster_itest_util.cc
+++ b/src/kudu/integration-tests/cluster_itest_util.cc
@@ -23,6 +23,7 @@
#include <boost/optional/optional.hpp>
#include <glog/logging.h>
#include <glog/stl_logging.h>
+#include <gtest/gtest.h>
#include <rapidjson/document.h>
#include "kudu/client/schema.h"
@@ -58,6 +59,8 @@
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/status.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
namespace kudu {
namespace itest {
@@ -366,7 +369,7 @@ Status WaitUntilNoPendingConfig(const TServerDetails* replica,
SecureShortDebugString(cstate_tmp), s.ToString()));
}
-Status WaitUntilCommittedConfigNumVotersIs(int config_size,
+Status WaitUntilCommittedConfigNumVotersIs(int num_voters,
const TServerDetails* replica,
const std::string& tablet_id,
const MonoDelta& timeout) {
@@ -381,7 +384,7 @@ Status WaitUntilCommittedConfigNumVotersIs(int config_size,
MonoDelta remaining_timeout = deadline - MonoTime::Now();
s = GetConsensusState(replica, tablet_id, remaining_timeout, &cstate);
if (s.ok()) {
- if (CountVoters(cstate.committed_config()) == config_size) {
+ if (CountVoters(cstate.committed_config()) == num_voters) {
return Status::OK();
}
}
@@ -394,10 +397,23 @@ Status WaitUntilCommittedConfigNumVotersIs(int config_size,
}
return Status::TimedOut(Substitute("Number of voters does not equal $0 after waiting for $1. "
"Last consensus state: $2. Last status: $3",
- config_size, timeout.ToString(),
+ num_voters, timeout.ToString(),
SecureShortDebugString(cstate), s.ToString()));
}
+void WaitUntilCommittedConfigNumMembersIs(int num_members,
+ const TServerDetails* replica,
+ const std::string& tablet_id,
+ const MonoDelta& timeout) {
+ MonoTime deadline = MonoTime::Now() + timeout;
+ AssertEventually([&] {
+ ConsensusStatePB cstate;
+ ASSERT_OK(GetConsensusState(replica, tablet_id, deadline - MonoTime::Now(), &cstate));
+ ASSERT_EQ(num_members, cstate.committed_config().peers_size());
+ }, timeout);
+ NO_PENDING_FATALS();
+}
+
Status WaitUntilCommittedConfigOpIdIndexIs(int64_t opid_index,
const TServerDetails* replica,
const std::string& tablet_id,
http://git-wip-us.apache.org/repos/asf/kudu/blob/9120cdd1/src/kudu/integration-tests/cluster_itest_util.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/cluster_itest_util.h b/src/kudu/integration-tests/cluster_itest_util.h
index 64260f1..215a90c 100644
--- a/src/kudu/integration-tests/cluster_itest_util.h
+++ b/src/kudu/integration-tests/cluster_itest_util.h
@@ -159,12 +159,19 @@ Status WaitUntilNoPendingConfig(const TServerDetails* replica,
consensus::ConsensusStatePB* cstate = nullptr);
// Wait until the number of voters in the committed consensus configuration is
-// 'quorum_size', according to the specified replica.
-Status WaitUntilCommittedConfigNumVotersIs(int config_size,
+// 'num_voters', according to the specified replica.
+Status WaitUntilCommittedConfigNumVotersIs(int num_voters,
const TServerDetails* replica,
const std::string& tablet_id,
const MonoDelta& timeout);
+// Wait until the number of voters in the committed consensus configuration is
+// 'num_members', according to the specified replica.
+void WaitUntilCommittedConfigNumMembersIs(int num_members,
+ const TServerDetails* replica,
+ const std::string& tablet_id,
+ const MonoDelta& timeout);
+
// Wait until the opid_index of the committed consensus config on the
// specified tablet is 'opid_index'.
Status WaitUntilCommittedConfigOpIdIndexIs(int64_t opid_index,
http://git-wip-us.apache.org/repos/asf/kudu/blob/9120cdd1/src/kudu/integration-tests/raft_config_change-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/raft_config_change-itest.cc b/src/kudu/integration-tests/raft_config_change-itest.cc
index 8df15a5..4099727 100644
--- a/src/kudu/integration-tests/raft_config_change-itest.cc
+++ b/src/kudu/integration-tests/raft_config_change-itest.cc
@@ -219,7 +219,7 @@ TEST_F(RaftConfigChangeITest, TestNonVoterPromotion) {
});
// Wait for there to be 4 voters in the config.
- ASSERT_OK(WaitUntilCommittedConfigNumVotersIs(/*config_size=*/ 4,
+ ASSERT_OK(WaitUntilCommittedConfigNumVotersIs(/*num_voters=*/ 4,
leader_replica,
tablet_id,
kTimeout));
http://git-wip-us.apache.org/repos/asf/kudu/blob/9120cdd1/src/kudu/tools/kudu-admin-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tools/kudu-admin-test.cc b/src/kudu/tools/kudu-admin-test.cc
index 1506238..7581116 100644
--- a/src/kudu/tools/kudu-admin-test.cc
+++ b/src/kudu/tools/kudu-admin-test.cc
@@ -96,6 +96,13 @@ namespace kudu {
namespace tools {
class AdminCliTest : public tserver::TabletServerIntegrationTestBase {
+ protected:
+ enum EnableKudu1097 {
+ kDisableKudu1097,
+ kEnableKudu1097
+ };
+
+ void DoTestMoveTablet(EnableKudu1097 enable_kudu_1097);
};
// Test config change while running a workload.
@@ -211,10 +218,17 @@ TEST_F(AdminCliTest, TestChangeConfig) {
// 3. Start a workload.
// 4. Using the CLI, move the 3 replicas around the 5 TS.
// 5. Profit!
-TEST_F(AdminCliTest, TestMoveTablet) {
+void AdminCliTest::DoTestMoveTablet(EnableKudu1097 enable_kudu_1097) {
+ const string kKudu1097Flag = "--raft_prepare_replacement_before_eviction=true";
+
FLAGS_num_tablet_servers = 5;
FLAGS_num_replicas = 3;
- NO_FATALS(BuildAndStart());
+
+ vector<string> ts_flags, master_flags;
+ if (enable_kudu_1097) {
+ ts_flags = master_flags = { kKudu1097Flag };
+ }
+ NO_FATALS(BuildAndStart(ts_flags, master_flags));
vector<string> tservers;
AppendKeysFromMap(tablet_servers_, &tservers);
@@ -249,15 +263,28 @@ TEST_F(AdminCliTest, TestMoveTablet) {
for (int i = 0; i < num_moves; i++) {
const string remove = active_tservers.front();
const string add = inactive_tservers.front();
- ASSERT_OK(RunKuduTool({
+ vector<string> tool_command = {
"tablet",
"change_config",
"move_replica",
+ };
+ vector<string> kudu_1097_args = {
+ "--unlock_experimental_flags",
+ kKudu1097Flag,
+ };
+ vector<string> tool_args = {
cluster_->master()->bound_rpc_addr().ToString(),
tablet_id_,
remove,
- add
- }));
+ add,
+ };
+ if (enable_kudu_1097 == kEnableKudu1097) {
+ // Only add these arguments if we running with Kudu 1097 enabled.
+ tool_command.insert(tool_command.end(), kudu_1097_args.begin(), kudu_1097_args.end());
+ }
+ tool_command.insert(tool_command.end(), tool_args.begin(), tool_args.end());
+
+ ASSERT_OK(RunKuduTool(tool_command));
active_tservers.pop_front();
active_tservers.push_back(add);
inactive_tservers.pop_front();
@@ -269,8 +296,13 @@ TEST_F(AdminCliTest, TestMoveTablet) {
for (const string& uuid : active_tservers) {
InsertOrDie(&active_tservers_map, uuid, tablet_servers_[uuid]);
}
- ASSERT_OK(WaitUntilCommittedConfigNumVotersIs(FLAGS_num_replicas, active_tservers_map[add],
+ ASSERT_OK(WaitUntilCommittedConfigNumVotersIs(/*num_voters=*/ FLAGS_num_replicas,
+ active_tservers_map[add],
tablet_id_, MonoDelta::FromSeconds(30)));
+ NO_FATALS(WaitUntilCommittedConfigNumMembersIs(/*num_members=*/ FLAGS_num_replicas,
+ active_tservers_map[add],
+ tablet_id_, MonoDelta::FromSeconds(30)));
+
}
workload.StopAndJoin();
@@ -278,6 +310,14 @@ TEST_F(AdminCliTest, TestMoveTablet) {
NO_FATALS(v.CheckCluster());
}
+TEST_F(AdminCliTest, TestMoveTablet_pre_KUDU_1097) {
+ DoTestMoveTablet(kDisableKudu1097);
+}
+
+TEST_F(AdminCliTest, TestMoveTablet_KUDU_1097) {
+ DoTestMoveTablet(kEnableKudu1097);
+}
+
Status RunUnsafeChangeConfig(const string& tablet_id,
const string& dst_host,
vector<string> peer_uuid_list) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/9120cdd1/src/kudu/tools/tool_action_tablet.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tools/tool_action_tablet.cc b/src/kudu/tools/tool_action_tablet.cc
index 1a27ecc..c6c9209 100644
--- a/src/kudu/tools/tool_action_tablet.cc
+++ b/src/kudu/tools/tool_action_tablet.cc
@@ -28,6 +28,7 @@
#include <boost/optional/optional.hpp>
#include <gflags/gflags.h>
+#include <gflags/gflags_declare.h>
#include <glog/logging.h>
#include "kudu/client/client.h"
@@ -38,6 +39,7 @@
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/opid.pb.h"
#include "kudu/consensus/opid_util.h"
+#include "kudu/gutil/basictypes.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/split.h"
@@ -56,6 +58,7 @@ DEFINE_int64(move_copy_timeout_sec, 600,
"Number of seconds to wait for tablet copy to complete when relocating a tablet");
DEFINE_int64(move_leader_timeout_sec, 30,
"Number of seconds to wait for a leader when relocating a leader tablet");
+DECLARE_bool(raft_prepare_replacement_before_eviction);
namespace kudu {
namespace tools {
@@ -64,6 +67,8 @@ using client::KuduClient;
using client::KuduClientBuilder;
using client::KuduTablet;
using client::KuduTabletServer;
+using consensus::ADD_PEER;
+using consensus::BulkChangeConfigRequestPB;
using consensus::ChangeConfigType;
using consensus::ConsensusServiceProxy;
using consensus::ConsensusStatePB;
@@ -71,6 +76,7 @@ using consensus::GetConsensusStateRequestPB;
using consensus::GetConsensusStateResponsePB;
using consensus::GetLastOpIdRequestPB;
using consensus::GetLastOpIdResponsePB;
+using consensus::MODIFY_PEER;
using consensus::OpId;
using consensus::RaftPeerPB;
using rpc::RpcController;
@@ -385,32 +391,119 @@ Status MoveReplica(const RunnerContext &context) {
const string& from_ts_uuid = FindOrDie(context.required_args, kFromTsUuidArg);
const string& to_ts_uuid = FindOrDie(context.required_args, kToTsUuidArg);
- // Check the tablet is in perfect health and, if so, add the new server.
+ // Check the tablet is in perfect health first.
RETURN_NOT_OK_PREPEND(DoKsckForTablet(master_addresses, tablet_id),
"ksck pre-move health check failed");
- RETURN_NOT_OK(DoChangeConfig(master_addresses, tablet_id, to_ts_uuid,
- RaftPeerPB::VOTER, consensus::ADD_PEER));
- // Wait until the tablet copy completes and the tablet returns to perfect health.
- MonoDelta copy_timeout = MonoDelta::FromSeconds(FLAGS_move_copy_timeout_sec);
- RETURN_NOT_OK_PREPEND(WaitForCleanKsck(master_addresses, tablet_id, copy_timeout),
- "failed waiting for clean ksck after add server");
+ // The pre- KUDU-1097 way of moving a replica involves first adding a new
+ // replica and then evicting the old one.
+ if (!FLAGS_raft_prepare_replacement_before_eviction) {
+ RETURN_NOT_OK(DoChangeConfig(master_addresses, tablet_id, to_ts_uuid,
+ RaftPeerPB::VOTER, consensus::ADD_PEER));
+
+ // Wait until the tablet copy completes and the tablet returns to perfect health.
+ MonoDelta copy_timeout = MonoDelta::FromSeconds(FLAGS_move_copy_timeout_sec);
+ RETURN_NOT_OK_PREPEND(WaitForCleanKsck(master_addresses, tablet_id, copy_timeout),
+ "failed waiting for clean ksck after add server");
+
+ // Finally, remove the chosen replica.
+ // If it is the leader, it will be asked to step down.
+ client::sp::shared_ptr<KuduClient> client;
+ RETURN_NOT_OK(KuduClientBuilder().master_server_addrs(master_addresses).Build(&client));
+ string leader_uuid;
+ HostPort leader_hp;
+ RETURN_NOT_OK(GetTabletLeader(client, tablet_id, &leader_uuid, &leader_hp));
+ if (from_ts_uuid == leader_uuid) {
+ RETURN_NOT_OK_PREPEND(ChangeLeader(client, tablet_id,
+ leader_uuid, leader_hp,
+ MonoDelta::FromSeconds(FLAGS_move_leader_timeout_sec)),
+ "failed changing leadership from the replica to be removed");
+ }
+ return DoChangeConfig(master_addresses, tablet_id, from_ts_uuid,
+ boost::none, consensus::REMOVE_PEER);
+ }
+
+ // In a post- KUDU-1097 world, the procedure to move a replica is to add the
+ // replace=true attribute to the replica to remove while simultaneously
+ // adding the replacement as a non-voter with promote=true.
+ // The following code implements tablet movement in that paradigm.
- // Finally, remove the chosen replica.
- // If it is the leader, it will be asked to step down.
client::sp::shared_ptr<KuduClient> client;
- RETURN_NOT_OK(KuduClientBuilder().master_server_addrs(master_addresses).Build(&client));
+ RETURN_NOT_OK(KuduClientBuilder()
+ .master_server_addrs(master_addresses)
+ .Build(&client));
+
+ BulkChangeConfigRequestPB bulk_req;
+ {
+ auto* change = bulk_req.add_config_changes();
+ change->set_type(MODIFY_PEER);
+ *change->mutable_peer()->mutable_permanent_uuid() = from_ts_uuid;
+ change->mutable_peer()->mutable_attrs()->set_replace(true);
+ }
+ {
+ auto* change = bulk_req.add_config_changes();
+ change->set_type(ADD_PEER);
+ *change->mutable_peer()->mutable_permanent_uuid() = to_ts_uuid;
+ change->mutable_peer()->set_member_type(RaftPeerPB::NON_VOTER);
+ change->mutable_peer()->mutable_attrs()->set_promote(true);
+ HostPort hp;
+ RETURN_NOT_OK(GetRpcAddressForTS(client, to_ts_uuid, &hp));
+ RETURN_NOT_OK(HostPortToPB(hp, change->mutable_peer()->mutable_last_known_addr()));
+ }
+
+ // Find this tablet's leader replica. We need its UUID and RPC address.
string leader_uuid;
HostPort leader_hp;
RETURN_NOT_OK(GetTabletLeader(client, tablet_id, &leader_uuid, &leader_hp));
- if (from_ts_uuid == leader_uuid) {
- RETURN_NOT_OK_PREPEND(ChangeLeader(client, tablet_id,
- leader_uuid, leader_hp,
- MonoDelta::FromSeconds(FLAGS_move_leader_timeout_sec)),
- "failed changing leadership from the replica to be removed");
+ unique_ptr<ConsensusServiceProxy> proxy;
+ RETURN_NOT_OK(BuildProxy(leader_hp.host(), leader_hp.port(), &proxy));
+
+ BulkChangeConfigRequestPB req;
+ consensus::ChangeConfigResponsePB resp;
+ RpcController rpc;
+ rpc.set_timeout(client->default_admin_operation_timeout());
+ bulk_req.set_dest_uuid(leader_uuid);
+ bulk_req.set_tablet_id(tablet_id);
+ RETURN_NOT_OK(proxy->BulkChangeConfig(bulk_req, &resp, &rpc));
+ if (resp.has_error()) {
+ return StatusFromPB(resp.error().status());
+ }
+
+ // Wait until the tablet copy completes and the tablet returns to perfect health.
+ MonoDelta copy_timeout = MonoDelta::FromSeconds(FLAGS_move_copy_timeout_sec);
+ MonoTime start = MonoTime::Now();
+ MonoTime deadline = start + copy_timeout;
+ while (MonoTime::Now() < deadline) {
+ Status s = DoKsckForTablet(master_addresses, tablet_id);
+ if (s.ok()) {
+ // Get the latest leader info.
+ RETURN_NOT_OK(GetTabletLeader(client, tablet_id, &leader_uuid, &leader_hp));
+ RETURN_NOT_OK(BuildProxy(leader_hp.host(), leader_hp.port(), &proxy));
+
+ // Wait until 'from_ts_uuid' is no longer a member of the config.
+ ConsensusStatePB cstate;
+ RETURN_NOT_OK(GetConsensusState(proxy, tablet_id, leader_uuid,
+ client->default_admin_operation_timeout(), &cstate));
+ bool from_ts_uuid_in_config = false; // Is 'from_ts_uuid' still in the config?
+ for (const auto& peer : cstate.committed_config().peers()) {
+ if (peer.permanent_uuid() == from_ts_uuid) {
+ if (peer.attrs().replace() && peer.permanent_uuid() == leader_uuid) {
+ // The leader is the node we intend to remove and is currently
+ // marked for replacement; Make it step down.
+ ignore_result(DoLeaderStepDown(client, tablet_id, leader_uuid, leader_hp));
+ }
+ from_ts_uuid_in_config = true;
+ break;
+ }
+ }
+ if (!from_ts_uuid_in_config) {
+ return Status::OK();
+ }
+ }
+ SleepFor(MonoDelta::FromMilliseconds(50));
}
- return DoChangeConfig(master_addresses, tablet_id, from_ts_uuid,
- boost::none, consensus::REMOVE_PEER);
+ return Status::TimedOut(Substitute("unable to complete tablet replica move after $0",
+ (MonoTime::Now() - start).ToString()));
}
} // anonymous namespace
[2/3] kudu git commit: KUDU-1097 (patch 5a): Implement a bulk config
change API
Posted by mp...@apache.org.
KUDU-1097 (patch 5a): Implement a bulk config change API
A "bulk" config change API is required to support simultaneously
modifying attributes on more than one peer in a config, such as when we
want to move a replica from one location to another.
The "traditional" config change API has been re-routed through the bulk
API so that we get some basic test coverage from the existing tests.
In addition to adding the bulk API, the following changes were made to
the MODIFY_PEER config change API which is currently unused:
* The 'member_type' field is no longer required to be modified by a
MODIFY_PEER config change operation, but *something* still must be
modified to allow it to go through. This is enforced by checking the
"before" and "after" RaftPeerPB instances with MessageDifferencer.
* We now allow to modifying the leader replica's attributes (but still
not its 'member_type' field).
A functional test was added to verify the new functionality of the bulk
change API.
Change-Id: I928a1622d48049c9ad3223b76549a2822bccc30c
Reviewed-on: http://gerrit.cloudera.org:8080/8644
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/eb0718e1
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/eb0718e1
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/eb0718e1
Branch: refs/heads/master
Commit: eb0718e16895b4f851d232f26662b6c16ec51b2f
Parents: 3f65862
Author: Mike Percy <mp...@apache.org>
Authored: Sun Nov 26 23:31:00 2017 -0800
Committer: Mike Percy <mp...@apache.org>
Committed: Wed Nov 29 07:20:45 2017 +0000
----------------------------------------------------------------------
src/kudu/consensus/consensus.proto | 30 +++
src/kudu/consensus/raft_consensus.cc | 251 ++++++++++++-------
src/kudu/consensus/raft_consensus.h | 5 +
.../integration-tests/cluster_itest_util.cc | 55 ++++
src/kudu/integration-tests/cluster_itest_util.h | 17 ++
.../raft_config_change-itest.cc | 229 +++++++++++++++++
.../raft_consensus_nonvoter-itest.cc | 9 +-
src/kudu/tserver/tablet_service.cc | 25 ++
src/kudu/tserver/tablet_service.h | 5 +
9 files changed, 537 insertions(+), 89 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/consensus/consensus.proto
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus.proto b/src/kudu/consensus/consensus.proto
index 6ff4186..1c7f56c 100644
--- a/src/kudu/consensus/consensus.proto
+++ b/src/kudu/consensus/consensus.proto
@@ -151,6 +151,30 @@ message ChangeConfigRequestPB {
optional int64 cas_config_opid_index = 5;
}
+// A config change request that specifies many things to change.
+message BulkChangeConfigRequestPB {
+ // Each sub-request that is being made as part of this bulk change config request.
+ message ConfigChangeItemPB {
+ // The type of config change.
+ optional ChangeConfigType type = 1;
+ // The peer to add, remove, or modify.
+ // The same rules apply here as above in ChangeConfigRequestPB.
+ optional RaftPeerPB peer = 2;
+ }
+
+ // UUID of server this request is addressed to.
+ optional bytes dest_uuid = 1;
+
+ required bytes tablet_id = 2;
+
+ repeated ConfigChangeItemPB config_changes = 3;
+
+ // The OpId index of the committed config to replace.
+ // This parameter serves the same purpose as specified in
+ // ChangeConfigRequestPB.
+ optional int64 cas_config_opid_index = 4;
+}
+
// The configuration change response. If any immediate error occurred
// the 'error' field is set with it, otherwise 'new_configuration' is set.
message ChangeConfigResponsePB {
@@ -528,6 +552,12 @@ service ConsensusService {
// An OK response means the operation was successful.
rpc ChangeConfig(ChangeConfigRequestPB) returns (ChangeConfigResponsePB);
+ // Implements a one-by-one config change interface that allows for multiple
+ // peers to be modified at once, including setting attributes and adding or
+ // removing various peers, as long not more than one voter is added, removed,
+ // or demoted in a single operation.
+ rpc BulkChangeConfig(BulkChangeConfigRequestPB) returns (ChangeConfigResponsePB);
+
// Implements unsafe config change operation for manual recovery use cases.
rpc UnsafeChangeConfig(UnsafeChangeConfigRequestPB) returns (UnsafeChangeConfigResponsePB);
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/consensus/raft_consensus.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.cc b/src/kudu/consensus/raft_consensus.cc
index cdfb0fb..291c146 100644
--- a/src/kudu/consensus/raft_consensus.cc
+++ b/src/kudu/consensus/raft_consensus.cc
@@ -31,6 +31,7 @@
#include <boost/optional/optional.hpp>
#include <gflags/gflags.h>
#include <gflags/gflags_declare.h>
+#include <google/protobuf/util/message_differencer.h>
#include "kudu/common/timestamp.h"
#include "kudu/common/wire_protocol.h"
@@ -144,12 +145,14 @@ METRIC_DEFINE_gauge_int64(tablet, raft_term,
"each time a leader election is started.");
using boost::optional;
+using google::protobuf::util::MessageDifferencer;
using kudu::pb_util::SecureShortDebugString;
using kudu::rpc::PeriodicTimer;
using kudu::tserver::TabletServerErrorPB;
using std::shared_ptr;
using std::string;
using std::unique_ptr;
+using std::unordered_set;
using std::weak_ptr;
using strings::Substitute;
@@ -1635,17 +1638,32 @@ Status RaftConsensus::ChangeConfig(const ChangeConfigRequestPB& req,
"peer", peer_uuid(),
"tablet", options_.tablet_id);
- if (PREDICT_FALSE(!req.has_type())) {
- return Status::InvalidArgument("Must specify 'type' argument to ChangeConfig()",
- SecureShortDebugString(req));
+ BulkChangeConfigRequestPB bulk_req;
+ *bulk_req.mutable_tablet_id() = req.tablet_id();
+
+ if (req.has_dest_uuid()) {
+ *bulk_req.mutable_dest_uuid() = req.dest_uuid();
}
- if (PREDICT_FALSE(!req.has_server())) {
- *error_code = TabletServerErrorPB::INVALID_CONFIG;
- return Status::InvalidArgument("Must specify 'server' argument to ChangeConfig()",
- SecureShortDebugString(req));
+ if (req.has_cas_config_opid_index()) {
+ bulk_req.set_cas_config_opid_index(req.cas_config_opid_index());
+ }
+ auto* change = bulk_req.add_config_changes();
+ if (req.has_type()) {
+ change->set_type(req.type());
+ }
+ if (req.has_server()) {
+ *change->mutable_peer() = req.server();
}
- ChangeConfigType type = req.type();
- const RaftPeerPB& server = req.server();
+
+ return BulkChangeConfig(bulk_req, std::move(client_cb), error_code);
+}
+
+Status RaftConsensus::BulkChangeConfig(const BulkChangeConfigRequestPB& req,
+ StdStatusCallback client_cb,
+ boost::optional<TabletServerErrorPB::Code>* error_code) {
+ TRACE_EVENT2("consensus", "RaftConsensus::BulkChangeConfig",
+ "peer", peer_uuid(),
+ "tablet", options_.tablet_id);
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
@@ -1660,11 +1678,7 @@ Status RaftConsensus::ChangeConfig(const ChangeConfigRequestPB& req,
return Status::IllegalState("Leader has not yet committed an operation in its own term");
}
- if (!server.has_permanent_uuid()) {
- return Status::InvalidArgument("server must have permanent_uuid specified",
- SecureShortDebugString(req));
- }
- RaftConfigPB committed_config = cmeta_->CommittedConfig();
+ const RaftConfigPB committed_config = cmeta_->CommittedConfig();
// Support atomic ChangeConfig requests.
if (req.has_cas_config_opid_index()) {
@@ -1678,82 +1692,151 @@ Status RaftConsensus::ChangeConfig(const ChangeConfigRequestPB& req,
}
}
+ // 'new_config' will be modified in-place and validated before being used
+ // as the new Raft configuration.
RaftConfigPB new_config = committed_config;
- new_config.clear_opid_index();
- const string& server_uuid = server.permanent_uuid();
- switch (type) {
- case ADD_PEER:
- // Ensure the server we are adding is not already a member of the configuration.
- if (IsRaftConfigMember(server_uuid, committed_config)) {
- return Status::InvalidArgument(
- Substitute("Server with UUID $0 is already a member of the config. RaftConfig: $1",
- server_uuid, SecureShortDebugString(committed_config)));
- }
- if (!server.has_member_type()) {
- return Status::InvalidArgument("server must have member_type specified",
- SecureShortDebugString(req));
- }
- if (!server.has_last_known_addr()) {
- return Status::InvalidArgument("server must have last_known_addr specified",
- SecureShortDebugString(req));
- }
- *new_config.add_peers() = server;
- break;
- case REMOVE_PEER:
- if (server_uuid == peer_uuid()) {
- return Status::InvalidArgument(
- Substitute("Cannot remove peer $0 from the config because it is the leader. "
- "Force another leader to be elected to remove this server. "
- "Consensus state: $1",
- server_uuid,
- SecureShortDebugString(cmeta_->ToConsensusStatePB())));
- }
- if (!RemoveFromRaftConfig(&new_config, server_uuid)) {
- return Status::NotFound(
- Substitute("Server with UUID $0 not a member of the config. RaftConfig: $1",
- server_uuid, SecureShortDebugString(committed_config)));
- }
- break;
+ // Enforce the "one by one" config change rules, even with the bulk API.
+ // Keep track of total voters added, including non-voters promoted to
+ // voters, and removed, including voters demoted to non-voters.
+ int num_voters_modified = 0;
- case MODIFY_PEER:
- if (server.member_type() == RaftPeerPB::UNKNOWN_MEMBER_TYPE) {
- return Status::InvalidArgument("Cannot change replica type to UNKNOWN_MEMBER_TYPE");
- }
- if (server_uuid == peer_uuid()) {
- return Status::InvalidArgument(
- Substitute("Cannot modify peer $0 because it is the leader. "
- "Force another leader to be elected to modify this replica. "
- "Consensus state: $1",
- server_uuid,
- SecureShortDebugString(cmeta_->ToConsensusStatePB())));
- }
- RaftPeerPB* peer_pb;
- RETURN_NOT_OK(GetRaftConfigMember(&new_config, server_uuid, &peer_pb));
- // Validate the changes.
- if (ReplicaTypesEqual(*peer_pb, server)) {
- return Status::InvalidArgument("Cannot change replica type to same type");
- }
- peer_pb->set_member_type(server.member_type());
- // Override attributes only if they are explicitly passed in the request.
- // TODO(mpercy): It seems cleaner to bulk-overwrite 'attrs' with
- // whatever is passed into the request, but that would make it more
- // complicated to correctly handle a non-voter that had both its
- // "promote" and "replace" flags set.
- if (server.attrs().has_promote()) {
- peer_pb->mutable_attrs()->set_promote(server.attrs().promote());
- }
- if (server.attrs().has_replace()) {
- peer_pb->mutable_attrs()->set_replace(server.attrs().replace());
+ // A record of the peers being modified so that we can enforce only one
+ // change per peer per request.
+ unordered_set<string> peers_modified;
+
+ for (const auto& item : req.config_changes()) {
+ if (PREDICT_FALSE(!item.has_type())) {
+ *error_code = TabletServerErrorPB::INVALID_CONFIG;
+ return Status::InvalidArgument("Must specify 'type' argument",
+ SecureShortDebugString(req));
+ }
+ if (PREDICT_FALSE(!item.has_peer())) {
+ *error_code = TabletServerErrorPB::INVALID_CONFIG;
+ return Status::InvalidArgument("Must specify 'peer' argument",
+ SecureShortDebugString(req));
+ }
+
+ ChangeConfigType type = item.type();
+ const RaftPeerPB& peer = item.peer();
+
+ if (PREDICT_FALSE(!peer.has_permanent_uuid())) {
+ return Status::InvalidArgument("peer must have permanent_uuid specified",
+ SecureShortDebugString(req));
+ }
+
+ if (!InsertIfNotPresent(&peers_modified, peer.permanent_uuid())) {
+ return Status::InvalidArgument(
+ Substitute("only one change allowed per peer: peer $0 appears more "
+ "than once in the config change request",
+ peer.permanent_uuid()),
+ SecureShortDebugString(req));
+ }
+
+ const string& server_uuid = peer.permanent_uuid();
+ switch (type) {
+ case ADD_PEER:
+ // Ensure the peer we are adding is not already a member of the configuration.
+ if (IsRaftConfigMember(server_uuid, committed_config)) {
+ return Status::InvalidArgument(
+ Substitute("Server with UUID $0 is already a member of the config. RaftConfig: $1",
+ server_uuid, SecureShortDebugString(committed_config)));
+ }
+ if (!peer.has_member_type()) {
+ return Status::InvalidArgument("peer must have member_type specified",
+ SecureShortDebugString(req));
+ }
+ if (!peer.has_last_known_addr()) {
+ return Status::InvalidArgument("peer must have last_known_addr specified",
+ SecureShortDebugString(req));
+ }
+ if (peer.member_type() == RaftPeerPB::VOTER) {
+ num_voters_modified++;
+ }
+ *new_config.add_peers() = peer;
+ break;
+
+ case REMOVE_PEER:
+ if (server_uuid == peer_uuid()) {
+ return Status::InvalidArgument(
+ Substitute("Cannot remove peer $0 from the config because it is the leader. "
+ "Force another leader to be elected to remove this peer. "
+ "Consensus state: $1",
+ server_uuid,
+ SecureShortDebugString(cmeta_->ToConsensusStatePB())));
+ }
+ if (!RemoveFromRaftConfig(&new_config, server_uuid)) {
+ return Status::NotFound(
+ Substitute("Server with UUID $0 not a member of the config. RaftConfig: $1",
+ server_uuid, SecureShortDebugString(committed_config)));
+ }
+ if (IsRaftConfigVoter(server_uuid, committed_config)) {
+ num_voters_modified++;
+ }
+ break;
+
+ case MODIFY_PEER: {
+ RaftPeerPB* modified_peer;
+ RETURN_NOT_OK(GetRaftConfigMember(&new_config, server_uuid, &modified_peer));
+ const RaftPeerPB orig_peer(*modified_peer);
+ // Override 'member_type' and items within 'attrs' only if they are
+ // explicitly passed in the request. At least one field must be
+ // modified to be a valid request.
+ if (peer.has_member_type() && peer.member_type() != modified_peer->member_type()) {
+ if (modified_peer->member_type() == RaftPeerPB::VOTER ||
+ peer.member_type() == RaftPeerPB::VOTER) {
+ // This is a 'member_type' change involving a VOTER, i.e. a
+ // promotion or demotion.
+ num_voters_modified++;
+ }
+ // A leader must be forced to step down before demoting it.
+ if (server_uuid == peer_uuid()) {
+ return Status::InvalidArgument(
+ Substitute("Cannot modify member type of peer $0 because it is the leader. "
+ "Cause another leader to be elected to modify this peer. "
+ "Consensus state: $1",
+ server_uuid,
+ SecureShortDebugString(cmeta_->ToConsensusStatePB())));
+ }
+ modified_peer->set_member_type(peer.member_type());
+ }
+ if (peer.attrs().has_promote()) {
+ modified_peer->mutable_attrs()->set_promote(peer.attrs().promote());
+ }
+ if (peer.attrs().has_replace()) {
+ modified_peer->mutable_attrs()->set_replace(peer.attrs().replace());
+ }
+ // Ensure that MODIFY_PEER actually modified something.
+ if (MessageDifferencer::Equals(orig_peer, *modified_peer)) {
+ return Status::InvalidArgument("must modify a field when calling MODIFY_PEER");
+ }
+ break;
}
- break;
- default:
- return Status::NotSupported(Substitute(
- "$0: unsupported type of configuration change",
- ChangeConfigType_Name(type)));
+ default:
+ return Status::NotSupported(Substitute(
+ "$0: unsupported type of configuration change",
+ ChangeConfigType_Name(type)));
+ }
+ }
+
+ // Don't allow no-op config changes to be committed.
+ if (MessageDifferencer::Equals(committed_config, new_config)) {
+ return Status::InvalidArgument("requested configuration change does not "
+ "actually modify the config",
+ SecureShortDebugString(req));
}
+ // Ensure this wasn't an illegal bulk change.
+ if (num_voters_modified > 1) {
+ return Status::InvalidArgument("it is not safe to modify the VOTER status "
+ "of more than one peer at a time",
+ SecureShortDebugString(req));
+ }
+
+ // We'll assign a new opid_index to this config change.
+ new_config.clear_opid_index();
+
RETURN_NOT_OK(ReplicateConfigChangeUnlocked(
std::move(committed_config), std::move(new_config), std::bind(
&RaftConsensus::MarkDirtyOnSuccess,
@@ -1761,7 +1844,7 @@ Status RaftConsensus::ChangeConfig(const ChangeConfigRequestPB& req,
string("Config change replication complete"),
std::move(client_cb),
std::placeholders::_1)));
- }
+ } // Release lock before signaling request.
peer_manager_->SignalRequest();
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/consensus/raft_consensus.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.h b/src/kudu/consensus/raft_consensus.h
index 3bdb53a..2c438d1 100644
--- a/src/kudu/consensus/raft_consensus.h
+++ b/src/kudu/consensus/raft_consensus.h
@@ -252,6 +252,11 @@ class RaftConsensus : public std::enable_shared_from_this<RaftConsensus>,
StdStatusCallback client_cb,
boost::optional<tserver::TabletServerErrorPB::Code>* error_code);
+ // Implement a BulkChangeConfig() request.
+ Status BulkChangeConfig(const BulkChangeConfigRequestPB& req,
+ StdStatusCallback client_cb,
+ boost::optional<tserver::TabletServerErrorPB::Code>* error_code);
+
// Implement an UnsafeChangeConfig() request.
Status UnsafeChangeConfig(const UnsafeChangeConfigRequestPB& req,
tserver::TabletServerErrorPB::Code* error_code);
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/integration-tests/cluster_itest_util.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/cluster_itest_util.cc b/src/kudu/integration-tests/cluster_itest_util.cc
index ddbe798..757f6d0 100644
--- a/src/kudu/integration-tests/cluster_itest_util.cc
+++ b/src/kudu/integration-tests/cluster_itest_util.cc
@@ -64,6 +64,7 @@ namespace itest {
using client::KuduSchema;
using client::KuduSchemaBuilder;
+using consensus::BulkChangeConfigRequestPB;
using consensus::ChangeConfigRequestPB;
using consensus::ChangeConfigResponsePB;
using consensus::ConsensusStatePB;
@@ -340,6 +341,31 @@ Status GetConsensusState(const TServerDetails* replica,
return Status::OK();
}
+Status WaitUntilNoPendingConfig(const TServerDetails* replica,
+ const std::string& tablet_id,
+ const MonoDelta& timeout,
+ consensus::ConsensusStatePB* cstate) {
+ ConsensusStatePB cstate_tmp;
+ MonoTime start = MonoTime::Now();
+ MonoTime deadline = start + timeout;
+ MonoTime now;
+ Status s;
+ while ((now = MonoTime::Now()) < deadline) {
+ s = GetConsensusState(replica, tablet_id, deadline - now, &cstate_tmp);
+ if (s.ok() && !cstate_tmp.has_pending_config()) {
+ if (cstate) {
+ *cstate = std::move(cstate_tmp);
+ }
+ return Status::OK();
+ }
+ SleepFor(MonoDelta::FromMilliseconds(30));
+ }
+ return Status::TimedOut(Substitute("There is still a pending config after waiting for $0. "
+ "Last consensus state: $1. Last status: $2",
+ (MonoTime::Now() - start).ToString(),
+ SecureShortDebugString(cstate_tmp), s.ToString()));
+}
+
Status WaitUntilCommittedConfigNumVotersIs(int config_size,
const TServerDetails* replica,
const std::string& tablet_id,
@@ -780,6 +806,35 @@ Status ChangeReplicaType(const TServerDetails* leader,
return Status::OK();
}
+Status BulkChangeConfig(const TServerDetails* leader,
+ const std::string& tablet_id,
+ const vector<BulkChangeConfigRequestPB::ConfigChangeItemPB>& changes,
+ const MonoDelta& timeout,
+ const boost::optional<int64_t>& cas_config_index,
+ tserver::TabletServerErrorPB::Code* error_code) {
+ BulkChangeConfigRequestPB req;
+ req.set_dest_uuid(leader->uuid());
+ req.set_tablet_id(tablet_id);
+ if (cas_config_index) {
+ req.set_cas_config_opid_index(*cas_config_index);
+ }
+ for (const auto& change : changes) {
+ *req.add_config_changes() = change;
+ }
+
+ ChangeConfigResponsePB resp;
+ RpcController rpc;
+ rpc.set_timeout(timeout);
+ RETURN_NOT_OK(leader->consensus_proxy->BulkChangeConfig(req, &resp, &rpc));
+ if (resp.has_error()) {
+ if (error_code) {
+ *error_code = resp.error().code();
+ }
+ return StatusFromPB(resp.error().status());
+ }
+ return Status::OK();
+}
+
Status ListTablets(const TServerDetails* ts,
const MonoDelta& timeout,
vector<ListTabletsResponsePB::StatusAndSchemaPB>* tablets) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/integration-tests/cluster_itest_util.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/cluster_itest_util.h b/src/kudu/integration-tests/cluster_itest_util.h
index e9e13cc..64260f1 100644
--- a/src/kudu/integration-tests/cluster_itest_util.h
+++ b/src/kudu/integration-tests/cluster_itest_util.h
@@ -150,6 +150,14 @@ Status GetConsensusState(const TServerDetails* replica,
const MonoDelta& timeout,
consensus::ConsensusStatePB* consensus_state);
+// Wait until there is no longer a pending config on the specified server.
+// If OK is returned, the consensus state is also returned in 'cstate' if it is
+// not set to nullptr.
+Status WaitUntilNoPendingConfig(const TServerDetails* replica,
+ const std::string& tablet_id,
+ const MonoDelta& timeout,
+ consensus::ConsensusStatePB* cstate = nullptr);
+
// Wait until the number of voters in the committed consensus configuration is
// 'quorum_size', according to the specified replica.
Status WaitUntilCommittedConfigNumVotersIs(int config_size,
@@ -296,6 +304,15 @@ Status ChangeReplicaType(const TServerDetails* leader,
const boost::optional<int64_t>& cas_config_index = boost::none,
tserver::TabletServerErrorPB::Code* error_code = nullptr);
+// Convenience function for bulk change config API.
+Status BulkChangeConfig(const TServerDetails* leader,
+ const std::string& tablet_id,
+ const std::vector<consensus::BulkChangeConfigRequestPB
+ ::ConfigChangeItemPB>& changes,
+ const MonoDelta& timeout,
+ const boost::optional<int64_t>& cas_config_index = boost::none,
+ tserver::TabletServerErrorPB::Code* error_code = nullptr);
+
// Get the list of tablets from the remote server.
Status ListTablets(const TServerDetails* ts,
const MonoDelta& timeout,
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/integration-tests/raft_config_change-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/raft_config_change-itest.cc b/src/kudu/integration-tests/raft_config_change-itest.cc
index b0f0cba..8df15a5 100644
--- a/src/kudu/integration-tests/raft_config_change-itest.cc
+++ b/src/kudu/integration-tests/raft_config_change-itest.cc
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+#include <cstdint>
#include <memory>
#include <ostream>
#include <string>
@@ -23,11 +24,15 @@
#include <utility>
#include <vector>
+#include <boost/optional/optional.hpp>
#include <glog/logging.h>
#include <gtest/gtest.h>
+#include "kudu/common/common.pb.h"
+#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/consensus.pb.h"
#include "kudu/consensus/metadata.pb.h"
+#include "kudu/consensus/quorum_util.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/cluster_itest_util.h"
@@ -37,14 +42,22 @@
#include "kudu/master/master.pb.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/util/monotime.h"
+#include "kudu/util/pb_util.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
+using kudu::consensus::ADD_PEER;
using kudu::consensus::COMMITTED_OPID;
+using kudu::consensus::ConsensusStatePB;
+using kudu::consensus::GetRaftConfigMember;
+using kudu::consensus::MODIFY_PEER;
using kudu::consensus::RaftPeerAttrsPB;
using kudu::consensus::RaftPeerPB;
+using kudu::consensus::REMOVE_PEER;
+using kudu::itest::BulkChangeConfig;
using kudu::itest::TServerDetails;
+using kudu::pb_util::SecureShortDebugString;
using std::string;
using std::unordered_set;
using std::vector;
@@ -214,4 +227,220 @@ TEST_F(RaftConfigChangeITest, TestNonVoterPromotion) {
NO_FATALS(cluster_->AssertNoCrashes());
}
+// Functional test for the BulkChangeConfig RPC API.
+TEST_F(RaftConfigChangeITest, TestBulkChangeConfig) {
+ const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
+ const int kNumTabletServers = 4;
+ const int kNumInitialReplicas = 3;
+ NO_FATALS(StartCluster({"--enable_leader_failure_detection=false",
+ "--raft_prepare_replacement_before_eviction=true"},
+ {"--catalog_manager_wait_for_new_tablets_to_elect_leader=false",
+ "--raft_prepare_replacement_before_eviction=true"},
+ kNumTabletServers));
+
+ // Create a table.
+ TestWorkload workload(cluster_.get());
+ workload.Setup();
+
+ ASSERT_OK(inspect_->WaitForReplicaCount(kNumInitialReplicas));
+ master::GetTableLocationsResponsePB table_locations;
+ ASSERT_OK(itest::GetTableLocations(cluster_->master_proxy(), TestWorkload::kDefaultTableName,
+ kTimeout, &table_locations));
+ ASSERT_EQ(1, table_locations.tablet_locations_size()); // Only 1 tablet.
+ ASSERT_EQ(kNumInitialReplicas, table_locations.tablet_locations().begin()->replicas_size());
+ string tablet_id = table_locations.tablet_locations().begin()->tablet_id();
+ unordered_set<int> replica_indexes;
+ for (int i = 0; i < table_locations.tablet_locations().begin()->replicas_size(); i++) {
+ const auto& replica = table_locations.tablet_locations().begin()->replicas(i);
+ int idx = cluster_->tablet_server_index_by_uuid(replica.ts_info().permanent_uuid());
+ replica_indexes.emplace(idx);
+ ASSERT_OK(itest::WaitUntilTabletRunning(ts_map_[cluster_->tablet_server(idx)->uuid()],
+ tablet_id, kTimeout));
+ }
+ ASSERT_EQ(kNumInitialReplicas, replica_indexes.size());
+ const int kLeaderIndex = *replica_indexes.begin();
+ int new_replica_index = -1;
+ for (int i = 0; i < kNumTabletServers; i++) {
+ if (!ContainsKey(replica_indexes, i)) {
+ new_replica_index = i;
+ }
+ }
+ ASSERT_NE(-1, new_replica_index);
+
+ string leader_uuid = cluster_->tablet_server(kLeaderIndex)->uuid();
+ auto* leader_replica = ts_map_[leader_uuid];
+ ASSERT_OK(itest::StartElection(leader_replica, tablet_id, kTimeout));
+ workload.Start();
+ while (workload.rows_inserted() < 100) {
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+ workload.StopAndJoin();
+
+ // We don't want the master interfering with the rest of the test.
+ cluster_->master()->Shutdown();
+
+ struct BulkSpec {
+ consensus::ChangeConfigType change_type;
+ int tserver_index;
+ RaftPeerPB::MemberType member_type;
+ bool replace;
+ bool promote;
+ };
+
+ // Now comes the actual config change testing.
+ auto bulk_change = [&](const vector<BulkSpec>& changes,
+ boost::optional<int64_t> cas_config_index = boost::none) {
+ vector<consensus::BulkChangeConfigRequestPB::ConfigChangeItemPB> changes_pb;
+ for (const auto& chg : changes) {
+ const auto& ts_uuid = cluster_->tablet_server(chg.tserver_index)->uuid();
+ auto* replica = ts_map_[ts_uuid];
+
+ consensus::BulkChangeConfigRequestPB::ConfigChangeItemPB change_pb;
+ change_pb.set_type(chg.change_type);
+
+ RaftPeerPB* peer = change_pb.mutable_peer();
+ peer->set_permanent_uuid(ts_uuid);
+ peer->set_member_type(chg.member_type);
+ peer->mutable_attrs()->set_replace(chg.replace);
+ peer->mutable_attrs()->set_promote(chg.promote);
+ *peer->mutable_last_known_addr() = replica->registration.rpc_addresses(0);
+ changes_pb.emplace_back(std::move(change_pb));
+ }
+
+ LOG(INFO) << "submitting config change with changes:";
+ for (const auto& change_pb : changes_pb) {
+ LOG(INFO) << SecureShortDebugString(change_pb);
+ }
+ return BulkChangeConfig(leader_replica, tablet_id, changes_pb,
+ kTimeout, cas_config_index);
+ };
+
+ // 1) Add a voter. Change config to: V, V, V, V.
+ ASSERT_OK(bulk_change({ { ADD_PEER, new_replica_index, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ false } }));
+ ConsensusStatePB cstate;
+ ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
+ ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size());
+ ASSERT_EQ(kNumTabletServers, CountVoters(cstate.committed_config()));
+
+ // 2) Simultaneous voter modification and attribute modification.
+ // Change config to: V, V, N, V+p.
+ // Note: setting a VOTER's attribute promote=true is meaningless.
+ int replica1_idx = (kLeaderIndex + 1) % kNumTabletServers;
+ int replica2_idx = (kLeaderIndex + 2) % kNumTabletServers;
+ ASSERT_OK(bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::NON_VOTER,
+ /*replace=*/ false, /*promote=*/ false },
+ { MODIFY_PEER, replica2_idx, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ true } }));
+
+ ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
+ ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size());
+ ASSERT_EQ(kNumInitialReplicas, CountVoters(cstate.committed_config()));
+
+ RaftPeerPB* peer;
+ ASSERT_OK(GetRaftConfigMember(cstate.mutable_committed_config(),
+ cluster_->tablet_server(replica2_idx)->uuid(), &peer));
+ ASSERT_EQ(RaftPeerPB::VOTER, peer->member_type());
+ ASSERT_TRUE(peer->attrs().promote()) << SecureShortDebugString(*peer);
+
+ // 3) Single-attribute modification. Change config to: V, V, N+r, V+p.
+ // Note: at the time of writing, if the master is disabled this
+ // configuration will not trigger any actions such as promotion or
+ // eviction.
+ ASSERT_OK(bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::NON_VOTER,
+ /*replace=*/ true, /*promote=*/ false } }));
+
+ ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
+ ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size())
+ << SecureShortDebugString(cstate);
+ ASSERT_EQ(kNumInitialReplicas, CountVoters(cstate.committed_config()))
+ << SecureShortDebugString(cstate);
+
+ ASSERT_OK(GetRaftConfigMember(cstate.mutable_committed_config(),
+ cluster_->tablet_server(replica1_idx)->uuid(), &peer));
+ ASSERT_EQ(RaftPeerPB::NON_VOTER, peer->member_type());
+ ASSERT_TRUE(peer->attrs().replace()) << SecureShortDebugString(*peer);
+
+ // 4) Deny changing config (illegally) from: { V, V, N, V } to: { V, V, V, N }
+ // because that would be both a promotion and a demotion in one step.
+ Status s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ false },
+ { MODIFY_PEER, replica2_idx, RaftPeerPB::NON_VOTER,
+ /*replace=*/ false, /*promote=*/ false } });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "not safe to modify the VOTER status of "
+ "more than one peer at a time");
+
+ // 5) The caller must not be allowed to make the leader a NON_VOTER.
+ s = bulk_change({ { MODIFY_PEER, kLeaderIndex, RaftPeerPB::NON_VOTER,
+ /*replace=*/ false, /*promote=*/ false } });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "Cannot modify member type of peer .* because it is the leader");
+
+ // 6) The 'cas_config_index' flag must be respected, if set.
+ int64_t committed_config_opid_index = cstate.committed_config().opid_index();
+ s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::NON_VOTER,
+ /*replace=*/ false, /*promote=*/ true } }, committed_config_opid_index + 1);
+ ASSERT_TRUE(s.IsIllegalState()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "specified cas_config_opid_index of .* but "
+ "the committed config has opid_index of .*");
+
+ // 7) Evict down to 2 voters. We will evict a voter and a non-voter at once.
+ ASSERT_OK(bulk_change({ { REMOVE_PEER, replica1_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
+ /*replace=*/ false, /*promote=*/ false },
+ { REMOVE_PEER, replica2_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
+ /*replace=*/ false, /*promote=*/ false } }));
+ ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
+ ASSERT_EQ(2, cstate.committed_config().peers_size());
+ ASSERT_EQ(2, CountVoters(cstate.committed_config()));
+
+ // 8) We should reject adding multiple voters at once.
+ s = bulk_change({ { ADD_PEER, replica1_idx, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ false },
+ { ADD_PEER, replica2_idx, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ false } });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "not safe to modify the VOTER status of "
+ "more than one peer at a time");
+
+ // 9) Add them back one at a time so we get to full strength (4 voters) again.
+ auto to_restore = { replica1_idx, replica2_idx };
+ for (auto r : to_restore) {
+ ASSERT_OK(bulk_change({ { ADD_PEER, r, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ false } }));
+ ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
+ }
+ ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size());
+ ASSERT_EQ(kNumTabletServers, CountVoters(cstate.committed_config()));
+
+ // 10) We should reject removing multiple voters at once.
+ s = bulk_change({ { REMOVE_PEER, replica1_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
+ /*replace=*/ false, /*promote=*/ false },
+ { REMOVE_PEER, replica2_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
+ /*replace=*/ false, /*promote=*/ false } });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "not safe to modify the VOTER status of "
+ "more than one peer at a time");
+
+ // 11) Reject no-ops.
+ s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ false } });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "must modify a field when calling MODIFY_PEER");
+
+ // 12) Reject empty bulk change config operations.
+ s = bulk_change({ });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "requested configuration change does not "
+ "actually modify the config");
+
+ // 13) Reject multiple changes to the same peer in a single request.
+ s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
+ /*replace=*/ true, /*promote=*/ false },
+ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
+ /*replace=*/ false, /*promote=*/ true } });
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_MATCHES(s.ToString(), "only one change allowed per peer");
+}
+
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/integration-tests/raft_consensus_nonvoter-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/raft_consensus_nonvoter-itest.cc b/src/kudu/integration-tests/raft_consensus_nonvoter-itest.cc
index 4853e5f..8c37676 100644
--- a/src/kudu/integration-tests/raft_consensus_nonvoter-itest.cc
+++ b/src/kudu/integration-tests/raft_consensus_nonvoter-itest.cc
@@ -670,7 +670,7 @@ TEST_F(RaftConsensusNonVoterITest, PromoteAndDemote) {
tablet_id, replica, kTimeout);
const auto s_str = s.ToString();
ASSERT_TRUE(s.IsInvalidArgument()) << s_str;
- ASSERT_STR_CONTAINS(s_str, "Cannot change replica type to same type");
+ ASSERT_STR_CONTAINS(s_str, "must modify a field when calling MODIFY_PEER");
}
// It should be possible to promote a non-voter to voter.
@@ -683,7 +683,7 @@ TEST_F(RaftConsensusNonVoterITest, PromoteAndDemote) {
tablet_id, replica, kTimeout);
const auto s_str = s.ToString();
ASSERT_TRUE(s.IsInvalidArgument()) << s_str;
- ASSERT_STR_CONTAINS(s_str, "Cannot change replica type to same type");
+ ASSERT_STR_CONTAINS(s_str, "must modify a field when calling MODIFY_PEER");
}
{
@@ -695,7 +695,7 @@ TEST_F(RaftConsensusNonVoterITest, PromoteAndDemote) {
const auto s_demote_str = s_demote.ToString();
ASSERT_TRUE(s_demote.IsInvalidArgument()) << s_demote_str;
ASSERT_STR_MATCHES(s_demote_str,
- "Cannot modify peer .* because it is the leader");
+ "Cannot modify member type of peer .* because it is the leader");
// It should be impossible to promote a leader replica since it's
// already a voter.
@@ -703,8 +703,7 @@ TEST_F(RaftConsensusNonVoterITest, PromoteAndDemote) {
tablet_id, leader, kTimeout);
const auto s_promote_str = s_promote.ToString();
ASSERT_TRUE(s_promote.IsInvalidArgument()) << s_promote_str;
- ASSERT_STR_MATCHES(s_promote_str,
- "Cannot modify peer .* because it is the leader");
+ ASSERT_STR_CONTAINS(s_promote_str, "must modify a field when calling MODIFY_PEER");
}
// Demote the replica back.
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/tserver/tablet_service.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index 37c0f5c..22753d8 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -136,6 +136,7 @@ DECLARE_int32(memory_limit_warn_threshold_percentage);
DECLARE_int32(tablet_history_max_age_sec);
using google::protobuf::RepeatedPtrField;
+using kudu::consensus::BulkChangeConfigRequestPB;
using kudu::consensus::ChangeConfigRequestPB;
using kudu::consensus::ChangeConfigResponsePB;
using kudu::consensus::ConsensusRequestPB;
@@ -1039,6 +1040,30 @@ void ConsensusServiceImpl::ChangeConfig(const ChangeConfigRequestPB* req,
// The success case is handled when the callback fires.
}
+void ConsensusServiceImpl::BulkChangeConfig(const BulkChangeConfigRequestPB* req,
+ ChangeConfigResponsePB* resp,
+ RpcContext* context) {
+ VLOG(1) << "Received BulkChangeConfig RPC: " << SecureDebugString(*req);
+ if (!CheckUuidMatchOrRespond(tablet_manager_, "BulkChangeConfig", req, resp, context)) {
+ return;
+ }
+ scoped_refptr<TabletReplica> replica;
+ if (!LookupRunningTabletReplicaOrRespond(tablet_manager_, req->tablet_id(), resp, context,
+ &replica)) {
+ return;
+ }
+
+ shared_ptr<RaftConsensus> consensus;
+ if (!GetConsensusOrRespond(replica, resp, context, &consensus)) return;
+ boost::optional<TabletServerErrorPB::Code> error_code;
+ Status s = consensus->BulkChangeConfig(*req, BindHandleResponse(req, resp, context), &error_code);
+ if (PREDICT_FALSE(!s.ok())) {
+ HandleErrorResponse(req, resp, context, error_code, s);
+ return;
+ }
+ // The success case is handled when the callback fires.
+}
+
void ConsensusServiceImpl::UnsafeChangeConfig(const UnsafeChangeConfigRequestPB* req,
UnsafeChangeConfigResponsePB* resp,
RpcContext* context) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb0718e1/src/kudu/tserver/tablet_service.h
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_service.h b/src/kudu/tserver/tablet_service.h
index 5f9cc02..cb24cd3 100644
--- a/src/kudu/tserver/tablet_service.h
+++ b/src/kudu/tserver/tablet_service.h
@@ -45,6 +45,7 @@ class ServerBase;
} // namespace server
namespace consensus {
+class BulkChangeConfigRequestPB;
class ChangeConfigRequestPB;
class ChangeConfigResponsePB;
class ConsensusRequestPB;
@@ -200,6 +201,10 @@ class ConsensusServiceImpl : public consensus::ConsensusServiceIf {
consensus::ChangeConfigResponsePB* resp,
rpc::RpcContext* context) OVERRIDE;
+ virtual void BulkChangeConfig(const consensus::BulkChangeConfigRequestPB* req,
+ consensus::ChangeConfigResponsePB* resp,
+ rpc::RpcContext* context) OVERRIDE;
+
virtual void UnsafeChangeConfig(const consensus::UnsafeChangeConfigRequestPB* req,
consensus::UnsafeChangeConfigResponsePB* resp,
rpc::RpcContext* context) OVERRIDE;