You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by gr...@apache.org on 2019/01/10 23:05:16 UTC
[kudu] branch master updated (2831c91 -> d51889f)
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 2831c91 Support location awareness in READ_CLOSEST for the C++ client
new 8fb3022 Reduce election-related logging
new eae3504 Assign locations to tablet servers and the client in Java
new 7c17dc2 Eliminate redundant VLOG_IS_ON calls
new 7274710 Support location awareness in READ_CLOSEST for the Java client
new 5646a05 Fix DOS line endings in TestServerInfo.java
new 0fcb889 [examples] fix name of the class for spark-submit
new 441fc76 [backup] Workaround KUDU-1868 errors
new 61c4478 [backup] Make the scanRequestTimeout unit clear
new d51889f [spark-tools] Fix DistributedDataGenerator num-tasks
The 9 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:
examples/scala/spark-example/README.adoc | 18 ++-
.../scala/org/apache/kudu/backup/KuduBackup.scala | 9 +-
.../org/apache/kudu/backup/KuduBackupOptions.scala | 10 +-
.../org/apache/kudu/backup/KuduBackupRDD.scala | 2 +-
.../org/apache/kudu/client/AsyncKuduClient.java | 47 +++++--
.../java/org/apache/kudu/client/KuduClient.java | 9 ++
.../java/org/apache/kudu/client/RemoteTablet.java | 41 ++++--
.../org/apache/kudu/client/ReplicaSelection.java | 6 +-
.../java/org/apache/kudu/client/ServerInfo.java | 25 +++-
.../apache/kudu/client/ITScannerMultiTablet.java | 2 +-
.../apache/kudu/client/TestConnectionCache.java | 6 +-
.../org/apache/kudu/client/TestKuduClient.java | 22 ++-
.../org/apache/kudu/client/TestRemoteTablet.java | 103 +++++++++++---
.../org/apache/kudu/client/TestServerInfo.java | 155 +++++++++++----------
.../kudu/client/TestTableLocationsCache.java | 2 +-
.../src/test/resources/assign-location.py | 1 +
.../spark/tools/DistributedDataGenerator.scala | 4 +-
.../java/org/apache/kudu/test/KuduTestHarness.java | 20 +++
.../apache/kudu/test/cluster/MiniKuduCluster.java | 59 ++++++--
src/kudu/client/batcher.cc | 7 +-
src/kudu/clock/hybrid_clock.cc | 17 ++-
src/kudu/consensus/consensus_queue.cc | 46 +++---
src/kudu/consensus/leader_election.cc | 50 +++++--
src/kudu/consensus/leader_election.h | 3 +
src/kudu/consensus/raft_consensus.cc | 14 +-
src/kudu/tablet/tablet-test-base.h | 6 +-
src/kudu/tablet/tablet-test-util.h | 14 +-
src/kudu/tablet/tablet_bootstrap.cc | 4 +-
src/kudu/tablet/tablet_history_gc-test.cc | 9 +-
src/kudu/tserver/tablet_server-test.cc | 4 +-
src/kudu/tserver/tablet_service.cc | 6 +-
src/kudu/util/logging-test.cc | 17 +++
32 files changed, 518 insertions(+), 220 deletions(-)
create mode 120000 java/kudu-client/src/test/resources/assign-location.py
[kudu] 07/09: [backup] Workaround KUDU-1868 errors
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 441fc769661d32ab846dbb44bc9b668145eee7b2
Author: Grant Henke <gr...@apache.org>
AuthorDate: Wed Jan 9 15:27:17 2019 -0600
[backup] Workaround KUDU-1868 errors
As a workaround for KUDU-1868 the socketReadTimeout is
matched to the scanRequestTimeout. Without this
"Invalid call sequence ID" errors can occur under heavy load.
Change-Id: I9fb16cb5b2d4518534e3752526fa830161fe13b6
Reviewed-on: http://gerrit.cloudera.org:8080/12209
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins
---
.../src/main/scala/org/apache/kudu/backup/KuduBackup.scala | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
index bec3111..55541a2 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
@@ -40,7 +40,11 @@ object KuduBackup {
def run(options: KuduBackupOptions, session: SparkSession): Unit = {
val context =
- new KuduContext(options.kuduMasterAddresses, session.sparkContext)
+ new KuduContext(options.kuduMasterAddresses, session.sparkContext,
+ // TODO: As a workaround for KUDU-1868 the socketReadTimeout is
+ // matched to the scanRequestTimeout. Without this
+ // "Invalid call sequence ID" errors can occur under heavy load.
+ Some(options.scanRequestTimeout))
val path = options.path
log.info(s"Backing up to path: $path")
[kudu] 03/09: Eliminate redundant VLOG_IS_ON calls
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 7c17dc2ceeec0c27f7b36ff24bd5a725026d8da8
Author: Will Berkeley <wd...@gmail.org>
AuthorDate: Tue Jan 8 14:20:42 2019 -0800
Eliminate redundant VLOG_IS_ON calls
In an expression like
VLOG(3) << Substitute("Your $0 is $1", expensive_foo(), costly_bar());
The expression on the righthand side of the << operator is evaluated
only when VLOG_IS_ON(3) is true. However, there were some cases of
defensive programming like
if (VLOG_IS_ON(3)) {
VLOG(3) << Substitute("Your $0 is $1", expensive_foo(), costly_bar());
}
with a redundant call to VLOG_IS_ON. This patch removes such cases as I
found from a simple review of all VLOG_IS_ON callsites. I also sometimes
transformed VLOG messages from a chain of <<'s to calls to Substitute.
This patch contains no functional changes.
Change-Id: I0781e64f95e33fe067f9f3e65e77aec8654f4ba3
Reviewed-on: http://gerrit.cloudera.org:8080/12194
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
src/kudu/client/batcher.cc | 7 ++---
src/kudu/clock/hybrid_clock.cc | 17 ++++++------
src/kudu/consensus/consensus_queue.cc | 46 +++++++++++++++++--------------
src/kudu/consensus/raft_consensus.cc | 6 ++--
src/kudu/tablet/tablet-test-base.h | 6 ++--
src/kudu/tablet/tablet-test-util.h | 14 ++++++----
src/kudu/tablet/tablet_bootstrap.cc | 4 +--
src/kudu/tablet/tablet_history_gc-test.cc | 9 ++----
src/kudu/tserver/tablet_server-test.cc | 4 +--
src/kudu/tserver/tablet_service.cc | 6 ++--
src/kudu/util/logging-test.cc | 17 ++++++++++++
11 files changed, 75 insertions(+), 61 deletions(-)
diff --git a/src/kudu/client/batcher.cc b/src/kudu/client/batcher.cc
index 0710f9a..c213d00 100644
--- a/src/kudu/client/batcher.cc
+++ b/src/kudu/client/batcher.cc
@@ -53,6 +53,7 @@
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/rpc/connection.h"
+#include "kudu/rpc/request_tracker.h"
#include "kudu/rpc/response_callback.h"
#include "kudu/rpc/retriable_rpc.h"
#include "kudu/rpc/rpc.h"
@@ -79,7 +80,6 @@ class Schema;
namespace rpc {
class Messenger;
-class RequestTracker;
}
using pb_util::SecureDebugString;
@@ -317,9 +317,8 @@ WriteRpc::WriteRpc(const scoped_refptr<Batcher>& batcher,
VLOG(4) << ++ctr << ". Encoded row " << op->ToString();
}
- if (VLOG_IS_ON(3)) {
- VLOG(3) << "Created batch for " << tablet_id << ":\n" << SecureShortDebugString(req_);
- }
+ VLOG(3) << Substitute("Created batch for $0:\n$1",
+ tablet_id, SecureShortDebugString(req_));
}
WriteRpc::~WriteRpc() {
diff --git a/src/kudu/clock/hybrid_clock.cc b/src/kudu/clock/hybrid_clock.cc
index 7e87d4d..bec8cba 100644
--- a/src/kudu/clock/hybrid_clock.cc
+++ b/src/kudu/clock/hybrid_clock.cc
@@ -180,11 +180,10 @@ void HybridClock::NowWithError(Timestamp* timestamp, uint64_t* max_error_usec) {
next_timestamp_ = candidate_phys_timestamp;
*timestamp = Timestamp(next_timestamp_++);
*max_error_usec = error_usec;
- if (PREDICT_FALSE(VLOG_IS_ON(2))) {
- VLOG(2) << "Current clock is higher than the last one. Resetting logical values."
- << " Physical Value: " << now_usec << " usec Logical Value: 0 Error: "
- << error_usec;
- }
+ VLOG(2) << Substitute("Current clock is higher than the last one. "
+ "Resetting logical values. Physical Value: $0 usec "
+ "Logical Value: 0 Error: $1",
+ now_usec, error_usec);
return;
}
@@ -208,10 +207,10 @@ void HybridClock::NowWithError(Timestamp* timestamp, uint64_t* max_error_usec) {
*max_error_usec = (next_timestamp_ >> kBitsToShift) - (now_usec - error_usec);
*timestamp = Timestamp(next_timestamp_++);
- if (PREDICT_FALSE(VLOG_IS_ON(2))) {
- VLOG(2) << "Current clock is lower than the last one. Returning last read and incrementing"
- " logical values. Clock: " + Stringify(*timestamp) << " Error: " << *max_error_usec;
- }
+ VLOG(2) << Substitute("Current clock is lower than the last one. Returning "
+ "last read and incrementing logical values. "
+ "Clock: $0 Error: $1",
+ Stringify(*timestamp), *max_error_usec);
}
Status HybridClock::Update(const Timestamp& to_update) {
diff --git a/src/kudu/consensus/consensus_queue.cc b/src/kudu/consensus/consensus_queue.cc
index c618002..383a5cd 100644
--- a/src/kudu/consensus/consensus_queue.cc
+++ b/src/kudu/consensus/consensus_queue.cc
@@ -750,13 +750,17 @@ Status PeerMessageQueue::RequestForPeer(const string& uuid,
if (PREDICT_FALSE(VLOG_IS_ON(2))) {
if (request->ops_size() > 0) {
- VLOG_WITH_PREFIX_UNLOCKED(2) << "Sending request with operations to Peer: " << uuid
- << ". Size: " << request->ops_size()
- << ". From: " << SecureShortDebugString(request->ops(0).id()) << ". To: "
- << SecureShortDebugString(request->ops(request->ops_size() - 1).id());
+ VLOG_WITH_PREFIX_UNLOCKED(2)
+ << Substitute("Sending request with operations to Peer: $0. Size: $1. From: $2. To: $3",
+ uuid,
+ request->ops_size(),
+ SecureShortDebugString(request->ops(0).id()),
+ SecureShortDebugString(request->ops(request->ops_size() - 1).id()));
} else {
- VLOG_WITH_PREFIX_UNLOCKED(2) << "Sending status only request to Peer: " << uuid
- << ": " << SecureDebugString(*request);
+ VLOG_WITH_PREFIX_UNLOCKED(2)
+ << Substitute("Sending status only request to Peer: $0: $1",
+ uuid,
+ SecureDebugString(*request));
}
}
@@ -797,12 +801,14 @@ void PeerMessageQueue::AdvanceQueueWatermark(const char* type,
ReplicaTypes replica_types,
const TrackedPeer* who_caused) {
- if (VLOG_IS_ON(2)) {
- VLOG_WITH_PREFIX_UNLOCKED(2) << "Updating " << type << " watermark: "
- << "Peer (" << who_caused->ToString() << ") changed from "
- << replicated_before << " to " << replicated_after << ". "
- << "Current value: " << *watermark;
- }
+ VLOG_WITH_PREFIX_UNLOCKED(2)
+ << Substitute("Updating $0 watermark: Peer ($1) changed from $2 to $3. "
+ "Current value: $4",
+ type,
+ who_caused->ToString(),
+ OpIdToString(replicated_before),
+ OpIdToString(replicated_after),
+ *watermark);
// Go through the peer's watermarks, we want the highest watermark that
// 'num_peers_required' of peers has replicated. To find this we do the
@@ -855,15 +861,16 @@ void PeerMessageQueue::AdvanceQueueWatermark(const char* type,
int64_t old_watermark = *watermark;
*watermark = new_watermark;
- VLOG_WITH_PREFIX_UNLOCKED(1) << "Updated " << type << " watermark "
- << "from " << old_watermark << " to " << new_watermark;
+ VLOG_WITH_PREFIX_UNLOCKED(1)
+ << Substitute("Updated $0 watermark from $1 to $2",
+ type, old_watermark, new_watermark);
if (VLOG_IS_ON(3)) {
VLOG_WITH_PREFIX_UNLOCKED(3) << "Peers: ";
- for (const PeersMap::value_type& peer : peers_map_) {
+ for (const auto& peer : peers_map_) {
VLOG_WITH_PREFIX_UNLOCKED(3) << "Peer: " << peer.second->ToString();
}
VLOG_WITH_PREFIX_UNLOCKED(3) << "Sorted watermarks:";
- for (int64_t watermark : watermarks) {
+ for (const auto watermark : watermarks) {
VLOG_WITH_PREFIX_UNLOCKED(3) << "Watermark: " << watermark;
}
}
@@ -1176,10 +1183,9 @@ bool PeerMessageQueue::ResponseFromPeer(const std::string& peer_uuid,
CHECK_LE(response.responder_term(), queue_state_.current_term);
}
- if (PREDICT_FALSE(VLOG_IS_ON(2))) {
- VLOG_WITH_PREFIX_UNLOCKED(2) << "Received Response from Peer (" << peer->ToString() << "). "
- << "Response: " << SecureShortDebugString(response);
- }
+ VLOG_WITH_PREFIX_UNLOCKED(2)
+ << Substitute("Received Response from Peer ($0). Response: $1",
+ peer->ToString(), SecureShortDebugString(response));
mode_copy = queue_state_.mode;
diff --git a/src/kudu/consensus/raft_consensus.cc b/src/kudu/consensus/raft_consensus.cc
index 7428fe0..3435d5d 100644
--- a/src/kudu/consensus/raft_consensus.cc
+++ b/src/kudu/consensus/raft_consensus.cc
@@ -1012,9 +1012,9 @@ Status RaftConsensus::Update(const ConsensusRequestPB* request,
Status s = UpdateReplica(request, response);
if (PREDICT_FALSE(VLOG_IS_ON(1))) {
if (request->ops().empty()) {
- VLOG_WITH_PREFIX(1) << "Replica replied to status only request. Replica: "
- << ToString() << ". Response: "
- << SecureShortDebugString(*response);
+ VLOG_WITH_PREFIX(1)
+ << Substitute("Replica replied to status only request. Replica: $0. Response: $1",
+ ToString(), SecureShortDebugString(*response));
}
}
return s;
diff --git a/src/kudu/tablet/tablet-test-base.h b/src/kudu/tablet/tablet-test-base.h
index 3386f56..d6977f6 100644
--- a/src/kudu/tablet/tablet-test-base.h
+++ b/src/kudu/tablet/tablet-test-base.h
@@ -453,10 +453,8 @@ class TabletTestBase : public KuduTabletTest {
ASSERT_OK_FAST(iter->NextBlock(&block));
RowBlockRow rb_row = block.row(0);
- if (VLOG_IS_ON(2)) {
- VLOG(2) << "Fetched batch of " << block.nrows() << "\n"
- << "First row: " << schema_.DebugRow(rb_row);
- }
+ VLOG(2) << Substitute("Fetched batch of $0\nFirst row: $1",
+ block.nrows(), schema_.DebugRow(rb_row));
for (int i = 0; i < block.nrows(); i++) {
rb_row.Reset(&block, i);
diff --git a/src/kudu/tablet/tablet-test-util.h b/src/kudu/tablet/tablet-test-util.h
index b5947ea..22651af 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -81,6 +81,8 @@
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
+using strings::Substitute;
+
namespace kudu {
namespace clock {
@@ -418,8 +420,8 @@ class MirroredDeltas {
const SelectionVector& filter) {
if (VLOG_IS_ON(3)) {
std::string lower_ts_str = lower_ts ? lower_ts->ToString() : "INF";
- VLOG(3) << "Begin applying for timestamps [" << lower_ts_str << ","
- << upper_ts << ")";
+ VLOG(3) << Substitute("Begin applying for timestamps [$0, $1)",
+ lower_ts_str, upper_ts.ToString());
}
for (int i = 0; i < cb->nrows(); i++) {
rowid_t row_idx = start_row_idx + i;
@@ -902,14 +904,14 @@ void RunDeltaFuzzTest(const DeltaStore& store,
AllowIsDeleted::NO);
size_t projection_vc_is_deleted_idx =
projection.find_first_is_deleted_virtual_column();
- SCOPED_TRACE(strings::Substitute("Projection $0", projection.ToString()));
+ SCOPED_TRACE(Substitute("Projection $0", projection.ToString()));
RowIteratorOptions opts;
opts.projection = &projection;
if (lower_ts) {
opts.snap_to_exclude = MvccSnapshot(*lower_ts);
}
opts.snap_to_include = MvccSnapshot(upper_ts);
- SCOPED_TRACE(strings::Substitute("Timestamps: [$0,$1)",
+ SCOPED_TRACE(Substitute("Timestamps: [$0,$1)",
lower_ts ? lower_ts->ToString() : "INF",
upper_ts.ToString()));
DeltaIterator* raw_iter;
@@ -928,7 +930,7 @@ void RunDeltaFuzzTest(const DeltaStore& store,
rowid_t start_row_idx = 0;
while (iter->HasNext()) {
int batch_size = prng->Uniform(kMaxBatchSize) + 1;
- SCOPED_TRACE(strings::Substitute("batch starting at $0 ($1 rows)",
+ SCOPED_TRACE(Substitute("batch starting at $0 ($1 rows)",
start_row_idx, batch_size));
int prepare_flags = DeltaIterator::PREPARE_FOR_APPLY |
DeltaIterator::PREPARE_FOR_COLLECT;
@@ -969,7 +971,7 @@ void RunDeltaFuzzTest(const DeltaStore& store,
// Test ApplyUpdates: all relevant updates are applied to the column block.
for (int j = 0; j < opts.projection->num_columns(); j++) {
- SCOPED_TRACE(strings::Substitute("Column $0", j));
+ SCOPED_TRACE(Substitute("Column $0", j));
bool col_is_nullable = opts.projection->column(j).is_nullable();
ScopedColumnBlock<UINT32> expected_scb(batch_size, col_is_nullable);
ScopedColumnBlock<UINT32> actual_scb(batch_size, col_is_nullable);
diff --git a/src/kudu/tablet/tablet_bootstrap.cc b/src/kudu/tablet/tablet_bootstrap.cc
index d32316c..084b1e0 100644
--- a/src/kudu/tablet/tablet_bootstrap.cc
+++ b/src/kudu/tablet/tablet_bootstrap.cc
@@ -835,9 +835,7 @@ Status TabletBootstrap::HandleEntry(const IOContext* io_context,
unique_ptr<LogEntryPB> entry,
string* entry_debug_info) {
DCHECK(entry);
- if (VLOG_IS_ON(1)) {
- VLOG_WITH_PREFIX(1) << "Handling entry: " << SecureShortDebugString(*entry);
- }
+ VLOG_WITH_PREFIX(1) << "Handling entry: " << SecureShortDebugString(*entry);
const auto entry_type = entry->type();
switch (entry_type) {
diff --git a/src/kudu/tablet/tablet_history_gc-test.cc b/src/kudu/tablet/tablet_history_gc-test.cc
index 37397aa..4969900 100644
--- a/src/kudu/tablet/tablet_history_gc-test.cc
+++ b/src/kudu/tablet/tablet_history_gc-test.cc
@@ -17,7 +17,6 @@
#include <algorithm>
#include <atomic>
-
#include <cstdint>
#include <memory>
#include <ostream>
@@ -40,6 +39,7 @@
#include "kudu/gutil/casts.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/ref_counted.h"
+#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/walltime.h"
#include "kudu/tablet/compaction.h"
@@ -49,6 +49,7 @@
#include "kudu/tablet/rowset_metadata.h"
#include "kudu/tablet/tablet-harness.h"
#include "kudu/tablet/tablet-test-base.h"
+#include "kudu/tablet/tablet-test-util.h"
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet_metadata.h"
#include "kudu/tablet/tablet_metrics.h"
@@ -540,11 +541,7 @@ TEST_F(TabletHistoryGcTest, TestGcWithConcurrentCompaction) {
vector<string> rows;
ASSERT_OK(IterateToStringList(&rows));
- if (VLOG_IS_ON(2)) {
- for (const string& r : rows) {
- VLOG(2) << r;
- }
- }
+ VLOG(2) << JoinStrings(rows, "\n");
vector<int32_t> expected_rows = { 1, 3, 5, 7, 9 };
for (int i = 0; i < expected_rows.size(); i++) {
diff --git a/src/kudu/tserver/tablet_server-test.cc b/src/kudu/tserver/tablet_server-test.cc
index bfdd7ab..dd60bd1 100644
--- a/src/kudu/tserver/tablet_server-test.cc
+++ b/src/kudu/tserver/tablet_server-test.cc
@@ -1782,8 +1782,8 @@ TEST_F(TabletServerTest, TestSnapshotScan) {
ASSERT_EQ(expected_num_rows, results.size());
if (VLOG_IS_ON(2)) {
- VLOG(2) << "Scanner: " << resp.scanner_id() << " performing a snapshot read at: "
- << read_timestamp.ToString() << " got back: ";
+ VLOG(2) << Substitute("Scanner: $0 performing a snapshot read at $1 got back: ",
+ resp.scanner_id(), read_timestamp.ToString());
for (const string& result : results) {
VLOG(2) << result;
}
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index 0ec4e5f..40c855d 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -1756,10 +1756,8 @@ static Status SetupScanSpec(const NewScanRequestPB& scan_pb,
auto pred = ColumnPredicate::InclusiveRange(col, lower_bound, upper_bound, scanner->arena());
if (pred) {
- if (VLOG_IS_ON(3)) {
- VLOG(3) << "Parsed predicate " << pred->ToString()
- << " from " << SecureShortDebugString(scan_pb);
- }
+ VLOG(3) << Substitute("Parsed predicate $0 from $1",
+ pred->ToString(), SecureShortDebugString(scan_pb));
ret->AddPredicate(*pred);
}
}
diff --git a/src/kudu/util/logging-test.cc b/src/kudu/util/logging-test.cc
index cceece8..9562919 100644
--- a/src/kudu/util/logging-test.cc
+++ b/src/kudu/util/logging-test.cc
@@ -246,4 +246,21 @@ TEST(LoggingTest, TestLogTiming) {
ASSERT_EQ("hello", s2);
}
+
+// Test that VLOG(n) does not evaluate its message if the verbose level is < n,
+// ensuring that it is perf-safe to write things like
+//
+// VLOG(1) << Substitute("your foo is $0", compute_costly_bar_string());
+//
+// in hot code paths.
+TEST(LoggingTest, TestVlogDoesNotEvaluateMessage) {
+ if (VLOG_IS_ON(1)) {
+ LOG(INFO) << "Test skipped: verbose level is at least 1";
+ return;
+ }
+
+ int numVlogs = 0;
+ VLOG(1) << "This shouldn't be logged: " << numVlogs++;
+ ASSERT_EQ(0, numVlogs);
+}
} // namespace kudu
[kudu] 05/09: Fix DOS line endings in TestServerInfo.java
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 5646a05917230fa3f4156099c29a77559fdd2cea
Author: Will Berkeley <wd...@gmail.org>
AuthorDate: Tue Jan 8 10:32:18 2019 -0800
Fix DOS line endings in TestServerInfo.java
Change-Id: Id0c470ed24238e0461e96bd443f3d649e1e11acb
Reviewed-on: http://gerrit.cloudera.org:8080/12180
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
.../org/apache/kudu/client/TestServerInfo.java | 162 ++++++++++-----------
1 file changed, 81 insertions(+), 81 deletions(-)
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java
index 63fd093..8a67cec 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java
@@ -1,81 +1,81 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package org.apache.kudu.client;
-
-import java.net.InetAddress;
-
-import org.apache.kudu.test.cluster.FakeDNS;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.net.UnknownHostException;
-
-public class TestServerInfo {
- /**
- * Test for KUDU-2103. Checks if the original hostnames is returned if unknown.
- */
- @Test
- public void testGetAndCanonicalizeUnknownHostname() throws Exception {
- installFakeDNS("master1.example.com", "server123.example.com", "10.1.2.3");
-
- ServerInfo serverInfo = new ServerInfo(
- "nevermind",
- new HostAndPort("master2.example.com", 12345),
- InetAddress.getByName("10.1.2.3"),
- /*location=*/"");
-
- Assert.assertEquals("master2.example.com", serverInfo.getAndCanonicalizeHostname());
- Assert.assertEquals("", serverInfo.getLocation());
- }
-
- /**
- * Test for KUDU-2103. Checks if the canonical hostname is returned instead
- * of the one it's set to.
- */
- @Test
- public void testGetAndCanonicalizeHostname() throws Exception {
- installFakeDNS("master1.example.com", "server123.example.com", "10.1.2.3");
-
- ServerInfo serverInfo = new ServerInfo(
- "abcdef", // uuid
- new HostAndPort("master1.example.com", 12345),
- InetAddress.getByName("10.1.2.3"),
- /* location= */"/foo");
-
- Assert.assertEquals("server123.example.com", serverInfo.getAndCanonicalizeHostname());
- Assert.assertEquals("/foo", serverInfo.getLocation());
- Assert.assertEquals("abcdef(master1.example.com:12345)", serverInfo.toString());
- }
-
- /**
- * Helper method to install FakeDNS with the expected values for the tests
- *
- * @param alias alias to be set for forward resolution
- * @param canonical canonical to be set for reverse resolution
- * @param ip IP both hostnames point to
- * @throws UnknownHostException if the "ip" is an unknown host
- */
- private void installFakeDNS(String alias, String canonical, String ip)
- throws UnknownHostException {
- FakeDNS fakeDNS = FakeDNS.getInstance();
- fakeDNS.install();
- InetAddress inetAddress = InetAddress.getByName(ip);
- fakeDNS.addForwardResolution(alias, inetAddress);
- fakeDNS.addForwardResolution(canonical, inetAddress);
- fakeDNS.addReverseResolution(inetAddress, canonical);
- }
-}
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.kudu.client;
+
+import java.net.InetAddress;
+
+import org.apache.kudu.test.cluster.FakeDNS;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.net.UnknownHostException;
+
+public class TestServerInfo {
+ /**
+ * Test for KUDU-2103. Checks if the original hostnames is returned if unknown.
+ */
+ @Test
+ public void testGetAndCanonicalizeUnknownHostname() throws Exception {
+ installFakeDNS("master1.example.com", "server123.example.com", "10.1.2.3");
+
+ ServerInfo serverInfo = new ServerInfo(
+ "nevermind",
+ new HostAndPort("master2.example.com", 12345),
+ InetAddress.getByName("10.1.2.3"),
+ /*location=*/"");
+
+ Assert.assertEquals("master2.example.com", serverInfo.getAndCanonicalizeHostname());
+ Assert.assertEquals("", serverInfo.getLocation());
+ }
+
+ /**
+ * Test for KUDU-2103. Checks if the canonical hostname is returned instead
+ * of the one it's set to.
+ */
+ @Test
+ public void testGetAndCanonicalizeHostname() throws Exception {
+ installFakeDNS("master1.example.com", "server123.example.com", "10.1.2.3");
+
+ ServerInfo serverInfo = new ServerInfo(
+ "abcdef", // uuid
+ new HostAndPort("master1.example.com", 12345),
+ InetAddress.getByName("10.1.2.3"),
+ /* location= */"/foo");
+
+ Assert.assertEquals("server123.example.com", serverInfo.getAndCanonicalizeHostname());
+ Assert.assertEquals("/foo", serverInfo.getLocation());
+ Assert.assertEquals("abcdef(master1.example.com:12345)", serverInfo.toString());
+ }
+
+ /**
+ * Helper method to install FakeDNS with the expected values for the tests
+ *
+ * @param alias alias to be set for forward resolution
+ * @param canonical canonical to be set for reverse resolution
+ * @param ip IP both hostnames point to
+ * @throws UnknownHostException if the "ip" is an unknown host
+ */
+ private void installFakeDNS(String alias, String canonical, String ip)
+ throws UnknownHostException {
+ FakeDNS fakeDNS = FakeDNS.getInstance();
+ fakeDNS.install();
+ InetAddress inetAddress = InetAddress.getByName(ip);
+ fakeDNS.addForwardResolution(alias, inetAddress);
+ fakeDNS.addForwardResolution(canonical, inetAddress);
+ fakeDNS.addReverseResolution(inetAddress, canonical);
+ }
+}
[kudu] 04/09: Support location awareness in READ_CLOSEST for the
Java client
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 7274710bb9210ae812619545540ccf08b052d3c8
Author: Will Berkeley <wd...@gmail.org>
AuthorDate: Wed Dec 26 16:16:53 2018 -0500
Support location awareness in READ_CLOSEST for the Java client
Change-Id: Ief0f07058cefd0037f4b0f7c60c8b7809dc8313f
Reviewed-on: http://gerrit.cloudera.org:8080/12175
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
.../org/apache/kudu/client/AsyncKuduClient.java | 14 ++-
.../java/org/apache/kudu/client/RemoteTablet.java | 41 +++++++--
.../org/apache/kudu/client/ReplicaSelection.java | 6 +-
.../java/org/apache/kudu/client/ServerInfo.java | 14 ++-
.../apache/kudu/client/ITScannerMultiTablet.java | 2 +-
.../org/apache/kudu/client/TestRemoteTablet.java | 102 +++++++++++++++++----
.../kudu/client/TestTableLocationsCache.java | 2 +-
7 files changed, 143 insertions(+), 38 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
index e71293b..4ed25f6 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
@@ -515,7 +515,7 @@ public class AsyncKuduClient implements AutoCloseable {
/**
* Returns a string representation of this client's location. If this
* client was not assigned a location, returns the empty string.
- *
+ *
* @return a string representation of this client's location
*/
public String getLocationString() {
@@ -1111,7 +1111,8 @@ public class AsyncKuduClient implements AutoCloseable {
// Important to increment the attempts before the next if statement since
// getSleepTimeForRpc() relies on it if the client is null or dead.
nextRequest.attempt++;
- final ServerInfo info = tablet.getReplicaSelectedServerInfo(nextRequest.getReplicaSelection());
+ final ServerInfo info = tablet.getReplicaSelectedServerInfo(nextRequest.getReplicaSelection(),
+ location);
if (info == null) {
return delayedSendRpcToTablet(nextRequest, new RecoverableException(Status.RemoteError(
String.format("No information on servers hosting tablet %s, will retry later",
@@ -1137,7 +1138,8 @@ public class AsyncKuduClient implements AutoCloseable {
return Deferred.fromResult(null);
}
final KuduRpc<AsyncKuduScanner.Response> closeRequest = scanner.getCloseRequest();
- final ServerInfo info = tablet.getReplicaSelectedServerInfo(closeRequest.getReplicaSelection());
+ final ServerInfo info = tablet.getReplicaSelectedServerInfo(closeRequest.getReplicaSelection(),
+ location);
if (info == null) {
return Deferred.fromResult(null);
}
@@ -1165,7 +1167,8 @@ public class AsyncKuduClient implements AutoCloseable {
}
final KuduRpc<Void> keepAliveRequest = scanner.getKeepAliveRequest();
- final ServerInfo info = tablet.getReplicaSelectedServerInfo(keepAliveRequest.getReplicaSelection());
+ final ServerInfo info = tablet.getReplicaSelectedServerInfo(keepAliveRequest.getReplicaSelection(),
+ location);
if (info == null) {
return Deferred.fromResult(null);
}
@@ -1218,7 +1221,8 @@ public class AsyncKuduClient implements AutoCloseable {
// If we found a tablet, we'll try to find the TS to talk to.
if (entry != null) {
RemoteTablet tablet = entry.getTablet();
- ServerInfo info = tablet.getReplicaSelectedServerInfo(request.getReplicaSelection());
+ ServerInfo info = tablet.getReplicaSelectedServerInfo(request.getReplicaSelection(),
+ location);
if (info != null) {
Deferred<R> d = request.getDeferred();
request.setTablet(tablet);
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
index ed8c5cc..9d77275 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/RemoteTablet.java
@@ -172,25 +172,45 @@ public class RemoteTablet implements Comparable<RemoteTablet> {
}
/**
- * Get the information on the closest server. If none is closer than the others,
- * return the information on a randomly picked server.
+ * Get the information on the closest server. Servers are ranked from closest to furthest as
+ * follows:
+ * - Local servers
+ * - Servers in the same location as the client
+ * - All other servers
*
- * @return the information on the closest server, which might be any if none is closer, or null
- * if this cache doesn't know any servers.
+ * @param location the location of the client
+ * @return the information for a closest server, or null if this cache doesn't know any servers.
*/
@Nullable
- ServerInfo getClosestServerInfo() {
+ ServerInfo getClosestServerInfo(String location) {
+ // TODO(KUDU-2348) this doesn't return a random server, but rather returns
+ // 1. whichever local server's hashcode places it first among local servers,
+ // if there is a local server, or
+ // 2. whichever server in the same location has a hashcode that places it
+ // first among servers in the same location, if there is a server in the
+ // same location, or, finally,
+ // 3. whichever server's hashcode places it last.
+ // That might be the same "random" choice across all clients, which is not
+ // so good. Unfortunately, the client depends on this method returning the
+ // same tablet server given the same state. See
+ // testGetReplicaSelectedServerInfoDeterminism in TestRemoteTablet.java.
+ // TODO(wdberkeley): Eventually, the client might use the hierarchical
+ // structure of a location to determine proximity.
synchronized (tabletServers) {
ServerInfo last = null;
+ ServerInfo lastInSameLocation = null;
for (ServerInfo e : tabletServers.values()) {
last = e;
if (e.isLocal()) {
return e;
}
+ if (e.inSameLocation(location)) {
+ lastInSameLocation = e;
+ }
+ }
+ if (lastInSameLocation != null) {
+ return lastInSameLocation;
}
- // TODO(KUDU-2348) this doesn't return a random server, but rather returns
- // whichever one's hashcode places it last. That might be the same
- // "random" choice across all clients, which is not so good.
return last;
}
}
@@ -200,15 +220,16 @@ public class RemoteTablet implements Comparable<RemoteTablet> {
* mechanism.
*
* @param replicaSelection replica selection mechanism to use
+ * @param location the location of the client
* @return information on the server that matches the selection, can be null
*/
@Nullable
- ServerInfo getReplicaSelectedServerInfo(ReplicaSelection replicaSelection) {
+ ServerInfo getReplicaSelectedServerInfo(ReplicaSelection replicaSelection, String location) {
switch (replicaSelection) {
case LEADER_ONLY:
return getLeaderServerInfo();
case CLOSEST_REPLICA:
- return getClosestServerInfo();
+ return getClosestServerInfo(location);
default:
throw new RuntimeException("unknown replica selection mechanism " + replicaSelection);
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java
index 8fc0ebf..def81d3 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ReplicaSelection.java
@@ -31,7 +31,11 @@ public enum ReplicaSelection {
*/
LEADER_ONLY,
/**
- * Select the closest replica to the client, or a random one if all replicas are equidistant.
+ * Select the closest replica to the client. Replicas are classified from closest to furthest as
+ * follows:
+ * - Local replicas
+ * - Replicas whose tablet server has the same location as the client
+ * - All other replicas
*/
CLOSEST_REPLICA
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
index 67b2963..cad4b21 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
@@ -94,14 +94,24 @@ public class ServerInfo {
}
/**
- * Returns this server's location.
- * @return the server's location, or the empty string if no location was assigned.
+ * Returns this server's location. If no location is assigned, returns an empty string.
+ * @return the server's location
*/
public String getLocation() {
return location;
}
/**
+ * Returns true if the server is in the same location as 'location'.
+ * @return true if the server is in 'location'.
+ */
+ public boolean inSameLocation(String loc) {
+ Preconditions.checkNotNull(loc);
+ return !loc.isEmpty() &&
+ loc.equals(location);
+ }
+
+ /**
* Returns if this server is on this client's host.
* @return true if the server is local, else false
*/
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/ITScannerMultiTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/ITScannerMultiTablet.java
index 2a5c0fc..58ed5f3 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/ITScannerMultiTablet.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/ITScannerMultiTablet.java
@@ -177,7 +177,7 @@ public class ITScannerMultiTablet {
// Forcefully disconnects the current connection and fails all outstanding RPCs
// in the middle of scanning.
harness.getAsyncClient().newRpcProxy(scanner.currentTablet().getReplicaSelectedServerInfo(
- scanner.getReplicaSelection())).getConnection().disconnect();
+ scanner.getReplicaSelection(), /* location= */"")).getConnection().disconnect();
while (scanner.hasMoreRows()) {
loopCount++;
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
index f1c09ab..368ecbc 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
@@ -35,6 +35,9 @@ import org.apache.kudu.consensus.Metadata;
import org.apache.kudu.master.Master;
public class TestRemoteTablet {
+ private static final String kClientLocation = "/fake-client";
+ private static final String kLocation = "/fake-noclient";
+ private static final String kNoLocation = "";
private static final String[] kUuids = { "uuid-0", "uuid-1", "uuid-2" };
@Test
@@ -97,27 +100,77 @@ public class TestRemoteTablet {
@Test
public void testLocalReplica() {
- RemoteTablet tablet = getTablet(0, 0);
+ {
+ // Tablet with no replicas in the same location as the client.
+ RemoteTablet tablet = getTablet(0, 0, -1);
- assertEquals(kUuids[0], tablet.getClosestServerInfo().getUuid());
+ // No location for the client.
+ assertEquals(kUuids[0], tablet.getClosestServerInfo(kNoLocation).getUuid());
+
+ // Client with location.
+ assertEquals(kUuids[0], tablet.getClosestServerInfo(kClientLocation).getUuid());
+ }
+
+ {
+ // Tablet with a non-local replica in the same location as the client.
+ RemoteTablet tablet = getTablet(0, 0, 1);
+
+ // No location for the client.
+ assertEquals(kUuids[0], tablet.getClosestServerInfo(kNoLocation).getUuid());
+
+ // Client with location. The local replica should be chosen.
+ assertEquals(kUuids[0], tablet.getClosestServerInfo(kClientLocation).getUuid());
+ }
+
+ {
+ // Tablet with a local replica in the same location as the client.
+ RemoteTablet tablet = getTablet(0, 0, 0);
+
+ // No location for the client.
+ assertEquals(kUuids[0], tablet.getClosestServerInfo(kNoLocation).getUuid());
+
+ // Client with location. The local replica should be chosen.
+ assertEquals(kUuids[0], tablet.getClosestServerInfo(kClientLocation).getUuid());
+ }
}
@Test
- public void testNoLocalReplica() {
- RemoteTablet tablet = getTablet(0, -1);
+ public void testNoLocalOrSameLocationReplica() {
+ RemoteTablet tablet = getTablet(0, -1, -1);
// We just care about getting one back.
- assertNotNull(tablet.getClosestServerInfo().getUuid());
+ assertNotNull(tablet.getClosestServerInfo(kClientLocation).getUuid());
}
@Test
public void testReplicaSelection() {
- RemoteTablet tablet = getTablet(0, 1);
+ {
+ RemoteTablet tablet = getTablet(0, 1, 2);
+
+ // LEADER_ONLY picks the leader even if there's a local replica.
+ assertEquals(kUuids[0],
+ tablet.getReplicaSelectedServerInfo(ReplicaSelection.LEADER_ONLY, kClientLocation)
+ .getUuid());
+
+ // CLOSEST_REPLICA picks the local replica even if there's a replica in the same location.
+ assertEquals(kUuids[1],
+ tablet.getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA, kClientLocation)
+ .getUuid());
+ }
+
+ {
+ RemoteTablet tablet = getTablet(0, -1, 1);
- assertEquals(kUuids[0],
- tablet.getReplicaSelectedServerInfo(ReplicaSelection.LEADER_ONLY).getUuid());
- assertEquals(kUuids[1],
- tablet.getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA).getUuid());
+ // LEADER_ONLY picks the leader even if there's a replica with the same location.
+ assertEquals(kUuids[0],
+ tablet.getReplicaSelectedServerInfo(ReplicaSelection.LEADER_ONLY, kClientLocation)
+ .getUuid());
+
+ // CLOSEST_REPLICA picks the replica in the same location.
+ assertEquals(kUuids[1],
+ tablet.getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA, kClientLocation)
+ .getUuid());
+ }
}
// AsyncKuduClient has methods like scanNextRows, keepAlive, and closeScanner that rely on
@@ -126,33 +179,45 @@ public class TestRemoteTablet {
// This test ensures that remains true.
@Test
public void testGetReplicaSelectedServerInfoDeterminism() {
- RemoteTablet tabletWithLocal = getTablet(0, 0);
+ // There's a local leader replica.
+ RemoteTablet tabletWithLocal = getTablet(0, 0, 0);
verifyGetReplicaSelectedServerInfoDeterminism(tabletWithLocal);
- RemoteTablet tabletWithRemote = getTablet(0, -1);
+ // There's a leader in the same location as the client.
+ RemoteTablet tabletWithSameLocation = getTablet(0, -1, 0);
+ verifyGetReplicaSelectedServerInfoDeterminism(tabletWithSameLocation);
+
+ // There's no local replica or replica in the same location.
+ RemoteTablet tabletWithRemote = getTablet(0, -1, -1);
verifyGetReplicaSelectedServerInfoDeterminism(tabletWithRemote);
}
private void verifyGetReplicaSelectedServerInfoDeterminism(RemoteTablet tablet) {
- String init = tablet.getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA).getUuid();
+ String init = tablet
+ .getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA, kClientLocation)
+ .getUuid();
for (int i = 0; i < 10; i++) {
- String next = tablet.getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA).getUuid();
+ String next = tablet
+ .getReplicaSelectedServerInfo(ReplicaSelection.CLOSEST_REPLICA, kClientLocation)
+ .getUuid();
assertEquals("getReplicaSelectedServerInfo was not deterministic", init, next);
}
}
@Test
public void testToString() {
- RemoteTablet tablet = getTablet(0, 1);
+ RemoteTablet tablet = getTablet(0, 1, -1);
assertEquals("fake tablet@[uuid-0(host:1000)[L],uuid-1(host:1001),uuid-2(host:1002)]",
tablet.toString());
}
private RemoteTablet getTablet(int leaderIndex) {
- return getTablet(leaderIndex, -1);
+ return getTablet(leaderIndex, -1, -1);
}
- static RemoteTablet getTablet(int leaderIndex, int localReplicaIndex) {
+ static RemoteTablet getTablet(int leaderIndex,
+ int localReplicaIndex,
+ int sameLocationReplicaIndex) {
Master.TabletLocationsPB.Builder tabletPb = Master.TabletLocationsPB.newBuilder();
tabletPb.setPartition(ProtobufUtils.getFakePartitionPB());
@@ -171,10 +236,11 @@ public class TestRemoteTablet {
}
String uuid = kUuids[i];
+ String location = i == sameLocationReplicaIndex ? kClientLocation : kLocation;
servers.add(new ServerInfo(uuid,
new HostAndPort("host", 1000 + i),
addr,
- /*location=*/""));
+ location));
tabletPb.addReplicas(ProtobufUtils.getFakeTabletReplicaPB(
uuid, "host", i,
leaderIndex == i ? Metadata.RaftPeerPB.Role.LEADER : Metadata.RaftPeerPB.Role.FOLLOWER));
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestTableLocationsCache.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTableLocationsCache.java
index d61fb8e..c9de4d3 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestTableLocationsCache.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTableLocationsCache.java
@@ -45,7 +45,7 @@ public class TestTableLocationsCache {
@Test
public void testToString() {
- RemoteTablet tablet = TestRemoteTablet.getTablet(0, 1);
+ RemoteTablet tablet = TestRemoteTablet.getTablet(0, 1, -1);
List<RemoteTablet> tablets = ImmutableList.of(tablet);
cache.cacheTabletLocations(tablets,
tablet.getPartition().getPartitionKeyStart(),
[kudu] 06/09: [examples] fix name of the class for spark-submit
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 0fcb8890c64e948099a2bd156452ecfe15376b5f
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Wed Jan 9 17:01:01 2019 -0800
[examples] fix name of the class for spark-submit
I also updated the description of master RPC endpoints and
removed the redundant default port number for masters' RPC
endpoints in the example.
Change-Id: Ic36a6fb62bd55667ea9f11c0124b5388f0e5bc7b
Reviewed-on: http://gerrit.cloudera.org:8080/12206
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
Reviewed-by: Hao Hao <ha...@cloudera.com>
---
examples/scala/spark-example/README.adoc | 18 +++++++++++-------
1 file changed, 11 insertions(+), 7 deletions(-)
diff --git a/examples/scala/spark-example/README.adoc b/examples/scala/spark-example/README.adoc
index 8969325..f5c4e36 100644
--- a/examples/scala/spark-example/README.adoc
+++ b/examples/scala/spark-example/README.adoc
@@ -41,21 +41,25 @@ $ mvn package
To configure the kudu-spark example, there are two Java system properties
available:
-- kuduMasters: A comma-separated list of Kudu master addresses.
+- kuduMasters: A comma-separated list of Kudu master RPC endpoints, where
+ each endpoint is in form '<HostName|IPAddress>[:PortNumber]' (the port number
+ by default is 7051 if not specified).
Default: 'localhost:7051'.
- tableName: The name of the table to use for the example program. This
table should not exist in Kudu. Default: 'spark_test'.
The application can be run using `spark-submit`. For example, to run the
-example against a Spark cluster running on YARN, use a command like the
-following:
+example against a Spark cluster running on YARN with Kudu masters at nodes
+master1, master2, master3, use a command like the following:
[source.bash]
----
-$ spark-submit --class org.apache.kudu.examples.SparkExample --master yarn \
---driver-java-options \
-'-DkuduMasters=master1:7051,master2:7051,master3:7051 -DtableName=test_table' \
-target/kudu-spark-example-1.0-SNAPSHOT.jar
+$ spark-submit \
+ --class org.apache.kudu.spark.examples.SparkExample \
+ --master yarn \
+ --driver-java-options \
+ '-DkuduMasters=master1,master2,master3 -DtableName=test_table' \
+ target/kudu-spark-example-1.0-SNAPSHOT.jar
----
You will need the Kudu cluster to be up and running and Spark correctly
[kudu] 08/09: [backup] Make the scanRequestTimeout unit clear
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 61c44785763219df1225108dbde6ac6ada70e087
Author: Grant Henke <gr...@apache.org>
AuthorDate: Wed Jan 9 15:38:57 2019 -0600
[backup] Make the scanRequestTimeout unit clear
Changes the scanRequestTimeout parameter to be
scanRequestTimeoutMs to make sure the time unit is
clear.
Change-Id: Id44832a06ae5fb4762a402faff82012729afb5c4
Reviewed-on: http://gerrit.cloudera.org:8080/12210
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
.../src/main/scala/org/apache/kudu/backup/KuduBackup.scala | 7 +++++--
.../main/scala/org/apache/kudu/backup/KuduBackupOptions.scala | 10 +++++-----
.../src/main/scala/org/apache/kudu/backup/KuduBackupRDD.scala | 2 +-
3 files changed, 11 insertions(+), 8 deletions(-)
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
index 55541a2..5a30f15 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
@@ -40,11 +40,14 @@ object KuduBackup {
def run(options: KuduBackupOptions, session: SparkSession): Unit = {
val context =
- new KuduContext(options.kuduMasterAddresses, session.sparkContext,
+ new KuduContext(
+ options.kuduMasterAddresses,
+ session.sparkContext,
// TODO: As a workaround for KUDU-1868 the socketReadTimeout is
// matched to the scanRequestTimeout. Without this
// "Invalid call sequence ID" errors can occur under heavy load.
- Some(options.scanRequestTimeout))
+ Some(options.scanRequestTimeoutMs)
+ )
val path = options.path
log.info(s"Backing up to path: $path")
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupOptions.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupOptions.scala
index 82a8eb4..d366c7c 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupOptions.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupOptions.scala
@@ -32,14 +32,14 @@ case class KuduBackupOptions(
timestampMs: Long = System.currentTimeMillis(),
format: String = KuduBackupOptions.DefaultFormat,
scanBatchSize: Int = KuduBackupOptions.DefaultScanBatchSize,
- scanRequestTimeout: Long = KuduBackupOptions.DefaultScanRequestTimeout,
+ scanRequestTimeoutMs: Long = KuduBackupOptions.DefaultScanRequestTimeoutMs,
scanPrefetching: Boolean = KuduBackupOptions.DefaultScanPrefetching,
keepAlivePeriodMs: Long = KuduBackupOptions.defaultKeepAlivePeriodMs)
object KuduBackupOptions {
val DefaultFormat: String = "parquet"
val DefaultScanBatchSize: Int = 1024 * 1024 * 20 // 20 MiB
- val DefaultScanRequestTimeout: Long =
+ val DefaultScanRequestTimeoutMs: Long =
AsyncKuduClient.DEFAULT_OPERATION_TIMEOUT_MS // 30 seconds
val DefaultScanPrefetching
: Boolean = false // TODO: Add a test per KUDU-1260 and enable by default?
@@ -75,9 +75,9 @@ object KuduBackupOptions {
.text("The maximum number of bytes returned by the scanner, on each batch.")
.optional()
- opt[Int]("scanRequestTimeout")
- .action((v, o) => o.copy(scanRequestTimeout = v))
- .text("Sets how long each scan request to a server can last.")
+ opt[Int]("scanRequestTimeoutMs")
+ .action((v, o) => o.copy(scanRequestTimeoutMs = v))
+ .text("Sets how long in milliseconds each scan request to a server can last.")
.optional()
opt[Unit]("scanPrefetching")
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupRDD.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupRDD.scala
index 9be2bf2..2e45827 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupRDD.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackupRDD.scala
@@ -65,7 +65,7 @@ class KuduBackupRDD private[kudu] (
.readMode(ReadMode.READ_AT_SNAPSHOT)
.snapshotTimestampRaw(hybridTime)
.batchSizeBytes(options.scanBatchSize)
- .scanRequestTimeout(options.scanRequestTimeout)
+ .scanRequestTimeout(options.scanRequestTimeoutMs)
.prefetching(options.scanPrefetching)
.build()
[kudu] 09/09: [spark-tools] Fix DistributedDataGenerator num-tasks
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit d51889f70beaac138ece3a3ea00633d1165038d3
Author: Grant Henke <gr...@apache.org>
AuthorDate: Wed Jan 9 20:48:28 2019 -0600
[spark-tools] Fix DistributedDataGenerator num-tasks
This patch fixes the num-tasks argument. Previously
Spark was not creating the correct number of tasks
to match the passed argument.
Change-Id: I7fa560e71b2f84a75002e9e776011e1a11c5a1ff
Reviewed-on: http://gerrit.cloudera.org:8080/12211
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
.../scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala
index 2ad4629..ee0954f 100644
--- a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala
+++ b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala
@@ -148,8 +148,8 @@ object DistributedDataGenerator {
val sc = ss.sparkContext
val context = new KuduContext(options.masterAddresses, sc)
val metrics = GeneratorMetrics(sc)
- sc.parallelize(0 until options.numTasks)
- .foreach(taskNum => generateRows(context, options, taskNum, metrics))
+ sc.parallelize(0 until options.numTasks, numSlices = options.numTasks)
+ .foreachPartition(taskNum => generateRows(context, options, taskNum.next(), metrics))
log.info(s"Rows written: ${metrics.rowsWritten.value}")
log.info(s"Collisions: ${metrics.collisions.value}")
}
[kudu] 01/09: Reduce election-related logging
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 8fb302289b7fe8eefa9aa3cd9fb1ae97ac4ae5ec
Author: Will Berkeley <wd...@gmail.org>
AuthorDate: Tue Jan 8 16:03:14 2019 -0800
Reduce election-related logging
Frequent simultaneous pre-elections on lots of tablets are a common
symptom of underlying problems on Kudu clusters. This results in a ton
of log spam because a [pre-]election results in multiple log messages.
This patch makes a few conservative changes to reduce the amount of
election-related logging, especially on candidates.
1. No log message per successful VoteRequest response. The candidate
previously logged a message for every granted or denied vote from a
voter, e.g.
I1221 20:16:10.479441 1852 leader_election.cc:387] T cdd0e560bc5c451c978e38c0d7620e52 P dd658299c5194f4b88ae408cd3130103 [CANDIDATE]: Term 1 pre-election: Vote granted by peer 1a228ec03dbc4bb79ef66cd036a99bca (127.1.92.67:38265)
This seems like overkill. Instead, these messages have been moved to
VLOG(1) and the candidate will print a more detailed summary of the
election once it is decided:
I0109 12:39:42.203627 158707712 leader_election.cc:304] T 00000000000000000000000000000000 P 99b33389ebdc47b1ae18947dea49a5b7 [CANDIDATE]: Term 2 pre-election: Election decided. Result: candidate won. Election summary: received 2 responses out of 3 voters: 2 yes votes; 0 no votes. yes voters: 465a338e21d84ad1a4e03867eb608623, 99b33389ebdc47b1ae18947dea49a5b7; no voters:
Compared to the current state:
I1221 20:16:10.479472 1852 leader_election.cc:278] T cdd0e560bc5c451c978e38c0d7620e52 P dd658299c5194f4b88ae408cd3130103 [CANDIDATE]: Term 1 pre-election: Election decided. Result: candidate won.
Note that it is still possible to find a record of each vote in each
voter's INFO logs, and error condition logging has not been changed.
2. No log message per vote request. The candidate previously logged a
message for each voter it was requesting a vote from. e.g.
I1221 20:16:10.474704 1993 leader_election.cc:251] T cdd0e560bc5c451c978e38c0d7620e52 P 1a228ec03dbc4bb79ef66cd036a99bca [CANDIDATE]: Term 1 pre-election: Requesting pre-vote from peer dd658299c5194f4b88ae408cd3130103 (127.1.92.66:39093)
I1221 20:16:10.475266 1993 leader_election.cc:251] T cdd0e560bc5c451c978e38c0d7620e52 P 1a228ec03dbc4bb79ef66cd036a99bca [CANDIDATE]: Term 1 pre-election: Requesting pre-vote from peer c624ece2977e486fbab958afb000f122 (127.1.92.65:45651)
Instead, the candidate will summarize all the voters it sent requests
to:
I0109 12:39:42.206408 156561408 leader_election.cc:291] T 00000000000000000000000000000000 P 99b33389ebdc47b1ae18947dea49a5b7 [CANDIDATE]: Term 2 election: Requested vote from peers 465a338e21d84ad1a4e03867eb608623 (127.0.0.1:7053), 08b8de9c7a7c4ec5b72c9eb275d29cfd (127.0.0.1:7052)
3. No logging of failure detector sleeps. It seems like it might be
useful to log when the FD sleeps, and why, e.g.
I1221 20:16:12.399258 2803 raft_consensus.cc:2834] T cdd0e560bc5c451c978e38c0d7620e52 P c624ece2977e486fbab958afb000f122: Snoozing failure detection for 0.224s (vote granted)
But frankly in practice it's never been relevant or useful. Since a
message like this is produced multiple times on the candidate and
possibly on the voters during an election, I moved it to VLOG(1)
level to reduce the total amount of logging from elections.
Change-Id: Ie0ab1a37c2b2e9a3f37baf74772b694b907c6adf
Reviewed-on: http://gerrit.cloudera.org:8080/12200
Tested-by: Kudu Jenkins
Reviewed-by: Andrew Wong <aw...@cloudera.com>
---
src/kudu/consensus/leader_election.cc | 50 ++++++++++++++++++++++++++++-------
src/kudu/consensus/leader_election.h | 3 +++
src/kudu/consensus/raft_consensus.cc | 8 +++---
3 files changed, 48 insertions(+), 13 deletions(-)
diff --git a/src/kudu/consensus/leader_election.cc b/src/kudu/consensus/leader_election.cc
index bd2c8a9..931e5ef 100644
--- a/src/kudu/consensus/leader_election.cc
+++ b/src/kudu/consensus/leader_election.cc
@@ -21,6 +21,7 @@
#include <mutex>
#include <ostream>
#include <type_traits>
+#include <utility>
#include <vector>
#include <boost/bind.hpp> // IWYU pragma: keep
@@ -132,6 +133,29 @@ bool VoteCounter::AreAllVotesIn() const {
return GetTotalVotesCounted() == num_voters_;
}
+string VoteCounter::GetElectionSummary() const {
+ vector<string> yes_voter_uuids;
+ vector<string> no_voter_uuids;
+ for (const auto& entry : votes_) {
+ switch (entry.second) {
+ case VOTE_GRANTED:
+ yes_voter_uuids.push_back(entry.first);
+ break;
+ case VOTE_DENIED:
+ no_voter_uuids.push_back(entry.first);
+ break;
+ }
+ }
+ return Substitute("received $0 responses out of $1 voters: $2 yes votes; "
+ "$3 no votes. yes voters: $4; no voters: $5",
+ yes_votes_ + no_votes_,
+ num_voters_,
+ yes_votes_,
+ no_votes_,
+ JoinStrings(yes_voter_uuids, ", "),
+ JoinStrings(no_voter_uuids, ", "));
+}
+
///////////////////////////////////////////////////
// ElectionResult
///////////////////////////////////////////////////
@@ -224,6 +248,8 @@ void LeaderElection::Run() {
CheckForDecision();
// The rest of the code below is for a typical multi-node configuration.
+ vector<string> other_voter_info;
+ other_voter_info.reserve(other_voter_uuids.size());
for (const auto& voter_uuid : other_voter_uuids) {
VoterState* state = nullptr;
{
@@ -232,6 +258,7 @@ void LeaderElection::Run() {
// Safe to drop the lock because voter_state_ is not mutated outside of
// the constructor / destructor. We do this to avoid deadlocks below.
}
+ other_voter_info.push_back(state->PeerInfo());
// If we failed to construct the proxy, just record a 'NO' vote with the status
// that indicates why it failed.
@@ -248,9 +275,6 @@ void LeaderElection::Run() {
}
// Send the RPC request.
- LOG_WITH_PREFIX(INFO) << "Requesting "
- << (request_.is_pre_election() ? "pre-" : "")
- << "vote from peer " << state->PeerInfo();
state->rpc.set_timeout(timeout_);
state->request = request_;
@@ -265,6 +289,9 @@ void LeaderElection::Run() {
boost::bind(&Closure::Run,
Bind(&LeaderElection::VoteResponseRpcCallback, this, voter_uuid)));
}
+ LOG_WITH_PREFIX(INFO) << Substitute("Requested $0vote from peers $1",
+ request_.is_pre_election() ? "pre-" : "",
+ JoinStrings(other_voter_info, ", "));
}
void LeaderElection::CheckForDecision() {
@@ -275,9 +302,12 @@ void LeaderElection::CheckForDecision() {
if (!result_ && vote_counter_->IsDecided()) {
ElectionVote decision;
CHECK_OK(vote_counter_->GetDecision(&decision));
- LOG_WITH_PREFIX(INFO) << "Election decided. Result: candidate "
- << ((decision == VOTE_GRANTED) ? "won." : "lost.");
- string msg = (decision == VOTE_GRANTED) ?
+ const auto election_won = decision == VOTE_GRANTED;
+ LOG_WITH_PREFIX(INFO) << Substitute("Election decided. Result: candidate $0. "
+ "Election summary: $1",
+ election_won ? "won" : "lost",
+ vote_counter_->GetElectionSummary());
+ string msg = election_won ?
"achieved majority votes" : "could not achieve majority";
result_.reset(new ElectionResult(request_, decision, highest_voter_term_, msg));
}
@@ -384,7 +414,7 @@ void LeaderElection::HandleVoteGrantedUnlocked(const VoterState& state) {
}
DCHECK(state.response.vote_granted());
- LOG_WITH_PREFIX(INFO) << "Vote granted by peer " << state.PeerInfo();
+ VLOG_WITH_PREFIX(1) << "Vote granted by peer " << state.PeerInfo();
RecordVoteUnlocked(state, VOTE_GRANTED);
}
@@ -398,8 +428,10 @@ void LeaderElection::HandleVoteDeniedUnlocked(const VoterState& state) {
return HandleHigherTermUnlocked(state);
}
- LOG_WITH_PREFIX(INFO) << "Vote denied by peer " << state.PeerInfo() << ". Message: "
- << StatusFromPB(state.response.consensus_error().status()).ToString();
+ VLOG_WITH_PREFIX(1) << Substitute(
+ "Vote denied by peer $0. Message: $1",
+ state.PeerInfo(),
+ StatusFromPB(state.response.consensus_error().status()).ToString());
RecordVoteUnlocked(state, VOTE_DENIED);
}
diff --git a/src/kudu/consensus/leader_election.h b/src/kudu/consensus/leader_election.h
index 5e145e2..8770906 100644
--- a/src/kudu/consensus/leader_election.h
+++ b/src/kudu/consensus/leader_election.h
@@ -78,6 +78,9 @@ class VoteCounter {
// Return true iff GetTotalVotesCounted() == num_voters_;
bool AreAllVotesIn() const;
+ // Return a summary of the election so far, suitable for logging.
+ std::string GetElectionSummary() const;
+
private:
friend class VoteCounterTest;
diff --git a/src/kudu/consensus/raft_consensus.cc b/src/kudu/consensus/raft_consensus.cc
index d71f8e9..7428fe0 100644
--- a/src/kudu/consensus/raft_consensus.cc
+++ b/src/kudu/consensus/raft_consensus.cc
@@ -2831,10 +2831,10 @@ void RaftConsensus::SnoozeFailureDetector(boost::optional<string> reason_for_log
boost::optional<MonoDelta> delta) {
if (PREDICT_TRUE(failure_detector_ && FLAGS_enable_leader_failure_detection)) {
if (reason_for_log) {
- LOG(INFO) << LogPrefixThreadSafe()
- << Substitute("Snoozing failure detection for $0 ($1)",
- delta ? delta->ToString() : "election timeout",
- *reason_for_log);
+ VLOG(1) << LogPrefixThreadSafe()
+ << Substitute("Snoozing failure detection for $0 ($1)",
+ delta ? delta->ToString() : "election timeout",
+ *reason_for_log);
}
if (!delta) {
[kudu] 02/09: Assign locations to tablet servers and the client in
Java
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit eae3504a05fd4931418398b99d6d8a97d16d7ed1
Author: Will Berkeley <wd...@gmail.org>
AuthorDate: Wed Dec 26 16:16:10 2018 -0500
Assign locations to tablet servers and the client in Java
Change-Id: I9e2c74ab12f7da187bf6e75d42a3089bc20235db
Reviewed-on: http://gerrit.cloudera.org:8080/12174
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
---
.../org/apache/kudu/client/AsyncKuduClient.java | 35 +++++++++++--
.../java/org/apache/kudu/client/KuduClient.java | 9 ++++
.../java/org/apache/kudu/client/ServerInfo.java | 15 +++++-
.../apache/kudu/client/TestConnectionCache.java | 6 ++-
.../org/apache/kudu/client/TestKuduClient.java | 22 +++++++-
.../org/apache/kudu/client/TestRemoteTablet.java | 3 +-
.../org/apache/kudu/client/TestServerInfo.java | 37 ++++++++------
.../src/test/resources/assign-location.py | 1 +
.../java/org/apache/kudu/test/KuduTestHarness.java | 20 ++++++++
.../apache/kudu/test/cluster/MiniKuduCluster.java | 59 +++++++++++++++++-----
10 files changed, 170 insertions(+), 37 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
index 2807ef5..e71293b 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
@@ -323,6 +323,13 @@ public class AsyncKuduClient implements AutoCloseable {
private volatile boolean hasConnectedToMaster = false;
/**
+ * The location of this client as assigned by the leader master.
+ *
+ * If no location is assigned, will be an empty string.
+ */
+ private String location = "";
+
+ /**
* Semaphore used to rate-limit master lookups
* Once we have more than this number of concurrent master lookups, we'll
* start to throttle ourselves slightly.
@@ -422,7 +429,11 @@ public class AsyncKuduClient implements AutoCloseable {
return null;
}
return newRpcProxy(
- new ServerInfo(getFakeMasterUuid(hostPort), hostPort, inetAddress), credentialsPolicy);
+ new ServerInfo(getFakeMasterUuid(hostPort),
+ hostPort,
+ inetAddress,
+ /*locaton=*/""),
+ credentialsPolicy);
}
static String getFakeMasterUuid(HostAndPort hostPort) {
@@ -440,7 +451,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
/**
- * Report on the token re-acqusition results. The result authn token might be null: in that
+ * Report on the token re-acquisition results. The result authn token might be null: in that
* case the SASL credentials will be used to negotiate future connections.
*/
@Override
@@ -455,6 +466,9 @@ public class AsyncKuduClient implements AutoCloseable {
securityContext.setAuthenticationToken(null);
cb.call(false);
}
+ synchronized (AsyncKuduClient.this) {
+ location = masterResponsePB.getClientLocation();
+ }
return null;
}
}
@@ -499,6 +513,16 @@ public class AsyncKuduClient implements AutoCloseable {
}
/**
+ * Returns a string representation of this client's location. If this
+ * client was not assigned a location, returns the empty string.
+ *
+ * @return a string representation of this client's location
+ */
+ public String getLocationString() {
+ return location;
+ }
+
+ /**
* Returns a synchronous {@link KuduClient} which wraps this asynchronous client.
* Calling {@link KuduClient#close} on the returned client will close this client.
* If this asynchronous client should outlive the returned synchronous client,
@@ -1716,6 +1740,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
synchronized (AsyncKuduClient.this) {
AsyncKuduClient.this.hiveMetastoreConfig = hiveMetastoreConfig;
+ location = respPb.getClientLocation();
}
hasConnectedToMaster = true;
@@ -1955,8 +1980,8 @@ public class AsyncKuduClient implements AutoCloseable {
return null;
}
- // from meta_cache.cc
- // TODO: if the TS advertises multiple host/ports, pick the right one
+ // From meta_cache.cc:
+ // TODO: If the TS advertises multiple host/ports, pick the right one
// based on some kind of policy. For now just use the first always.
final HostAndPort hostPort = ProtobufHelper.hostAndPortFromPB(addresses.get(0));
final InetAddress inetAddress = NetUtil.getInetAddress(hostPort.getHost());
@@ -1964,7 +1989,7 @@ public class AsyncKuduClient implements AutoCloseable {
throw new UnknownHostException(
"Failed to resolve the IP of `" + addresses.get(0).getHost() + "'");
}
- return new ServerInfo(uuid, hostPort, inetAddress);
+ return new ServerInfo(uuid, hostPort, inetAddress, tsInfoPB.getLocation());
}
/** Callback executed when a master lookup completes. */
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
index dfe1e40..b21bbfb 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduClient.java
@@ -82,6 +82,15 @@ public class KuduClient implements AutoCloseable {
return asyncClient.hasLastPropagatedTimestamp();
}
+ /**
+ * Returns a string representation of this client's location. If this
+ * client was not assigned a location, returns the empty string.
+ *
+ * @return a string representation of this client's location
+ */
+ public String getLocationString() {
+ return asyncClient.getLocationString();
+ }
/**
* Returns the Hive Metastore configuration of the cluster.
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
index e2e4162..67b2963 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ServerInfo.java
@@ -35,6 +35,7 @@ public class ServerInfo {
private final String uuid;
private final HostAndPort hostPort;
private final InetSocketAddress resolvedAddr;
+ private final String location;
private final boolean local;
private static final ConcurrentHashMap<InetAddress, Boolean> isLocalAddressCache =
new ConcurrentHashMap<>();
@@ -45,13 +46,17 @@ public class ServerInfo {
* @param uuid server's UUID
* @param hostPort server's hostname and port
* @param resolvedAddr resolved address used to check if the server is local
+ * @param location the location assigned by the leader master, or an empty string if no location
+ * is assigned
*/
- public ServerInfo(String uuid, HostAndPort hostPort, InetAddress resolvedAddr) {
+ public ServerInfo(String uuid, HostAndPort hostPort, InetAddress resolvedAddr, String location) {
Preconditions.checkNotNull(uuid);
Preconditions.checkArgument(hostPort.getPort() > 0);
+ Preconditions.checkNotNull(location);
this.uuid = uuid;
this.hostPort = hostPort;
this.resolvedAddr = new InetSocketAddress(resolvedAddr, hostPort.getPort());
+ this.location = location;
Boolean isLocal = isLocalAddressCache.get(resolvedAddr);
if (isLocal == null) {
isLocal = NetUtil.isLocalAddress(resolvedAddr);
@@ -89,6 +94,14 @@ public class ServerInfo {
}
/**
+ * Returns this server's location.
+ * @return the server's location, or the empty string if no location was assigned.
+ */
+ public String getLocation() {
+ return location;
+ }
+
+ /**
* Returns if this server is on this client's host.
* @return true if the server is local, else false
*/
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
index 7cdc6f0..134b814 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestConnectionCache.java
@@ -49,8 +49,10 @@ public class TestConnectionCache {
client.getTablesList().join();
HostAndPort masterHostPort = cluster.getMasterServers().get(0);
- ServerInfo firstMaster = new ServerInfo("fake-uuid", masterHostPort,
- NetUtil.getInetAddress(masterHostPort.getHost()));
+ ServerInfo firstMaster = new ServerInfo("fake-uuid",
+ masterHostPort,
+ NetUtil.getInetAddress(masterHostPort.getHost()),
+ /*location=*/"");
// 3 masters in the cluster. Connections should have been cached since we forced
// a cluster connection above.
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
index 4901738..97df3e0 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
@@ -56,6 +56,7 @@ import com.google.common.collect.ImmutableList;
import com.stumbleupon.async.Deferred;
import org.apache.kudu.test.KuduTestHarness;
+import org.apache.kudu.test.KuduTestHarness.LocationConfig;
import org.apache.kudu.test.KuduTestHarness.TabletServerConfig;
import org.apache.kudu.test.ClientTestUtil;
import org.apache.kudu.util.TimestampUtil;
@@ -1113,7 +1114,7 @@ public class TestKuduClient {
future.get();
}
}
-
+
private void runTestCallDuringLeaderElection(String clientMethodName) throws Exception {
// This bit of reflection helps us avoid duplicating test code.
Method methodToInvoke = KuduClient.class.getMethod(clientMethodName);
@@ -1153,4 +1154,23 @@ public class TestKuduClient {
public void testGetHiveMetastoreConfigDuringLeaderElection() throws Exception {
runTestCallDuringLeaderElection("getHiveMetastoreConfig");
}
+ /**
+ * Test assignment of a location to the client.
+ */
+ @Test(timeout = 100000)
+ public void testClientLocationNoLocation() throws Exception {
+ // Do something that will cause the client to connect to the cluster.
+ client.listTabletServers();
+ assertEquals("", client.getLocationString());
+ }
+
+ @Test(timeout = 100000)
+ @LocationConfig(locations = {
+ "/L0:4",
+ })
+ public void testClientLocation() throws Exception {
+ // Do something that will cause the client to connect to the cluster.
+ client.listTabletServers();
+ assertEquals("/L0", client.getLocationString());
+ }
}
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
index a9b3747..f1c09ab 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestRemoteTablet.java
@@ -173,7 +173,8 @@ public class TestRemoteTablet {
String uuid = kUuids[i];
servers.add(new ServerInfo(uuid,
new HostAndPort("host", 1000 + i),
- addr));
+ addr,
+ /*location=*/""));
tabletPb.addReplicas(ProtobufUtils.getFakeTabletReplicaPB(
uuid, "host", i,
leaderIndex == i ? Metadata.RaftPeerPB.Role.LEADER : Metadata.RaftPeerPB.Role.FOLLOWER));
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java
index 233ae17..63fd093 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestServerInfo.java
@@ -1,16 +1,19 @@
-/**
- * Licensed 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. See accompanying LICENSE file.
- */
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
package org.apache.kudu.client;
import java.net.InetAddress;
@@ -32,9 +35,11 @@ public class TestServerInfo {
ServerInfo serverInfo = new ServerInfo(
"nevermind",
new HostAndPort("master2.example.com", 12345),
- InetAddress.getByName("10.1.2.3"));
+ InetAddress.getByName("10.1.2.3"),
+ /*location=*/"");
Assert.assertEquals("master2.example.com", serverInfo.getAndCanonicalizeHostname());
+ Assert.assertEquals("", serverInfo.getLocation());
}
/**
@@ -48,9 +53,11 @@ public class TestServerInfo {
ServerInfo serverInfo = new ServerInfo(
"abcdef", // uuid
new HostAndPort("master1.example.com", 12345),
- InetAddress.getByName("10.1.2.3"));
+ InetAddress.getByName("10.1.2.3"),
+ /* location= */"/foo");
Assert.assertEquals("server123.example.com", serverInfo.getAndCanonicalizeHostname());
+ Assert.assertEquals("/foo", serverInfo.getLocation());
Assert.assertEquals("abcdef(master1.example.com:12345)", serverInfo.toString());
}
diff --git a/java/kudu-client/src/test/resources/assign-location.py b/java/kudu-client/src/test/resources/assign-location.py
new file mode 120000
index 0000000..4196f51
--- /dev/null
+++ b/java/kudu-client/src/test/resources/assign-location.py
@@ -0,0 +1 @@
+../../../../../src/kudu/scripts/assign-location.py
\ No newline at end of file
diff --git a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/KuduTestHarness.java b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/KuduTestHarness.java
index 172989e..ed45bb5 100644
--- a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/KuduTestHarness.java
+++ b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/KuduTestHarness.java
@@ -46,6 +46,7 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
@@ -117,6 +118,13 @@ public class KuduTestHarness extends ExternalResource {
clusterBuilder.addMasterServerFlag(flag);
}
}
+ // Pass through any location mapping defined in the method level annotation.
+ LocationConfig locationConfig = description.getAnnotation(LocationConfig.class);
+ if (locationConfig != null) {
+ for (String location : locationConfig.locations()) {
+ clusterBuilder.addLocation(location);
+ }
+ }
// Set any tablet server flags defined in the method level annotation.
TabletServerConfig tabletServerConfig = description.getAnnotation(TabletServerConfig.class);
if (tabletServerConfig != null) {
@@ -441,4 +449,16 @@ public class KuduTestHarness extends ExternalResource {
public @interface TabletServerConfig {
String[] flags();
}
+
+ /**
+ * An annotation that can be added to each test method to
+ * define a location mapping for the cluster. Location
+ * mappings are defined as a series of 'location:number'
+ * pairs.
+ */
+ @Retention(RetentionPolicy.RUNTIME)
+ @Target({ElementType.METHOD})
+ public @interface LocationConfig {
+ String[] locations();
+ }
}
diff --git a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/MiniKuduCluster.java b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/MiniKuduCluster.java
index 793e65d..5dd7b12 100644
--- a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/MiniKuduCluster.java
+++ b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/MiniKuduCluster.java
@@ -102,6 +102,7 @@ public class MiniKuduCluster implements AutoCloseable {
private final int numTservers;
private final ImmutableList<String> extraTserverFlags;
private final ImmutableList<String> extraMasterFlags;
+ private final ImmutableList<String> locationInfo;
private final String clusterRoot;
private MiniKdcOptionsPB kdcOptionsPb;
@@ -112,6 +113,7 @@ public class MiniKuduCluster implements AutoCloseable {
int numTservers,
List<String> extraTserverFlags,
List<String> extraMasterFlags,
+ List<String> locationInfo,
MiniKdcOptionsPB kdcOptionsPb,
String clusterRoot,
Common.HmsMode hmsMode) {
@@ -120,11 +122,12 @@ public class MiniKuduCluster implements AutoCloseable {
this.numTservers = numTservers;
this.extraTserverFlags = ImmutableList.copyOf(extraTserverFlags);
this.extraMasterFlags = ImmutableList.copyOf(extraMasterFlags);
+ this.locationInfo = ImmutableList.copyOf(locationInfo);
this.kdcOptionsPb = kdcOptionsPb;
this.hmsMode = hmsMode;
if (clusterRoot == null) {
- // If a cluster root was not set, create a unique temp directory to use.
+ // If a cluster root was not set, create a unique temp directory to use.
// The mini cluster will clean this directory up on exit.
try {
File tempRoot = TempDirUtils.getTempDirectory("mini-kudu-cluster");
@@ -196,19 +199,35 @@ public class MiniKuduCluster implements AutoCloseable {
miniClusterErrorPrinter.setName("cluster stderr printer");
miniClusterErrorPrinter.start();
+ CreateClusterRequestPB.Builder createClusterRequestBuilder = CreateClusterRequestPB.newBuilder()
+ .setNumMasters(numMasters)
+ .setNumTservers(numTservers)
+ .setEnableKerberos(enableKerberos)
+ .setHmsMode(hmsMode)
+ .addAllExtraMasterFlags(extraMasterFlags)
+ .addAllExtraTserverFlags(extraTserverFlags)
+ .setMiniKdcOptions(kdcOptionsPb)
+ .setClusterRoot(clusterRoot);
+
+ // Set up the location mapping command flag if there is location info.
+ if (!locationInfo.isEmpty()) {
+ List<String> locationMappingCmd = new ArrayList<>();
+ locationMappingCmd.add(getClass().getResource("/assign-location.py").getFile());
+ String locationMappingCmdPath =
+ Paths.get(clusterRoot, "location-assignment.state").toString();
+ locationMappingCmd.add("--state_store=" + locationMappingCmdPath);
+ for (String location : locationInfo) {
+ locationMappingCmd.add("--map " + location);
+ }
+ String locationMappingCmdFlag = "--location_mapping_cmd=" +
+ Joiner.on(" ").join(locationMappingCmd);
+ createClusterRequestBuilder.addExtraMasterFlags(locationMappingCmdFlag);
+ }
+
// Create and start the cluster.
sendRequestToCluster(
ControlShellRequestPB.newBuilder()
- .setCreateCluster(CreateClusterRequestPB.newBuilder()
- .setNumMasters(numMasters)
- .setNumTservers(numTservers)
- .setEnableKerberos(enableKerberos)
- .setHmsMode(hmsMode)
- .addAllExtraMasterFlags(extraMasterFlags)
- .addAllExtraTserverFlags(extraTserverFlags)
- .setMiniKdcOptions(kdcOptionsPb)
- .setClusterRoot(clusterRoot)
- .build())
+ .setCreateCluster(createClusterRequestBuilder.build())
.build());
sendRequestToCluster(
ControlShellRequestPB.newBuilder()
@@ -540,6 +559,7 @@ public class MiniKuduCluster implements AutoCloseable {
private boolean enableKerberos = false;
private final List<String> extraTabletServerFlags = new ArrayList<>();
private final List<String> extraMasterServerFlags = new ArrayList<>();
+ private final List<String> locationInfo = new ArrayList<>();
private String clusterRoot = null;
private MiniKdcOptionsPB.Builder kdcOptionsPb = MiniKdcOptionsPB.newBuilder();
@@ -587,6 +607,21 @@ public class MiniKuduCluster implements AutoCloseable {
return this;
}
+ /**
+ * Adds one location to the minicluster configuration, consisting of a
+ * location and the total number of tablet servers and clients that
+ * can be assigned to the location. The 'location' string should be
+ * in the form 'location:number'. For example,
+ * "/L0:2"
+ * will add a location "/L0" that will accept up to two clients or
+ * tablet servers registered in it.
+ * @return this instance
+ */
+ public MiniKuduClusterBuilder addLocation(String location) {
+ locationInfo.add(location);
+ return this;
+ }
+
public MiniKuduClusterBuilder kdcTicketLifetime(String lifetime) {
this.kdcOptionsPb.setTicketLifetime(lifetime);
return this;
@@ -615,7 +650,7 @@ public class MiniKuduCluster implements AutoCloseable {
MiniKuduCluster cluster =
new MiniKuduCluster(enableKerberos,
numMasterServers, numTabletServers,
- extraTabletServerFlags, extraMasterServerFlags,
+ extraTabletServerFlags, extraMasterServerFlags, locationInfo,
kdcOptionsPb.build(), clusterRoot, hmsMode);
try {
cluster.start();