You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2024/11/01 00:43:28 UTC
(kudu) branch branch-1.18.x updated (b70a1528c -> 78455172c)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a change to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
from b70a1528c KUDU-613: Fix minor metrics related bug
new d1d76633c Update outdated Gradle command in documentation
new e73def54a KUDU-3589 Wait until masters become voters
new 8c11d588c KUDU-3620 fix heap-use-after-free and undefined behavior in OpDriver
new cb9dd0cd9 [security-itest] stop polluting TEST_TMPDIR with logs
new 098c94b0a [consensus] small clean-up on PendingRounds
new eb09a2428 [dynamic_multi_master-test] reduce the amount of generated data
new 155882b4f [dynamic_multi_master-test] add SKIP_IF_SLOW_NOT_ALLOWED
new 78455172c [java][build] KUDU-3610: kudu-client.jar dependencies
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
java/README.adoc | 2 +-
java/gradle/publishing.gradle | 15 ++++++
java/gradle/shadow.gradle | 50 -------------------
java/kudu-client/build.gradle | 6 ++-
src/kudu/consensus/pending_rounds.cc | 34 +++++++------
src/kudu/consensus/pending_rounds.h | 4 +-
src/kudu/integration-tests/security-itest.cc | 18 +++++++
src/kudu/master/dynamic_multi_master-test.cc | 55 ++++++++------------
src/kudu/mini-cluster/external_mini_cluster.cc | 29 ++++++++++-
src/kudu/mini-cluster/external_mini_cluster.h | 12 +++++
src/kudu/tablet/ops/op_driver.cc | 44 ++++++++++------
src/kudu/tablet/ops/op_driver.h | 28 ++++++-----
src/kudu/tablet/ops/op_tracker-test.cc | 41 ++++++++-------
src/kudu/tablet/ops/op_tracker.cc | 29 +++++------
src/kudu/tablet/ops/op_tracker.h | 26 ++++++----
src/kudu/tablet/tablet_replica-test.cc | 4 +-
src/kudu/tablet/tablet_replica.cc | 69 +++++++++++++++-----------
src/kudu/tablet/tablet_replica.h | 4 +-
src/kudu/tablet/txn_participant-test.cc | 6 ++-
19 files changed, 256 insertions(+), 220 deletions(-)
(kudu) 08/08: [java][build] KUDU-3610: kudu-client.jar dependencies
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 78455172ce0d5d2979b153a8dece3c81ba675d1c
Author: zchovan <zc...@cloudera.com>
AuthorDate: Wed Aug 28 10:52:14 2024 +0200
[java][build] KUDU-3610: kudu-client.jar dependencies
As reported in KUDU-3610 Impala build fails with the newly generated
kudu-client.jar pom.xml.
The issue is in the pom generation itself [1], that is being done by
the 'maven-publish' plugin (which is necessary for gradle 7+ due to
the old publish plugin being deprecated).
This means that the content of the jars are correct, however the
generated pom.xml lists some dependencies in incorrect (maven) scopes.
The fix this change proposes is by editing the generated pom xml
structure, so that the shaded dependencies are removed from the
generated pom. This was previously done by the shadow plugin, however
due to API changes this is no longer possible and has to be done in
maven-publish plugin.
Additional manual testing (building Impala) was carried out that
confirmed that the pom is now correct.
[1] https://discuss.gradle.org/t/maven-publish-plugin-generated-pom-making-dependency-scope-runtime/7494
Change-Id: I607d7c62c585b2aa1bee4ebf7907309e130092da
Reviewed-on: http://gerrit.cloudera.org:8080/21731
Reviewed-by: Alexey Serbin <al...@apache.org>
Tested-by: Alexey Serbin <al...@apache.org>
(cherry picked from commit 9befb1188ec2c8f75bf81f9cd5b17fec2db8c19b)
Reviewed-on: http://gerrit.cloudera.org:8080/21995
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
java/gradle/publishing.gradle | 15 +++++++++++++
java/gradle/shadow.gradle | 50 -------------------------------------------
java/kudu-client/build.gradle | 6 +++++-
3 files changed, 20 insertions(+), 51 deletions(-)
diff --git a/java/gradle/publishing.gradle b/java/gradle/publishing.gradle
index 56206c0db..e78b8f5af 100644
--- a/java/gradle/publishing.gradle
+++ b/java/gradle/publishing.gradle
@@ -68,6 +68,21 @@ publishing {
}
}
}
+
+ // Remove the shaded dependencies from the generated pom.
+ pom.withXml {
+ def apiDeps = project.configurations.api.allDependencies
+ def shadedDeps = project.configurations.shadow.allDependencies
+ def finalDeps = apiDeps - shadedDeps
+ asNode().dependencies.dependency.findAll{
+ finalDeps.findAll{ dep ->
+ dep.name == it.artifactId*.value()[0][0]
+ }
+ }.each() {
+ logger.info "Removing: ${it.artifactId*.value()[0][0]} from pom"
+ assert it.parent().remove(it)
+ }
+ }
}
}
diff --git a/java/gradle/shadow.gradle b/java/gradle/shadow.gradle
index 1f1786331..4cf0b9a58 100644
--- a/java/gradle/shadow.gradle
+++ b/java/gradle/shadow.gradle
@@ -19,7 +19,6 @@
// consistently when a subproject requires shaded artifacts.
import org.gradle.api.internal.artifacts.publish.ArchivePublishArtifact
-
apply plugin: "com.github.johnrengelman.shadow"
knows.enabled = false // Disable the "easter egg" knows task.
@@ -87,19 +86,6 @@ configurations.implementation.extendsFrom(configurations.compileUnshaded)
// We use afterEvaluate to add additional configuration once all the definitions
// in the projects build script have been applied
afterEvaluate {
-// Ensure compileUnshaded dependencies are included in the pom.
- tasks.withType(Upload) { task ->
- task.repositories.each {
- configure(it.pom.scopeMappings) {
- // The priority value is arbitrary.
- addMapping(
- MavenPlugin.COMPILE_PRIORITY,
- configurations.compileUnshaded,
- Conf2ScopeMappingContainer.COMPILE)
- }
- }
- }
-
// Ensure we never shade SLF4J unless we explicitly specify it.
// This is a workaround because in the shadow plugin exclusions from
// parent modules are not respected in modules that use them.
@@ -123,39 +109,3 @@ afterEvaluate {
}
}
-// Remove the shaded dependencies from the generated pom.
-// This hack allows the project to support partially shaded jars,
-// where the shadow plugin by default would remove all compile and runtime dependencies.
-tasks.withType(Upload) {
- def installer = install.repositories.mavenInstaller
- def deployer = uploadArchives.repositories.mavenDeployer
-
- // Handle install and deploy in the same way.
- [installer, deployer]*.pom*.whenConfigured { pom ->
- def filter = shadowJar.getDependencyFilter()
- def configs = shadowJar.getConfigurations()
-
- def shadowDependencies = configs.collectMany {
- // Find all dependencies included in the shaded jar.
- it.resolvedConfiguration.firstLevelModuleDependencies.findAll {
- filter.isIncluded(it)
- }
- }
-
- // Remove the shaded dependencies from the pom.
- shadowDependencies.each { shaded ->
- def depStr = "${shaded.getModuleGroup()}:${shaded.getModuleName()}:${shaded.getModuleVersion()}"
- logger.info "Excluding ${depStr} from the generated pom."
- pom.dependencies.removeAll { dep ->
- dep.groupId == shaded.getModuleGroup() &&
- dep.artifactId == shaded.getModuleName() &&
- dep.version == shaded.getModuleVersion()
- }
- }
-
- // Re-sort the generated maven dependencies to make pom comparisons easier.
- pom.dependencies = pom.dependencies.sort { dep ->
- "$dep.scope:$dep.optional:$dep.groupId:$dep.artifactId"
- }
- }
-}
\ No newline at end of file
diff --git a/java/kudu-client/build.gradle b/java/kudu-client/build.gradle
index 61abb96ba..abcc5c713 100644
--- a/java/kudu-client/build.gradle
+++ b/java/kudu-client/build.gradle
@@ -15,12 +15,16 @@
// specific language governing permissions and limitations
// under the License.
+plugins {
+ id("java-library")
+}
+
apply from: "$rootDir/gradle/shadow.gradle"
apply from: "$rootDir/gradle/benchmarks.gradle"
dependencies {
implementation project(path: ":kudu-proto")
- implementation libs.protobufJava
+ api libs.protobufJava
// Not shaded in the client JAR because it's part of the public API.
compileUnshaded (libs.async) {
(kudu) 03/08: KUDU-3620 fix heap-use-after-free and undefined behavior in OpDriver
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 8c11d588c4be8f02d1babdfc0840b394271cc857
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Wed Oct 16 18:19:39 2024 -0700
KUDU-3620 fix heap-use-after-free and undefined behavior in OpDriver
This patch fixes a long-standing heap-use-after-free issue in OpDriver.
The problem is addressed by supplying a valid reference to OpDriver
objects for callbacks to make sure objects aren't destroyed while
their methods are being invoked.
The reference-counting wrapper for OpDriver objects changed from
scoped_refptr to std::shared_ptr. It allows for passing weak pointers
(std::weak_ptr) instead of raw pointers to callbacks where reference
counting cycle was suspected (left the clarification of the latter as
TODOs), so a valid shared pointer can be constructed before invoking any
of the required OpDriver's methods.
I also updated the signature of OpTracker::GetPendingOps() to get rid of
output parameter.
Before this patch, running the following against Kudu bits built in
ASAN configuration would trigger ASAN and UBSAN warnings once in about
150 runs:
./ts_recovery-itest \
--gtest_filter='DifferentFaultPoints/Kudu969Test.Test/*' \
--gtest_repeat=10000
With this patch, more than 2K iterations of the same succeeded without
producing a single ASAN/UBSAN issue report.
Change-Id: Iadc58f1724bc03373a7e165fd3798b8868dc9d29
Reviewed-on: http://gerrit.cloudera.org:8080/21948
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Zoltan Martonka <zm...@cloudera.com>
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
(cherry picked from commit c51e1354322a4d2cad268de26f254dd52b0c9df1)
Reviewed-on: http://gerrit.cloudera.org:8080/22001
---
src/kudu/tablet/ops/op_driver.cc | 44 +++++++++++++--------
src/kudu/tablet/ops/op_driver.h | 28 ++++++-------
src/kudu/tablet/ops/op_tracker-test.cc | 41 ++++++++++----------
src/kudu/tablet/ops/op_tracker.cc | 29 ++++++--------
src/kudu/tablet/ops/op_tracker.h | 26 +++++++------
src/kudu/tablet/tablet_replica-test.cc | 4 +-
src/kudu/tablet/tablet_replica.cc | 69 +++++++++++++++++++--------------
src/kudu/tablet/tablet_replica.h | 4 +-
src/kudu/tablet/txn_participant-test.cc | 6 ++-
9 files changed, 137 insertions(+), 114 deletions(-)
diff --git a/src/kudu/tablet/ops/op_driver.cc b/src/kudu/tablet/ops/op_driver.cc
index 87c5bbaa1..41cae488a 100644
--- a/src/kudu/tablet/ops/op_driver.cc
+++ b/src/kudu/tablet/ops/op_driver.cc
@@ -62,6 +62,8 @@ using kudu::pb_util::SecureShortDebugString;
using kudu::rpc::RequestIdPB;
using kudu::rpc::ResultTracker;
using std::string;
+using std::shared_ptr;
+using std::weak_ptr;
using std::unique_ptr;
using strings::Substitute;
@@ -141,6 +143,7 @@ Status OpDriver::Init(unique_ptr<Op> op,
}
op_ = std::move(op);
+ auto self = shared_from_this();
if (type == consensus::FOLLOWER) {
std::lock_guard lock(opid_lock_);
op_id_copy_ = op_->state()->op_id();
@@ -171,14 +174,26 @@ Status OpDriver::Init(unique_ptr<Op> op,
unique_ptr<ReplicateMsg> replicate_msg;
op_->NewReplicateMsg(&replicate_msg);
if (consensus_) { // sometimes NULL in tests
- // A raw pointer is required to avoid a refcount cycle.
+ // A weak pointer is required to avoid a refcount cycle.
+ // TODO(aserbin): is the comment above still relevant?
+ weak_ptr<OpDriver> wp(self);
mutable_state()->set_consensus_round(
- consensus_->NewRound(std::move(replicate_msg),
- [this](const Status& s) { this->ReplicationFinished(s); }));
+ consensus_->NewRound(
+ std::move(replicate_msg),
+ [wp = std::move(wp)](const Status& s) {
+ shared_ptr<OpDriver> sp(wp.lock());
+ if (PREDICT_TRUE(sp)) {
+ sp->ReplicationFinished(s);
+ } else {
+ // TODO(aserbin): is this even possible?
+ LOG(DFATAL) << "OpDriver isn't around: has ReplicationFinished() "
+ "been called already?";
+ }
+ }));
}
}
- return op_tracker_->Add(this);
+ return op_tracker_->Add(std::move(self));
}
consensus::OpId OpDriver::GetOpId() {
@@ -226,10 +241,12 @@ void OpDriver::ExecuteAsync() {
s = consensus_->CheckLeadershipAndBindTerm(mutable_state()->consensus_round());
}
- if (s.ok()) {
- s = prepare_pool_token_->Submit([this]() { this->PrepareTask(); });
+ if (PREDICT_TRUE(s.ok())) {
+ // Provide a reference to the object to be able to call its methods
+ // regardless of what happens with other references around.
+ auto self = shared_from_this();
+ s = prepare_pool_token_->Submit([self = std::move(self)]() { self->PrepareTask(); });
}
-
if (PREDICT_FALSE(!s.ok())) {
HandleFailure(s);
}
@@ -501,7 +518,10 @@ Status OpDriver::ApplyAsync() {
}
TRACE_EVENT_FLOW_BEGIN0("op", "ApplyTask", this);
- return apply_pool_->Submit([this]() { this->ApplyTask(); });
+ // Provide a reference to the object to make sure it's safe to call its
+ // methods regardless of what happens with other references around.
+ auto self = shared_from_this();
+ return apply_pool_->Submit([self = std::move(self)]() { self->ApplyTask(); });
}
void OpDriver::ApplyTask() {
@@ -521,10 +541,6 @@ void OpDriver::ApplyTask() {
}
#endif // #if DCHECK_IS_ON() ...
- // We need to ref-count ourself, since FinishApplying() may run very quickly
- // and end up calling Finalize() while we're still in this code.
- scoped_refptr<OpDriver> ref(this);
-
{
CommitMsg* commit_msg;
Status s = op_->Apply(&commit_msg);
@@ -584,16 +600,12 @@ Status OpDriver::CommitWait() {
void OpDriver::Finalize() {
ADOPT_TRACE(trace());
- // TODO: this is an ugly hack so that the Release() call doesn't delete the
- // object while we still hold the lock.
- scoped_refptr<OpDriver> ref(this);
std::lock_guard lock(lock_);
op_->Finish(Op::APPLIED);
mutable_state()->completion_callback()->OpCompleted();
op_tracker_->Release(this);
}
-
std::string OpDriver::StateString(ReplicationState repl_state,
PrepareState prep_state) {
string state_str;
diff --git a/src/kudu/tablet/ops/op_driver.h b/src/kudu/tablet/ops/op_driver.h
index 9fba85303..3da26bd3d 100644
--- a/src/kudu/tablet/ops/op_driver.h
+++ b/src/kudu/tablet/ops/op_driver.h
@@ -28,6 +28,7 @@
#include "kudu/gutil/walltime.h"
#include "kudu/tablet/ops/op.h"
#include "kudu/util/locks.h"
+#include "kudu/util/make_shared.h"
#include "kudu/util/monotime.h"
#include "kudu/util/status.h"
#include "kudu/util/trace.h"
@@ -219,18 +220,10 @@ class OpTracker;
// still be ok, as it would get aborted because the replica wasn't a leader yet (constraints 1/2).
//
// This class is thread safe.
-class OpDriver : public RefCountedThreadSafe<OpDriver> {
-
+class OpDriver : public std::enable_shared_from_this<OpDriver>,
+ public enable_make_shared<OpDriver> {
public:
- // Construct OpDriver. OpDriver does not take ownership
- // of any of the objects pointed to in the constructor's arguments.
- OpDriver(OpTracker* op_tracker,
- consensus::RaftConsensus* consensus,
- log::Log* log,
- ThreadPoolToken* prepare_pool_token,
- ThreadPool* apply_pool,
- OpOrderVerifier* order_verifier,
- MonoTime deadline = MonoTime::Max());
+ ~OpDriver() = default;
// Perform any non-constructor initialization. Sets the op that will be
// executed.
@@ -276,6 +269,17 @@ class OpDriver : public RefCountedThreadSafe<OpDriver> {
Trace* trace() { return trace_.get(); }
+ protected:
+ // Construct OpDriver. OpDriver does not take ownership
+ // of any of the objects pointed to in the constructor's arguments.
+ OpDriver(OpTracker* op_tracker,
+ consensus::RaftConsensus* consensus,
+ log::Log* log,
+ ThreadPoolToken* prepare_pool_token,
+ ThreadPool* apply_pool,
+ OpOrderVerifier* order_verifier,
+ MonoTime deadline = MonoTime::Max());
+
private:
FRIEND_TEST(TabletReplicaTest, TestShuttingDownMVCC);
friend class RefCountedThreadSafe<OpDriver>;
@@ -301,8 +305,6 @@ class OpDriver : public RefCountedThreadSafe<OpDriver> {
PREPARED
};
- ~OpDriver() {}
-
// The task submitted to the prepare threadpool to prepare the op. If Prepare() fails,
// calls HandleFailure.
void PrepareTask();
diff --git a/src/kudu/tablet/ops/op_tracker-test.cc b/src/kudu/tablet/ops/op_tracker-test.cc
index c3c15d938..914e0d873 100644
--- a/src/kudu/tablet/ops/op_tracker-test.cc
+++ b/src/kudu/tablet/ops/op_tracker-test.cc
@@ -18,6 +18,7 @@
#include "kudu/tablet/ops/op_tracker.h"
#include <cstdint>
+#include <iterator>
#include <memory>
#include <ostream>
#include <string>
@@ -108,24 +109,23 @@ class OpTrackerTest : public KuduTest,
void RunOpsThread(CountDownLatch* finish_latch);
Status AddDrivers(int num_drivers,
- vector<scoped_refptr<OpDriver> >* drivers) {
- vector<scoped_refptr<OpDriver> > local_drivers;
+ vector<shared_ptr<OpDriver>>* drivers) {
+ vector<shared_ptr<OpDriver>> local_drivers;
+ local_drivers.reserve(num_drivers);
for (int i = 0; i < num_drivers; i++) {
- scoped_refptr<OpDriver> driver(
- new OpDriver(&tracker_,
- nullptr,
- nullptr,
- nullptr,
- nullptr,
- nullptr));
+ auto driver(OpDriver::make_shared(
+ &tracker_,
+ nullptr,
+ nullptr,
+ nullptr,
+ nullptr,
+ nullptr));
unique_ptr<NoOpOp> op(new NoOpOp(new NoOpOpState));
RETURN_NOT_OK(driver->Init(std::move(op), consensus::LEADER));
- local_drivers.push_back(driver);
+ local_drivers.emplace_back(std::move(driver));
}
- for (const scoped_refptr<OpDriver>& d : local_drivers) {
- drivers->push_back(d);
- }
+ std::move(local_drivers.begin(), local_drivers.end(), std::back_inserter(*drivers));
return Status::OK();
}
@@ -136,13 +136,12 @@ class OpTrackerTest : public KuduTest,
TEST_F(OpTrackerTest, TestGetPending) {
ASSERT_EQ(0, tracker_.GetNumPendingForTests());
- vector<scoped_refptr<OpDriver> > drivers;
+ vector<shared_ptr<OpDriver> > drivers;
ASSERT_OK(AddDrivers(1, &drivers));
- scoped_refptr<OpDriver> driver = drivers[0];
+ shared_ptr<OpDriver>& driver(drivers[0]);
ASSERT_EQ(1, tracker_.GetNumPendingForTests());
- vector<scoped_refptr<OpDriver> > pending_ops;
- tracker_.GetPendingOps(&pending_ops);
+ const auto pending_ops = tracker_.GetPendingOps();
ASSERT_EQ(1, pending_ops.size());
ASSERT_EQ(driver.get(), pending_ops.front().get());
@@ -156,7 +155,7 @@ TEST_F(OpTrackerTest, TestGetPending) {
void OpTrackerTest::RunOpsThread(CountDownLatch* finish_latch) {
const int kNumOps = 100;
// Start a bunch of ops.
- vector<scoped_refptr<OpDriver> > drivers;
+ vector<shared_ptr<OpDriver>> drivers;
ASSERT_OK(AddDrivers(kNumOps, &drivers));
// Wait for the main thread to tell us to proceed.
@@ -167,7 +166,7 @@ void OpTrackerTest::RunOpsThread(CountDownLatch* finish_latch) {
SleepFor(MonoDelta::FromMilliseconds(1));
// Finish all the ops
- for (const scoped_refptr<OpDriver>& driver : drivers) {
+ for (const auto& driver : drivers) {
// And mark the op as failed, which will cause it to unregister itself.
driver->Abort(Status::Aborted(""));
}
@@ -213,7 +212,7 @@ static void CheckMetrics(const scoped_refptr<MetricEntity>& entity,
TEST_F(OpTrackerTest, TestMetrics) {
NO_FATALS(CheckMetrics(entity_, 0, 0, 0, 0));
- vector<scoped_refptr<OpDriver> > drivers;
+ vector<shared_ptr<OpDriver> > drivers;
ASSERT_OK(AddDrivers(3, &drivers));
NO_FATALS(CheckMetrics(entity_, 3, 0, 0, 0));
@@ -251,7 +250,7 @@ TEST_P(OpTrackerTest, TestTooManyOps) {
// carries an empty ReplicateMsg), so we'll just add as many as possible
// and check that when we fail, it's because we've hit the limit.
Status s;
- vector<scoped_refptr<OpDriver>> drivers;
+ vector<shared_ptr<OpDriver>> drivers;
SCOPED_CLEANUP({
for (const auto &d : drivers) {
d->Abort(Status::Aborted(""));
diff --git a/src/kudu/tablet/ops/op_tracker.cc b/src/kudu/tablet/ops/op_tracker.cc
index 058d16429..141965ad2 100644
--- a/src/kudu/tablet/ops/op_tracker.cc
+++ b/src/kudu/tablet/ops/op_tracker.cc
@@ -131,10 +131,6 @@ OpTracker::Metrics::Metrics(const scoped_refptr<MetricEntity>& entity)
#undef GINIT
#undef MINIT
-OpTracker::State::State()
- : memory_footprint(0) {
-}
-
OpTracker::OpTracker() {
}
@@ -145,7 +141,7 @@ OpTracker::~OpTracker() {
#endif
}
-Status OpTracker::Add(OpDriver* driver) {
+Status OpTracker::Add(shared_ptr<OpDriver> driver) {
size_t driver_mem_footprint = driver->state()->request()->SpaceUsedLong();
if (mem_tracker_ && !mem_tracker_->TryConsume(driver_mem_footprint)) {
if (metrics_) {
@@ -175,10 +171,11 @@ Status OpTracker::Add(OpDriver* driver) {
// Cache the op memory footprint so we needn't refer to the request
// again, as it may disappear between now and then.
- State st;
- st.memory_footprint = driver_mem_footprint;
std::lock_guard l(lock_);
- InsertOrDie(&pending_ops_, driver, st);
+ const auto* driver_ptr = driver.get();
+ EmplaceOrDie(&pending_ops_,
+ driver_ptr,
+ State(std::move(driver), driver_mem_footprint));
return Status::OK();
}
@@ -243,14 +240,14 @@ void OpTracker::Release(OpDriver* driver) {
}
}
-void OpTracker::GetPendingOps(
- vector<scoped_refptr<OpDriver> >* pending_out) const {
- DCHECK(pending_out->empty());
+vector<shared_ptr<OpDriver>> OpTracker::GetPendingOps() const {
+ vector<shared_ptr<OpDriver>> ret;
std::lock_guard l(lock_);
- for (const TxnMap::value_type& e : pending_ops_) {
- // Increments refcount of each op.
- pending_out->push_back(e.first);
+ ret.reserve(pending_ops_.size());
+ for (const auto& [_, s]: pending_ops_) {
+ ret.emplace_back(s.driver);
}
+ return ret;
}
int OpTracker::GetNumPendingForTests() const {
@@ -271,9 +268,7 @@ Status OpTracker::WaitForAllToFinish(const MonoDelta& timeout) const {
MonoTime next_log_time = start_time + MonoDelta::FromSeconds(1);
while (1) {
- vector<scoped_refptr<OpDriver> > ops;
- GetPendingOps(&ops);
-
+ const auto ops = GetPendingOps();
if (ops.empty()) {
break;
}
diff --git a/src/kudu/tablet/ops/op_tracker.h b/src/kudu/tablet/ops/op_tracker.h
index 96b05cd39..3f51d38d2 100644
--- a/src/kudu/tablet/ops/op_tracker.h
+++ b/src/kudu/tablet/ops/op_tracker.h
@@ -19,11 +19,11 @@
#include <cstdint>
#include <memory>
#include <unordered_map>
+#include <utility>
#include <vector>
#include "kudu/gutil/macros.h"
#include "kudu/gutil/ref_counted.h"
-#include "kudu/tablet/ops/op_driver.h"
#include "kudu/util/locks.h"
#include "kudu/util/metrics.h"
#include "kudu/util/status.h"
@@ -35,6 +35,8 @@ class MonoDelta;
namespace tablet {
+class OpDriver;
+
// Each TabletReplica has a OpTracker which keeps track of pending ops.
// Each "LeaderOp" will register itself by calling Add().
// It will remove itself by calling Release().
@@ -47,14 +49,14 @@ class OpTracker {
//
// In the event that the tracker's memory limit is exceeded, returns a
// ServiceUnavailable status.
- Status Add(OpDriver* driver);
+ Status Add(std::shared_ptr<OpDriver> driver);
// Removes the op from the pending list.
// Also triggers the deletion of the Op object, if its refcount == 0.
void Release(OpDriver* driver);
- // Populates list of currently-running ops into 'pending_out' vector.
- void GetPendingOps(std::vector<scoped_refptr<OpDriver> >* pending_out) const;
+ // Returns currently-running ops.
+ std::vector<std::shared_ptr<OpDriver>> GetPendingOps() const;
// Returns number of pending ops.
int GetNumPendingForTests() const;
@@ -87,18 +89,20 @@ class OpTracker {
// Per-op state that is tracked along with the op itself.
struct State {
- State();
+ State(std::shared_ptr<OpDriver> driver, int64_t memory_footprint)
+ : driver(std::move(driver)),
+ memory_footprint(memory_footprint) {
+ }
+
+ // The reference to the driver.
+ std::shared_ptr<OpDriver> driver;
// Approximate memory footprint of the op.
- int64_t memory_footprint;
+ const int64_t memory_footprint;
};
// Protected by 'lock_'.
- typedef std::unordered_map<scoped_refptr<OpDriver>,
- State,
- ScopedRefPtrHashFunctor<OpDriver>,
- ScopedRefPtrEqualToFunctor<OpDriver> > TxnMap;
- TxnMap pending_ops_;
+ std::unordered_map<const OpDriver*, State> pending_ops_;
std::unique_ptr<Metrics> metrics_;
diff --git a/src/kudu/tablet/tablet_replica-test.cc b/src/kudu/tablet/tablet_replica-test.cc
index 453309330..be8a83a41 100644
--- a/src/kudu/tablet/tablet_replica-test.cc
+++ b/src/kudu/tablet/tablet_replica-test.cc
@@ -51,7 +51,7 @@
#include "kudu/tablet/lock_manager.h"
#include "kudu/tablet/ops/alter_schema_op.h"
#include "kudu/tablet/ops/op.h"
-#include "kudu/tablet/ops/op_driver.h"
+#include "kudu/tablet/ops/op_driver.h" // IWYU pragma: keep
#include "kudu/tablet/ops/op_tracker.h"
#include "kudu/tablet/ops/write_op.h"
#include "kudu/tablet/tablet.h"
@@ -451,7 +451,7 @@ TEST_F(TabletReplicaTest, TestActiveOpPreventsLogGC) {
&apply_continue,
std::move(op_state)));
- scoped_refptr<OpDriver> driver;
+ shared_ptr<OpDriver> driver;
ASSERT_OK(tablet_replica_->NewLeaderOpDriver(std::move(op),
&driver,
MonoTime::Max()));
diff --git a/src/kudu/tablet/tablet_replica.cc b/src/kudu/tablet/tablet_replica.cc
index 30ec21ae3..bc90e1169 100644
--- a/src/kudu/tablet/tablet_replica.cc
+++ b/src/kudu/tablet/tablet_replica.cc
@@ -149,6 +149,7 @@ using kudu::tserver::ParticipantRequestPB;
using kudu::tserver::TabletServerErrorPB;
using std::deque;
using std::map;
+using std::make_shared;
using std::shared_ptr;
using std::string;
using std::unique_ptr;
@@ -597,7 +598,7 @@ Status TabletReplica::SubmitWrite(unique_ptr<WriteOpState> op_state,
op_state->SetResultTracker(result_tracker_);
unique_ptr<WriteOp> op(new WriteOp(std::move(op_state), consensus::LEADER));
- scoped_refptr<OpDriver> driver;
+ shared_ptr<OpDriver> driver;
RETURN_NOT_OK(NewLeaderOpDriver(std::move(op), &driver, deadline));
driver->ExecuteAsync();
return Status::OK();
@@ -608,7 +609,7 @@ Status TabletReplica::SubmitTxnParticipantOp(std::unique_ptr<ParticipantOpState>
op_state->SetResultTracker(result_tracker_);
unique_ptr<ParticipantOp> op(new ParticipantOp(std::move(op_state), consensus::LEADER));
- scoped_refptr<OpDriver> driver;
+ shared_ptr<OpDriver> driver;
RETURN_NOT_OK(NewLeaderOpDriver(std::move(op), &driver, MonoTime::Max()));
driver->ExecuteAsync();
return Status::OK();
@@ -620,7 +621,7 @@ Status TabletReplica::SubmitAlterSchema(unique_ptr<AlterSchemaOpState> state,
unique_ptr<AlterSchemaOp> op(
new AlterSchemaOp(std::move(state), consensus::LEADER));
- scoped_refptr<OpDriver> driver;
+ shared_ptr<OpDriver> driver;
RETURN_NOT_OK(NewLeaderOpDriver(std::move(op), &driver, deadline));
driver->ExecuteAsync();
return Status::OK();
@@ -708,9 +709,8 @@ string TabletReplica::HumanReadableState() const {
void TabletReplica::GetInFlightOps(Op::TraceType trace_type,
vector<consensus::OpStatusPB>* out) const {
- vector<scoped_refptr<OpDriver> > pending_ops;
- op_tracker_.GetPendingOps(&pending_ops);
- for (const scoped_refptr<OpDriver>& driver : pending_ops) {
+ const auto pending_ops = op_tracker_.GetPendingOps();
+ for (const auto& driver : pending_ops) {
if (driver->state() != nullptr) {
consensus::OpStatusPB status_pb;
status_pb.mutable_op_id()->CopyFrom(driver->GetOpId());
@@ -763,9 +763,8 @@ log::RetentionIndexes TabletReplica::GetRetentionIndexes() const {
<< Substitute("{dur: $0, peers: $1}", ret.for_durability, ret.for_peers);
// Next, interrogate the OpTracker.
- vector<scoped_refptr<OpDriver> > pending_ops;
- op_tracker_.GetPendingOps(&pending_ops);
- for (const scoped_refptr<OpDriver>& driver : pending_ops) {
+ const auto pending_ops = op_tracker_.GetPendingOps();
+ for (const auto& driver : pending_ops) {
OpId op_id = driver->GetOpId();
// A op which doesn't have an opid hasn't been submitted for replication yet and
// thus has no need to anchor the log.
@@ -848,13 +847,23 @@ Status TabletReplica::StartFollowerOp(const scoped_refptr<ConsensusRound>& round
OpState* state = op->state();
state->set_consensus_round(round);
- scoped_refptr<OpDriver> driver;
+ shared_ptr<OpDriver> driver;
RETURN_NOT_OK(NewFollowerOpDriver(std::move(op), &driver));
- // A raw pointer is required to avoid a refcount cycle.
- auto* driver_raw = driver.get();
+ // A weak pointer is required to avoid a refcount cycle.
+ // TODO(aserbin): is the comment above still relevant?
+ std::weak_ptr<OpDriver> wp(driver);
state->consensus_round()->SetConsensusReplicatedCallback(
- [driver_raw](const Status& s) { driver_raw->ReplicationFinished(s); });
+ [wp = std::move(wp)](const Status& s) {
+ shared_ptr<OpDriver> sp(wp.lock());
+ if (PREDICT_TRUE(sp)) {
+ sp->ReplicationFinished(s);
+ } else {
+ // TODO(aserbin): is this even possible?
+ LOG(DFATAL) << "OpDriver isn't around: has ReplicationFinished() "
+ "been called already?";
+ }
+ });
driver->ExecuteAsync();
return Status::OK();
@@ -899,16 +908,16 @@ void TabletReplica::FinishConsensusOnlyRound(ConsensusRound* round) {
}
Status TabletReplica::NewLeaderOpDriver(unique_ptr<Op> op,
- scoped_refptr<OpDriver>* driver,
+ shared_ptr<OpDriver>* driver,
MonoTime deadline) {
- scoped_refptr<OpDriver> op_driver = new OpDriver(
- &op_tracker_,
- consensus_.get(),
- log_.get(),
- prepare_pool_token_.get(),
- apply_pool_,
- &op_order_verifier_,
- deadline);
+ auto op_driver = OpDriver::make_shared(
+ &op_tracker_,
+ consensus_.get(),
+ log_.get(),
+ prepare_pool_token_.get(),
+ apply_pool_,
+ &op_order_verifier_,
+ deadline);
RETURN_NOT_OK(op_driver->Init(std::move(op), consensus::LEADER));
*driver = std::move(op_driver);
@@ -916,14 +925,14 @@ Status TabletReplica::NewLeaderOpDriver(unique_ptr<Op> op,
}
Status TabletReplica::NewFollowerOpDriver(unique_ptr<Op> op,
- scoped_refptr<OpDriver>* driver) {
- scoped_refptr<OpDriver> op_driver = new OpDriver(
- &op_tracker_,
- consensus_.get(),
- log_.get(),
- prepare_pool_token_.get(),
- apply_pool_,
- &op_order_verifier_);
+ shared_ptr<OpDriver>* driver) {
+ auto op_driver = OpDriver::make_shared(
+ &op_tracker_,
+ consensus_.get(),
+ log_.get(),
+ prepare_pool_token_.get(),
+ apply_pool_,
+ &op_order_verifier_);
RETURN_NOT_OK(op_driver->Init(std::move(op), consensus::FOLLOWER));
*driver = std::move(op_driver);
diff --git a/src/kudu/tablet/tablet_replica.h b/src/kudu/tablet/tablet_replica.h
index 2f7faa82f..8b9c69012 100644
--- a/src/kudu/tablet/tablet_replica.h
+++ b/src/kudu/tablet/tablet_replica.h
@@ -315,11 +315,11 @@ class TabletReplica : public RefCountedThreadSafe<TabletReplica>,
// that have timed out while in the prepare queue. For most call sites,
// the deadline is naturally defined by the corresponding RPC.
Status NewLeaderOpDriver(std::unique_ptr<Op> op,
- scoped_refptr<OpDriver>* driver,
+ std::shared_ptr<OpDriver>* driver,
MonoTime deadline);
Status NewFollowerOpDriver(std::unique_ptr<Op> op,
- scoped_refptr<OpDriver>* driver);
+ std::shared_ptr<OpDriver>* driver);
// Tells the tablet's log to garbage collect.
Status RunLogGC();
diff --git a/src/kudu/tablet/txn_participant-test.cc b/src/kudu/tablet/txn_participant-test.cc
index 920fb0039..fea0a9b57 100644
--- a/src/kudu/tablet/txn_participant-test.cc
+++ b/src/kudu/tablet/txn_participant-test.cc
@@ -26,6 +26,7 @@
#include <ostream>
#include <string>
#include <thread>
+#include <type_traits>
#include <unordered_map>
#include <utility>
#include <vector>
@@ -54,7 +55,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/tablet/metadata.pb.h"
#include "kudu/tablet/ops/op.h"
-#include "kudu/tablet/ops/op_driver.h"
+#include "kudu/tablet/ops/op_driver.h" // IWYU pragma: keep
#include "kudu/tablet/ops/op_tracker.h"
#include "kudu/tablet/ops/participant_op.h"
#include "kudu/tablet/tablet-test-util.h"
@@ -86,6 +87,7 @@ using kudu::tserver::WriteRequestPB;
using std::map;
using std::nullopt;
using std::optional;
+using std::shared_ptr;
using std::string;
using std::thread;
using std::unique_ptr;
@@ -997,9 +999,9 @@ TEST_F(TxnParticipantTest, TestActiveParticipantOpsAnchorWALs) {
op_state->set_completion_callback(std::unique_ptr<OpCompletionCallback>(
new LatchOpCompletionCallback<tserver::ParticipantResponsePB>(&latch, &resp)));
- scoped_refptr<OpDriver> driver;
unique_ptr<DelayedParticipantOp> op(
new DelayedParticipantOp(&apply_start, &apply_continue, std::move(op_state)));
+ shared_ptr<OpDriver> driver;
ASSERT_OK(tablet_replica_->NewLeaderOpDriver(std::move(op), &driver, MonoTime::Max()));
driver->ExecuteAsync();
// Wait for the apply to start, indicating that we have persisted and
(kudu) 02/08: KUDU-3589 Wait until masters become voters
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit e73def54a1955c2fa33664fcfd8c4c6614e3e178
Author: Ádám Bakai <ab...@cloudera.com>
AuthorDate: Mon Jul 15 12:25:04 2024 +0200
KUDU-3589 Wait until masters become voters
In SecurityITest.TestNonDefaultPrincipalMultipleMaster test, when the
leader was shut down, it was possible that some of the Raft members
are still in learner mode. This makes it impossible for the cluster to
recover and elect a new leader. To solve this test issue, there is an
explicit wait until masters become voters in the Raft consensus. This
was done by using the already existing VerifyVotersOnAllMasters
function. The function was moved from dynamic_multi_master-test to
external_mini_cluster to be usable from security-itest, too.
Change-Id: I1ea46345680ce506206e9335a147b887fc173b50
Reviewed-on: http://gerrit.cloudera.org:8080/21636
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Alexey Serbin <al...@apache.org>
(cherry picked from commit c5993dbf4f8e706784c690f3d5b38cad75276387)
Reviewed-on: http://gerrit.cloudera.org:8080/21996
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/integration-tests/security-itest.cc | 4 +++
src/kudu/master/dynamic_multi_master-test.cc | 37 +++-----------------------
src/kudu/mini-cluster/external_mini_cluster.cc | 29 ++++++++++++++++++--
src/kudu/mini-cluster/external_mini_cluster.h | 12 +++++++++
4 files changed, 46 insertions(+), 36 deletions(-)
diff --git a/src/kudu/integration-tests/security-itest.cc b/src/kudu/integration-tests/security-itest.cc
index 78452461b..f0509f920 100644
--- a/src/kudu/integration-tests/security-itest.cc
+++ b/src/kudu/integration-tests/security-itest.cc
@@ -906,6 +906,10 @@ TEST_F(SecurityITest, TestNonDefaultPrincipalMultipleMaster) {
ASSERT_OK(cluster_->AddMaster());
ASSERT_OK(cluster_->master(2)->WaitForCatalogManager());
+ ASSERT_EVENTUALLY([&] {
+ ASSERT_OK(cluster_->VerifyVotersOnAllMasters(3));
+ });
+
// Tablet servers need to be restarted in order to update their master list.
for (int i = 0; i < cluster_->num_tablet_servers(); i++) {
cluster_->tablet_server(i)->Shutdown();
diff --git a/src/kudu/master/dynamic_multi_master-test.cc b/src/kudu/master/dynamic_multi_master-test.cc
index ed72c3028..1479ef736 100644
--- a/src/kudu/master/dynamic_multi_master-test.cc
+++ b/src/kudu/master/dynamic_multi_master-test.cc
@@ -203,37 +203,6 @@ Status RunListMasters(ListMastersResponsePB* resp, ExternalMiniCluster* cluster)
return RunLeaderMasterRPC(list_masters, cluster);
}
-// Verify the 'cluster' contains 'num_masters' on all masters, and that they
-// are all voters. Returns an error if the expected state is not present.
-//
-// This should be used instead of VerifyVoterMastersForCluster() if it is
-// required that all masters have accepted the config change. E.g. tests that
-// restart a cluster after adding a master should verify that all masters agree
-// before restarting, in case lagging masters start up with stale configs.
-//
-// TODO(awong): we should be more robust to starting up with mismatched on-disk
-// configs, if we can help it.
-Status VerifyVotersOnAllMasters(int num_masters, ExternalMiniCluster* cluster) {
- for (int i = 0; i < cluster->num_masters(); i++) {
- ListMastersResponsePB resp;
- ListMastersRequestPB req;
- RpcController rpc;
- RETURN_NOT_OK(cluster->master_proxy(i)->ListMasters(req, &resp, &rpc));
- if (num_masters != resp.masters_size()) {
- return Status::IllegalState(Substitute("expected $0 masters but got $1",
- num_masters, resp.masters_size()));
- }
- for (const auto& master : resp.masters()) {
- if ((master.role() != RaftPeerPB::LEADER && master.role() != RaftPeerPB::FOLLOWER) ||
- master.member_type() != RaftPeerPB::VOTER ||
- master.registration().rpc_addresses_size() != 1) {
- return Status::IllegalState(Substitute("bad master: $0", SecureShortDebugString(master)));
- }
- }
- }
- return Status::OK();
-}
-
// Verify the ExternalMiniCluster 'cluster' contains 'num_masters' overall and
// are all VOTERS. Populates the new master addresses in 'master_hps', if not
// nullptr. Returns an error if the expected state is not present.
@@ -1699,7 +1668,7 @@ TEST_F(AutoAddMasterTest, TestAddWithOnGoingDdl) {
ASSERT_OK(new_master->WaitForCatalogManager());
num_masters++;
ASSERT_EVENTUALLY([&] {
- ASSERT_OK(VerifyVotersOnAllMasters(num_masters, cluster_.get()));
+ ASSERT_OK(cluster_->VerifyVotersOnAllMasters(num_masters));
});
{
std::lock_guard l(master_addrs_lock);
@@ -1784,7 +1753,7 @@ TEST_F(AutoAddMasterTest, TestAddNewMaster) {
ASSERT_OK(peer->WaitForCatalogManager());
auto expected_num_masters = ++idx;
ASSERT_EVENTUALLY([&] {
- ASSERT_OK(VerifyVotersOnAllMasters(expected_num_masters, cluster_.get()));
+ ASSERT_OK(cluster_->VerifyVotersOnAllMasters(expected_num_masters));
});
NO_FATALS(cluster_->AssertNoCrashes());
}
@@ -1811,7 +1780,7 @@ TEST_F(MinidumpTest, TestAddNewMasterMinidumpsEnabled) {
ASSERT_OK(peer->WaitForCatalogManager());
auto expected_num_masters = ++idx;
ASSERT_EVENTUALLY([&] {
- ASSERT_OK(VerifyVotersOnAllMasters(expected_num_masters, cluster_.get()));
+ ASSERT_OK(cluster_->VerifyVotersOnAllMasters(expected_num_masters));
});
NO_FATALS(cluster_->AssertNoCrashes());
}
diff --git a/src/kudu/mini-cluster/external_mini_cluster.cc b/src/kudu/mini-cluster/external_mini_cluster.cc
index a935e17c1..a8044e33c 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.cc
+++ b/src/kudu/mini-cluster/external_mini_cluster.cc
@@ -35,17 +35,20 @@
#include "kudu/client/client.h"
#include "kudu/client/master_rpc.h"
+#include "kudu/consensus/metadata.pb.h"
#include "kudu/fs/default_key_provider.h"
#include "kudu/fs/fs.pb.h"
#include "kudu/fs/key_provider.h"
#include "kudu/postgres/mini_postgres.h"
#include "kudu/ranger-kms/mini_ranger_kms.h"
#include "kudu/rpc/rpc_header.pb.h"
+#include "kudu/tablet/metadata.pb.h"
#if !defined(NO_CHRONY)
#include "kudu/clock/test/mini_chronyd.h"
#endif
#include "kudu/common/wire_protocol.h"
#include "kudu/common/wire_protocol.pb.h"
+#include "kudu/consensus/consensus.proxy.h"
#include "kudu/gutil/basictypes.h"
#include "kudu/gutil/strings/escaping.h"
#include "kudu/gutil/strings/join.h"
@@ -63,12 +66,10 @@
#include "kudu/security/test/mini_kdc.h"
#include "kudu/server/server_base.pb.h"
#include "kudu/server/server_base.proxy.h"
-#include "kudu/tablet/metadata.pb.h"
#include "kudu/tablet/tablet.pb.h"
#include "kudu/tserver/tserver.pb.h"
#include "kudu/tserver/tserver_admin.proxy.h"
#include "kudu/tserver/tserver_service.proxy.h"
-#include "kudu/consensus/consensus.proxy.h"
#include "kudu/util/async_util.h"
#include "kudu/util/env.h"
#include "kudu/util/env_util.h"
@@ -94,8 +95,11 @@ using kudu::client::internal::ConnectToClusterRpc;
#if !defined(NO_CHRONY)
using kudu::clock::MiniChronyd;
#endif
+using kudu::master::ListMastersRequestPB;
+using kudu::master::ListMastersResponsePB;
using kudu::master::ListTablesRequestPB;
using kudu::master::ListTablesResponsePB;
+using kudu::consensus::RaftPeerPB;
using kudu::master::MasterServiceProxy;
using kudu::pb_util::SecureDebugString;
using kudu::pb_util::SecureShortDebugString;
@@ -996,6 +1000,27 @@ Status ExternalMiniCluster::WaitForTabletsRunning(
return Status::TimedOut(SecureDebugString(resp));
}
+Status ExternalMiniCluster::VerifyVotersOnAllMasters(int num_masters) {
+ for (int i = 0; i < this->num_masters(); i++) {
+ ListMastersResponsePB resp;
+ ListMastersRequestPB req;
+ RpcController rpc;
+ RETURN_NOT_OK(master_proxy(i)->ListMasters(req, &resp, &rpc));
+ if (num_masters != resp.masters_size()) {
+ return Status::IllegalState(Substitute("expected $0 masters but got $1",
+ num_masters, resp.masters_size()));
+ }
+ for (const auto& master : resp.masters()) {
+ if ((master.role() != RaftPeerPB::LEADER && master.role() != RaftPeerPB::FOLLOWER) ||
+ master.member_type() != RaftPeerPB::VOTER ||
+ master.registration().rpc_addresses_size() != 1) {
+ return Status::IllegalState(Substitute("bad master: $0", SecureShortDebugString(master)));
+ }
+ }
+ }
+ return Status::OK();
+}
+
ExternalMaster* ExternalMiniCluster::leader_master() {
int idx = 0;
Status s = GetLeaderMasterIndex(&idx);
diff --git a/src/kudu/mini-cluster/external_mini_cluster.h b/src/kudu/mini-cluster/external_mini_cluster.h
index b60dcc4d1..449f0b5c2 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.h
+++ b/src/kudu/mini-cluster/external_mini_cluster.h
@@ -613,6 +613,18 @@ class ExternalMiniCluster : public MiniCluster {
size_t idx,
scoped_refptr<ExternalMaster>* master);
+ // Verify the cluster contains 'num_masters' on all masters, and that they
+ // are all voters. Returns an error if the expected state is not present.
+ //
+ // This should be used if it is required that all masters have accepted the config
+ // change. E.g. tests that restart a cluster after adding a master should verify
+ // that all masters agree before restarting, in case lagging masters start up with
+ // stale configs.
+ //
+ // TODO(awong): we should be more robust to starting up with mismatched on-disk
+ // configs, if we can help it.
+ Status VerifyVotersOnAllMasters(int num_masters);
+
private:
Status StartMasters();
(kudu) 06/08: [dynamic_multi_master-test] reduce the amount of generated data
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit eb09a2428b8cd89bc4108922510a5d17e536291c
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue Oct 8 17:07:25 2024 -0700
[dynamic_multi_master-test] reduce the amount of generated data
When I ran all C++ tests via ctest on one node having about 25GiBytes of
available disk space, it unexpectedly ran out of space. The culprit was
dynamic_multi_master-test: the scenarios from the suite keep their test
workloads running while waiting for events that might take a long time
to happen, especially on slow or very busy machines.
This patch updates the configuration of the TestWorkload objects to use
a single writer thread (the default is four) in the TestBasicAddition
scenario of the ParameterizedAutoAddMasterTest test. In addition,
a bit of extra delay is now added between writing batches of test rows.
Change-Id: Id0f0005b8f3384ad05eee6bf65e17a0e4ed289f1
Reviewed-on: http://gerrit.cloudera.org:8080/21935
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Marton Greber <gr...@gmail.com>
Reviewed-by: Attila Bukor <ab...@apache.org>
(cherry picked from commit dbb6d5320ad60016355d0c63b6dae1650ba4e414)
Reviewed-on: http://gerrit.cloudera.org:8080/22000
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/master/dynamic_multi_master-test.cc | 6 ++++++
1 file changed, 6 insertions(+)
diff --git a/src/kudu/master/dynamic_multi_master-test.cc b/src/kudu/master/dynamic_multi_master-test.cc
index 1479ef736..beb825789 100644
--- a/src/kudu/master/dynamic_multi_master-test.cc
+++ b/src/kudu/master/dynamic_multi_master-test.cc
@@ -1797,6 +1797,12 @@ class ParameterizedAutoAddMasterTest : public AutoAddMasterTest,
TEST_P(ParameterizedAutoAddMasterTest, TestBasicAddition) {
TestWorkload w(cluster_.get());
w.set_num_replicas(1);
+ // Using one thread and injecting delays between write batches to reduce
+ // the amount of data accumulated under the tablet server's WAL and data
+ // directories since this scenario might run for quite a long time,
+ // especially on slow machines.
+ w.set_num_write_threads(1);
+ w.set_write_interval_millis(5);
w.Setup();
w.Start();
int num_masters = args_.orig_num_masters;
(kudu) 04/08: [security-itest] stop polluting TEST_TMPDIR with logs
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit cb9dd0cd9302463f2faae0fd67875fff012480e1
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue Oct 8 15:24:42 2024 -0700
[security-itest] stop polluting TEST_TMPDIR with logs
By default, KuduTest-based scenarios have their FLAGS_log_dir pointed to
the TEST_TMPDIR directory, so if master/tserver is run using the
"kudu master run"/"kudu tserver run" CLI tool, the generated log files
aren't cleaned up. This patch addresses the issue by customizing the
--log_dir flag to keep all the logs under the test's directory which
is automatically removed once the scenario successfully completes.
This patch addresses KUDU-3615, at least partially.
Change-Id: Ib70b8a477beeab82af0ee6b4099114d5482f12cf
Reviewed-on: http://gerrit.cloudera.org:8080/21909
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: KeDeng <kd...@gmail.com>
(cherry picked from commit 446861c68cb3cc5e8ef8caaadf7d8eefc2ec1387)
Reviewed-on: http://gerrit.cloudera.org:8080/21999
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/integration-tests/security-itest.cc | 14 ++++++++++++++
1 file changed, 14 insertions(+)
diff --git a/src/kudu/integration-tests/security-itest.cc b/src/kudu/integration-tests/security-itest.cc
index f0509f920..b0ed90152 100644
--- a/src/kudu/integration-tests/security-itest.cc
+++ b/src/kudu/integration-tests/security-itest.cc
@@ -827,9 +827,16 @@ TEST_F(SecurityITest, TestWorldReadableKeytab) {
NO_FATALS(CreateWorldReadableFile(credentials_name));
string binary = "kudu";
NO_FATALS(GetFullBinaryPath(&binary));
+
+ // By default, KuduTest-based scenarios have their FLAGS_log_dir pointing to
+ // the TEST_TMPDIR directory, and the generated log files aren't cleaned up.
+ // So, customize the --log_dir flag to keep all the logs under the test's
+ // directory which is automatically removed once the scenario successfully
+ // completes.
const vector<string> argv = { binary,
"master",
"run",
+ Substitute("--log_dir=$0", GetTestDataDirectory()),
Substitute("--keytab_file=$0", credentials_name) };
string stderr;
Status s = Subprocess::Call(argv, "", nullptr, &stderr);
@@ -843,9 +850,16 @@ TEST_F(SecurityITest, TestWorldReadablePrivateKey) {
NO_FATALS(CreateWorldReadableFile(credentials_name));
string binary = "kudu";
NO_FATALS(GetFullBinaryPath(&binary));
+
+ // By default, KuduTest-based scenarios have their FLAGS_log_dir pointing to
+ // the TEST_TMPDIR directory, and the generated log files aren't cleaned up.
+ // So, customize the --log_dir flag to keep all the logs under the test's
+ // directory which is automatically removed once the scenario successfully
+ // completes.
const vector<string> argv = { binary,
"master",
"run",
+ Substitute("--log_dir=$0", GetTestDataDirectory()),
"--unlock_experimental_flags",
Substitute("--rpc_private_key_file=$0", credentials_name),
"--rpc_certificate_file=fake_file",
(kudu) 07/08: [dynamic_multi_master-test] add SKIP_IF_SLOW_NOT_ALLOWED
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 155882b4f43f962e604ae18624db328e1263e1d3
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Wed Oct 16 11:01:07 2024 -0700
[dynamic_multi_master-test] add SKIP_IF_SLOW_NOT_ALLOWED
This patch adds SKIP_IF_SLOW_NOT_ALLOWED() macro into appropriate
scenarios in the dynamic_multi_master-test. The criteria to tag a
scenario as 'slow' was the evidence of requiring more than 3 seconds
to complete on a more-or-less contemporary machine (DEBUG build).
The motivation for this changelist is to address the review feedback
for one of the prior patches [1].
[1] review URL: http://gerrit.cloudera.org:8080/21935
gerrit ID : Id0f0005b8f3384ad05eee6bf65e17a0e4ed289f1
git hash : dbb6d5320ad60016355d0c63b6dae1650ba4e414
Change-Id: Ib255caf1ae33ebf9df9e71d9ea784d6e7d350ed9
Reviewed-on: http://gerrit.cloudera.org:8080/21943
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Marton Greber <gr...@gmail.com>
(cherry picked from commit e0c7a37b9ded1b0052487a05fc6a65a8cb0efbbd)
Reviewed-on: http://gerrit.cloudera.org:8080/22003
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/master/dynamic_multi_master-test.cc | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/src/kudu/master/dynamic_multi_master-test.cc b/src/kudu/master/dynamic_multi_master-test.cc
index beb825789..d3c9e0a55 100644
--- a/src/kudu/master/dynamic_multi_master-test.cc
+++ b/src/kudu/master/dynamic_multi_master-test.cc
@@ -916,6 +916,8 @@ INSTANTIATE_TEST_SUITE_P(, ParameterizedRemoveMasterTest,
// Tests removing a non-leader master from the cluster.
TEST_P(ParameterizedRemoveMasterTest, TestRemoveMaster) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
NO_FATALS(StartCluster({// Keeping RPC timeouts short to quickly detect downed servers.
// This will put the health status into an UNKNOWN state until the point
// where they are considered FAILED.
@@ -1186,6 +1188,8 @@ TEST_F(DynamicMultiMasterTest, TestRemoveMasterFeatureFlagNotSpecified) {
// Test that attempts to request a non-leader master to add a new master.
TEST_F(DynamicMultiMasterTest, TestAddMasterToNonLeader) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
@@ -1225,6 +1229,8 @@ TEST_F(DynamicMultiMasterTest, TestAddMasterToNonLeader) {
// Test that attempts to request a non-leader master to remove a master.
TEST_F(DynamicMultiMasterTest, TestRemoveMasterToNonLeader) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
@@ -1295,6 +1301,8 @@ TEST_F(DynamicMultiMasterTest, TestAddMasterMissingAndIncorrectAddress) {
// Test that attempts to remove a master with missing master address and a non-existent
// hostname.
TEST_F(DynamicMultiMasterTest, TestRemoveMasterMissingAndIncorrectHostname) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
@@ -1533,6 +1541,8 @@ TEST_F(AutoAddMasterTest, TestSomeMastersUnreachable) {
// Test if we fail to replicate the AddMaster request.
TEST_F(AutoAddMasterTest, TestFailWithoutReplicatingAddMaster) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
// Make master followers unable to accept updates, including config changes.
// We'll set this for all masters including leaders for simplicity.
for (int i = 0; i < cluster_->num_masters(); i++) {
@@ -1795,6 +1805,8 @@ class ParameterizedAutoAddMasterTest : public AutoAddMasterTest,
};
TEST_P(ParameterizedAutoAddMasterTest, TestBasicAddition) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
TestWorkload w(cluster_.get());
w.set_num_replicas(1);
// Using one thread and injecting delays between write batches to reduce
(kudu) 01/08: Update outdated Gradle command in documentation
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit d1d76633c92b997d64276161e99c2265c54133be
Author: Marton Greber <gr...@gmail.com>
AuthorDate: Thu Oct 17 15:59:32 2024 +0200
Update outdated Gradle command in documentation
Replaced the incorrect ./gradlew install command with the updated
./gradlew publishToMavenLocal in the documentation. This change
reflects the current process for publishing to the local Maven repo.
Thanks to Zoltan Chovan for diggin out the up to date command!
Change-Id: I4dec2d92cea0035a368c882cc807b988d2fd0250
Reviewed-on: http://gerrit.cloudera.org:8080/21951
Tested-by: Kudu Jenkins
Reviewed-by: Zoltan Chovan <zc...@cloudera.com>
Reviewed-by: Attila Bukor <ab...@apache.org>
(cherry picked from commit 1c72398b47ebf830f76fa51719e97eadcf010166)
Reviewed-on: http://gerrit.cloudera.org:8080/21997
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
java/README.adoc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/java/README.adoc b/java/README.adoc
index e37846fb7..d5eda43b4 100644
--- a/java/README.adoc
+++ b/java/README.adoc
@@ -125,7 +125,7 @@ to rule out any build issues.
[source,bash]
----
-$ ./gradlew install
+$ ./gradlew publishToMavenLocal
----
=== Checking for Dependency Updates
(kudu) 05/08: [consensus] small clean-up on PendingRounds
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.18.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 098c94b0a6d9cdcfbe729502bcc0c43f57eaa064
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Wed Oct 2 15:33:44 2024 -0700
[consensus] small clean-up on PendingRounds
I was looking at the code around PendingRounds while troubleshooting
a race condition, and found that there is room for micro-optimisations
and syntactic sugar in the PendingRounds implementation.
This patch doesn't contain any functional modifications.
Change-Id: Iaa5b66d8a2cf179f43141de8d640251e1a2dc80f
Reviewed-on: http://gerrit.cloudera.org:8080/21878
Tested-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
(cherry picked from commit b6320038376b6fb35bfd1f1ac74323168df0884a)
Reviewed-on: http://gerrit.cloudera.org:8080/21998
---
src/kudu/consensus/pending_rounds.cc | 34 ++++++++++++++++++----------------
src/kudu/consensus/pending_rounds.h | 4 ++--
2 files changed, 20 insertions(+), 18 deletions(-)
diff --git a/src/kudu/consensus/pending_rounds.cc b/src/kudu/consensus/pending_rounds.cc
index 80a5d8660..3e1c25302 100644
--- a/src/kudu/consensus/pending_rounds.cc
+++ b/src/kudu/consensus/pending_rounds.cc
@@ -62,12 +62,13 @@ Status PendingRounds::CancelPendingOps() {
LOG_WITH_PREFIX(INFO) << "Trying to abort " << pending_ops_.size()
<< " pending ops.";
// Abort ops in reverse index order. See KUDU-1678.
+ static const auto kAbortedStatus = Status::Aborted("Op aborted");
for (auto op = pending_ops_.crbegin(); op != pending_ops_.crend(); ++op) {
const scoped_refptr<ConsensusRound>& round = op->second;
// We cancel only ops whose applies have not yet been triggered.
LOG_WITH_PREFIX(INFO) << "Aborting op as it isn't in flight: "
<< SecureShortDebugString(*round->replicate_msg());
- round->NotifyReplicationFinished(Status::Aborted("Op aborted"));
+ round->NotifyReplicationFinished(kAbortedStatus);
}
return Status::OK();
}
@@ -83,24 +84,25 @@ void PendingRounds::AbortOpsAfter(int64_t index) {
// Either the new preceding id is in the pendings set or it must be equal to the
// committed index since we can't truncate already committed operations.
- if (iter != pending_ops_.end() && (*iter).first == index) {
- new_preceding = (*iter).second->replicate_msg()->id();
+ if (iter != pending_ops_.end() && iter->first == index) {
+ new_preceding = iter->second->replicate_msg()->id();
++iter;
} else {
CHECK_EQ(index, last_committed_op_id_.index());
new_preceding = last_committed_op_id_;
}
+ static const auto kAbortedStatus = Status::Aborted("Op aborted by new leader");
for (; iter != pending_ops_.end();) {
- const scoped_refptr<ConsensusRound>& round = (*iter).second;
- auto op_type = round->replicate_msg()->op_type();
+ const scoped_refptr<ConsensusRound>& round = iter->second;
+ const auto op_type = round->replicate_msg()->op_type();
LOG_WITH_PREFIX(INFO)
<< "Aborting uncommitted " << OperationType_Name(op_type)
<< " operation due to leader change: " << round->replicate_msg()->id();
- round->NotifyReplicationFinished(Status::Aborted("Op aborted by new leader"));
+ round->NotifyReplicationFinished(kAbortedStatus);
// Erase the entry from pendings.
- pending_ops_.erase(iter++);
+ iter = pending_ops_.erase(iter);
}
}
@@ -114,7 +116,6 @@ scoped_refptr<ConsensusRound> PendingRounds::GetPendingOpByIndexOrNull(int64_t i
}
bool PendingRounds::IsOpCommittedOrPending(const OpId& op_id, bool* term_mismatch) {
-
*term_mismatch = false;
if (op_id.index() <= GetCommittedIndex()) {
@@ -135,7 +136,7 @@ bool PendingRounds::IsOpCommittedOrPending(const OpId& op_id, bool* term_mismatc
const OpId& PendingRounds::GetLastPendingOpOpId() const {
return pending_ops_.empty()
- ? MinimumOpId() : (--pending_ops_.end())->second->id();
+ ? MinimumOpId() : pending_ops_.crbegin()->second->id();
}
Status PendingRounds::AdvanceCommittedIndex(int64_t committed_index) {
@@ -160,18 +161,19 @@ Status PendingRounds::AdvanceCommittedIndex(int64_t committed_index) {
return Status::OK();
}
+ // Stop at the operation after the last one we must commit.
+ const auto end_iter = pending_ops_.upper_bound(committed_index);
+
// Start at the operation after the last committed one.
auto iter = pending_ops_.upper_bound(last_committed_op_id_.index());
- // Stop at the operation after the last one we must commit.
- auto end_iter = pending_ops_.upper_bound(committed_index);
- CHECK(iter != pending_ops_.end());
+ DCHECK(iter != pending_ops_.end());
VLOG_WITH_PREFIX(1) << "Last triggered apply was: "
- << last_committed_op_id_
- << " Starting to apply from log index: " << (*iter).first;
+ << last_committed_op_id_
+ << " Starting to apply from log index: " << iter->first;
while (iter != end_iter) {
- scoped_refptr<ConsensusRound> round = (*iter).second; // Make a copy.
+ scoped_refptr<ConsensusRound> round = iter->second; // Make a copy.
DCHECK(round);
const OpId& current_id = round->id();
@@ -179,7 +181,7 @@ Status PendingRounds::AdvanceCommittedIndex(int64_t committed_index) {
CHECK_OK(CheckOpInSequence(last_committed_op_id_, current_id));
}
- pending_ops_.erase(iter++);
+ iter = pending_ops_.erase(iter);
last_committed_op_id_ = round->id();
time_manager_->AdvanceSafeTimeWithMessage(*round->replicate_msg());
round->NotifyReplicationFinished(Status::OK());
diff --git a/src/kudu/consensus/pending_rounds.h b/src/kudu/consensus/pending_rounds.h
index e3f0e4376..38a9fb98a 100644
--- a/src/kudu/consensus/pending_rounds.h
+++ b/src/kudu/consensus/pending_rounds.h
@@ -40,7 +40,7 @@ class TimeManager;
// NOTE: each round is associated with a single round, though "round" refers to
// the logical Raft replication and "op" refers to the replicated batch of
// operations.
-class PendingRounds {
+class PendingRounds final {
public:
PendingRounds(const std::string& log_prefix, TimeManager* time_manager);
~PendingRounds() = default;
@@ -106,7 +106,7 @@ class PendingRounds {
// Index=>Round map that manages pending ops, i.e. operations for which we've
// received a replicate message from the leader but have yet to be committed.
// The key is the index of the replicate operation.
- typedef std::map<int64_t, scoped_refptr<ConsensusRound> > IndexToRoundMap;
+ typedef std::map<int64_t, scoped_refptr<ConsensusRound>> IndexToRoundMap;
IndexToRoundMap pending_ops_;
// The OpId of the round that was last committed. Initialized to MinimumOpId().