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 2019/02/28 20:29:40 UTC
[kudu] branch master updated (8f85019 -> 4ace917)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 8f85019 KUDU-2645: schema: Validate that defaults are provided for virtual columns
new 8fbf1cc Rename DeadlineTracker to TimeoutTracker
new ae6bbca [master] introduce cache for location mapping assignments
new 4ace917 [master] use cache for assigned locations
The 3 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:
.../org/apache/kudu/client/AlterTableRequest.java | 2 +-
.../org/apache/kudu/client/AsyncKuduClient.java | 38 ++---
.../org/apache/kudu/client/AsyncKuduScanner.java | 2 +-
.../main/java/org/apache/kudu/client/Batch.java | 6 +-
.../java/org/apache/kudu/client/BatchResponse.java | 2 +-
.../org/apache/kudu/client/ConnectToCluster.java | 2 +-
.../org/apache/kudu/client/CreateTableRequest.java | 2 +-
.../org/apache/kudu/client/DeadlineTracker.java | 159 ------------------
.../org/apache/kudu/client/DeleteTableRequest.java | 2 +-
.../apache/kudu/client/GetTableSchemaRequest.java | 2 +-
.../kudu/client/IsAlterTableDoneRequest.java | 2 +-
.../kudu/client/IsCreateTableDoneRequest.java | 2 +-
.../org/apache/kudu/client/KuduPartitioner.java | 6 +-
.../main/java/org/apache/kudu/client/KuduRpc.java | 10 +-
.../org/apache/kudu/client/ListTablesRequest.java | 2 +-
.../kudu/client/ListTabletServersRequest.java | 2 +-
.../org/apache/kudu/client/ListTabletsRequest.java | 2 +-
.../java/org/apache/kudu/client/Operation.java | 6 +-
.../java/org/apache/kudu/client/PingRequest.java | 2 +-
.../main/java/org/apache/kudu/client/RpcProxy.java | 4 +-
.../org/apache/kudu/client/TimeoutTracker.java | 159 ++++++++++++++++++
.../test/java/org/apache/kudu/client/ITClient.java | 8 +-
.../apache/kudu/client/TestConnectionCache.java | 6 +-
...eadlineTracker.java => TestTimeoutTracker.java} | 22 +--
.../java/org/apache/kudu/test/KuduTestHarness.java | 11 +-
.../org/apache/kudu/test/TestMiniKuduCluster.java | 4 +-
src/kudu/client/client-test.cc | 44 ++++-
.../integration-tests/location_assignment-itest.cc | 70 ++++++++
src/kudu/master/CMakeLists.txt | 2 +
src/kudu/master/location_cache-test.cc | 179 +++++++++++++++++++++
src/kudu/master/location_cache.cc | 152 +++++++++++++++++
src/kudu/master/location_cache.h | 88 ++++++++++
src/kudu/master/master-test.cc | 25 ++-
src/kudu/master/master.cc | 9 +-
src/kudu/master/master.h | 11 +-
src/kudu/master/master_service.cc | 10 +-
src/kudu/master/ts_descriptor-test.cc | 24 +--
src/kudu/master/ts_descriptor.cc | 72 +--------
src/kudu/master/ts_descriptor.h | 16 +-
src/kudu/master/ts_manager.cc | 9 +-
src/kudu/master/ts_manager.h | 10 +-
src/kudu/tools/ksck_remote-test.cc | 37 +++--
42 files changed, 863 insertions(+), 360 deletions(-)
delete mode 100644 java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
create mode 100644 java/kudu-client/src/main/java/org/apache/kudu/client/TimeoutTracker.java
rename java/kudu-client/src/test/java/org/apache/kudu/client/{TestDeadlineTracker.java => TestTimeoutTracker.java} (82%)
create mode 100644 src/kudu/master/location_cache-test.cc
create mode 100644 src/kudu/master/location_cache.cc
create mode 100644 src/kudu/master/location_cache.h
[kudu] 02/03: [master] introduce cache for location mapping
assignments
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit ae6bbcaabd20955119f1d945d276589538dae928
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Wed Feb 27 21:53:43 2019 -0800
[master] introduce cache for location mapping assignments
This changelist adds a very primitive cache for location assignments.
The cache does not prevent running multiple commands for the same
location key if an entry is not present in the cache.
A unit test is also added.
Change-Id: Icb5c436c9433acd87c44c4d81982420f33ebb4a4
Reviewed-on: http://gerrit.cloudera.org:8080/12634
Reviewed-by: Will Berkeley <wd...@gmail.com>
Tested-by: Kudu Jenkins
---
src/kudu/master/CMakeLists.txt | 2 +
src/kudu/master/location_cache-test.cc | 179 +++++++++++++++++++++++++++++++++
src/kudu/master/location_cache.cc | 152 ++++++++++++++++++++++++++++
src/kudu/master/location_cache.h | 88 ++++++++++++++++
4 files changed, 421 insertions(+)
diff --git a/src/kudu/master/CMakeLists.txt b/src/kudu/master/CMakeLists.txt
index 89fd9c6..79f38ce 100644
--- a/src/kudu/master/CMakeLists.txt
+++ b/src/kudu/master/CMakeLists.txt
@@ -35,6 +35,7 @@ ADD_EXPORTABLE_LIBRARY(master_proto
set(MASTER_SRCS
catalog_manager.cc
hms_notification_log_listener.cc
+ location_cache.cc
master.cc
master_cert_authority.cc
master_options.cc
@@ -79,6 +80,7 @@ SET_KUDU_TEST_LINK_LIBS(
ADD_KUDU_TEST(catalog_manager-test)
ADD_KUDU_TEST(hms_notification_log_listener-test)
+ADD_KUDU_TEST(location_cache-test DATA_FILES ../scripts/first_argument.sh)
ADD_KUDU_TEST(master-test RESOURCE_LOCK "master-web-port"
DATA_FILES ../scripts/first_argument.sh)
ADD_KUDU_TEST(mini_master-test RESOURCE_LOCK "master-web-port")
diff --git a/src/kudu/master/location_cache-test.cc b/src/kudu/master/location_cache-test.cc
new file mode 100644
index 0000000..96c1bea
--- /dev/null
+++ b/src/kudu/master/location_cache-test.cc
@@ -0,0 +1,179 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "kudu/master/location_cache.h"
+
+#include <cstdint>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include "kudu/gutil/ref_counted.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/util/metrics.h"
+#include "kudu/util/path_util.h"
+#include "kudu/util/status.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
+
+METRIC_DECLARE_counter(location_mapping_cache_hits);
+METRIC_DECLARE_counter(location_mapping_cache_queries);
+
+using std::string;
+using std::thread;
+using std::vector;
+using strings::Substitute;
+
+namespace kudu {
+namespace master {
+
+// Targeted test for LocationCache.
+class LocationCacheTest : public KuduTest {
+ protected:
+ void SetUp() override {
+ KuduTest::SetUp();
+ metric_entity_ = METRIC_ENTITY_server.Instantiate(&metric_registry_,
+ "LocationCacheTest");
+ }
+
+ void CheckMetrics(int64_t expected_queries, int64_t expected_hits) {
+ scoped_refptr<Counter> cache_queries(metric_entity_->FindOrCreateCounter(
+ &METRIC_location_mapping_cache_queries));
+ ASSERT_NE(nullptr, cache_queries.get());
+ ASSERT_EQ(expected_queries, cache_queries->value());
+
+ scoped_refptr<Counter> cache_hits(metric_entity_->FindOrCreateCounter(
+ &METRIC_location_mapping_cache_hits));
+ ASSERT_NE(nullptr, cache_hits.get());
+ ASSERT_EQ(expected_hits, cache_hits->value());
+ }
+
+ MetricRegistry metric_registry_;
+ scoped_refptr<MetricEntity> metric_entity_;
+};
+
+TEST_F(LocationCacheTest, EmptyMappingCommand) {
+ LocationCache cache(" ", metric_entity_.get());
+ string location;
+ auto s = cache.GetLocation("na", &location);
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+ ASSERT_STR_CONTAINS(
+ s.ToString(), "invalid empty location mapping command");
+ NO_FATALS(CheckMetrics(1, 0));
+}
+
+TEST_F(LocationCacheTest, MappingCommandFailureExitStatus) {
+ LocationCache cache("/sbin/nologin", metric_entity_.get());
+ string location;
+ auto s = cache.GetLocation("na", &location);
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+ ASSERT_STR_CONTAINS(
+ s.ToString(), "failed to run location mapping command: ");
+ NO_FATALS(CheckMetrics(1, 0));
+}
+
+TEST_F(LocationCacheTest, MappingCommandEmptyOutput) {
+ LocationCache cache("/bin/cat", metric_entity_.get());
+ string location;
+ auto s = cache.GetLocation("/dev/null", &location);
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+ ASSERT_STR_CONTAINS(
+ s.ToString(), "location mapping command returned invalid empty location");
+ NO_FATALS(CheckMetrics(1, 0));
+}
+
+TEST_F(LocationCacheTest, MappingCommandReturnsInvalidLocation) {
+ const string cmd_path = JoinPathSegments(GetTestExecutableDirectory(),
+ "testdata/first_argument.sh");
+ const string location_mapping_cmd = Substitute("$0 invalid.location",
+ cmd_path);
+ LocationCache cache(location_mapping_cmd, metric_entity_.get());
+ string location;
+ auto s = cache.GetLocation("na", &location);
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+ ASSERT_STR_CONTAINS(
+ s.ToString(), "location mapping command returned invalid location");
+ NO_FATALS(CheckMetrics(1, 0));
+}
+
+TEST_F(LocationCacheTest, HappyPath) {
+ const string kRefLocation = "/ref_location";
+ const string cmd_path = JoinPathSegments(GetTestExecutableDirectory(),
+ "testdata/first_argument.sh");
+ const string location_mapping_cmd = Substitute("$0 $1",
+ cmd_path, kRefLocation);
+ LocationCache cache(location_mapping_cmd, metric_entity_.get());
+ NO_FATALS(CheckMetrics(0, 0));
+
+ string location;
+ auto s = cache.GetLocation("key_0", &location);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(kRefLocation, location);
+ NO_FATALS(CheckMetrics(1, 0));
+
+ s = cache.GetLocation("key_1", &location);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(kRefLocation, location);
+ NO_FATALS(CheckMetrics(2, 0));
+
+ s = cache.GetLocation("key_1", &location);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(kRefLocation, location);
+ NO_FATALS(CheckMetrics(3, 1));
+
+ s = cache.GetLocation("key_0", &location);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(kRefLocation, location);
+ NO_FATALS(CheckMetrics(4, 2));
+}
+
+TEST_F(LocationCacheTest, ConcurrentRequests) {
+ static constexpr auto kNumThreads = 32;
+ const string kRefLocation = "/ref_location";
+ const string cmd_path = JoinPathSegments(GetTestExecutableDirectory(),
+ "testdata/first_argument.sh");
+ const string location_mapping_cmd = Substitute("$0 $1",
+ cmd_path, kRefLocation);
+ LocationCache cache(location_mapping_cmd, metric_entity_.get());
+ NO_FATALS(CheckMetrics(0, 0));
+
+ for (auto iter = 0; iter < 2; ++iter) {
+ vector<thread> threads;
+ threads.reserve(kNumThreads);
+ for (auto idx = 0; idx < kNumThreads; ++idx) {
+ threads.emplace_back([&cache, &kRefLocation, idx]() {
+ string location;
+ auto s = cache.GetLocation(Substitute("key_$0", idx), &location);
+ CHECK(s.ok()) << s.ToString();
+ CHECK_EQ(kRefLocation, location);
+ });
+ }
+ for (auto& t : threads) {
+ t.join();
+ }
+ // Expecting to account for every query, and the follow-up iteration
+ // should result in every query hitting the cache.
+ NO_FATALS(CheckMetrics(kNumThreads * (iter + 1),
+ kNumThreads * iter));
+ }
+}
+
+} // namespace master
+} // namespace kudu
diff --git a/src/kudu/master/location_cache.cc b/src/kudu/master/location_cache.cc
new file mode 100644
index 0000000..b79b259
--- /dev/null
+++ b/src/kudu/master/location_cache.cc
@@ -0,0 +1,152 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "kudu/master/location_cache.h"
+
+#include <cstdio>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <glog/logging.h>
+
+#include "kudu/gutil/map-util.h"
+#include "kudu/gutil/port.h"
+#include "kudu/gutil/strings/charset.h"
+#include "kudu/gutil/strings/split.h"
+#include "kudu/gutil/strings/strip.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/util/subprocess.h"
+#include "kudu/util/trace.h"
+
+METRIC_DEFINE_counter(server, location_mapping_cache_hits,
+ "Location Mapping Cache Hits",
+ kudu::MetricUnit::kCacheHits,
+ "Number of times location mapping assignment used "
+ "cached data");
+METRIC_DEFINE_counter(server, location_mapping_cache_queries,
+ "Location Mapping Cache Queries",
+ kudu::MetricUnit::kCacheQueries,
+ "Number of queries to the location mapping cache");
+
+using std::string;
+using std::vector;
+using strings::Substitute;
+
+namespace kudu {
+namespace master {
+
+namespace {
+// Returns if 'location' is a valid location string, i.e. it begins with /
+// and consists of /-separated tokens each of which contains only characters
+// from the set [a-zA-Z0-9_-.].
+bool IsValidLocation(const string& location) {
+ if (location.empty() || location[0] != '/') {
+ return false;
+ }
+ const strings::CharSet charset("abcdefghijklmnopqrstuvwxyz"
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+ "0123456789"
+ "_-./");
+ for (const auto c : location) {
+ if (!charset.Test(c)) {
+ return false;
+ }
+ }
+ return true;
+}
+} // anonymous namespace
+
+LocationCache::LocationCache(string location_mapping_cmd,
+ MetricEntity* metric_entity)
+ : location_mapping_cmd_(std::move(location_mapping_cmd)) {
+ if (metric_entity != nullptr) {
+ location_mapping_cache_hits_ = metric_entity->FindOrCreateCounter(
+ &METRIC_location_mapping_cache_hits);
+ location_mapping_cache_queries_ = metric_entity->FindOrCreateCounter(
+ &METRIC_location_mapping_cache_queries);
+ }
+}
+
+Status LocationCache::GetLocation(const string& key, string* location) {
+ if (PREDICT_TRUE(location_mapping_cache_queries_)) {
+ location_mapping_cache_queries_->Increment();
+ }
+ {
+ // First check whether the location for the key has already been assigned.
+ shared_lock<rw_spinlock> l(location_map_lock_);
+ const auto* value_ptr = FindOrNull(location_map_, key);
+ if (value_ptr) {
+ DCHECK(!value_ptr->empty());
+ *location = *value_ptr;
+ if (PREDICT_TRUE(location_mapping_cache_hits_)) {
+ location_mapping_cache_hits_->Increment();
+ }
+ return Status::OK();
+ }
+ }
+ string value;
+ TRACE(Substitute("key $0: assigning location", key));
+ Status s = GetLocationFromLocationMappingCmd(
+ location_mapping_cmd_, key, &value);
+ TRACE(Substitute("key $0: assigned location '$1'", key, value));
+ if (s.ok()) {
+ CHECK(!value.empty());
+ std::lock_guard<rw_spinlock> l(location_map_lock_);
+ // This simple implementation doesn't protect against multiple runs of the
+ // location-mapping command for the same key.
+ InsertIfNotPresent(&location_map_, key, value);
+ *location = value;
+ }
+ return s;
+}
+
+Status LocationCache::GetLocationFromLocationMappingCmd(const string& cmd,
+ const string& key,
+ string* location) {
+ DCHECK(location);
+ vector<string> argv = strings::Split(cmd, " ", strings::SkipEmpty());
+ if (argv.empty()) {
+ return Status::RuntimeError("invalid empty location mapping command");
+ }
+ argv.push_back(key);
+ string stderr, location_temp;
+ Status s = Subprocess::Call(argv, /*stdin_in=*/"", &location_temp, &stderr);
+ if (!s.ok()) {
+ return Status::RuntimeError(
+ Substitute("failed to run location mapping command: $0", s.ToString()),
+ stderr);
+ }
+ StripWhiteSpace(&location_temp);
+ // Special case an empty location for a better error.
+ if (location_temp.empty()) {
+ return Status::RuntimeError(
+ "location mapping command returned invalid empty location");
+ }
+ if (!IsValidLocation(location_temp)) {
+ return Status::RuntimeError(
+ "location mapping command returned invalid location",
+ location_temp);
+ }
+ *location = std::move(location_temp);
+ return Status::OK();
+}
+
+} // namespace master
+} // namespace kudu
diff --git a/src/kudu/master/location_cache.h b/src/kudu/master/location_cache.h
new file mode 100644
index 0000000..b89f2d9
--- /dev/null
+++ b/src/kudu/master/location_cache.h
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+#pragma once
+
+#include <string>
+#include <unordered_map>
+
+#include "kudu/gutil/macros.h"
+#include "kudu/gutil/ref_counted.h"
+#include "kudu/util/locks.h"
+#include "kudu/util/metrics.h"
+#include "kudu/util/status.h"
+
+namespace kudu {
+namespace master {
+
+// A primitive cache of unlimited capacity to store assigned locations for
+// a key. The cache entries are kept in the cache for the lifetime of the cache
+// itself.
+class LocationCache {
+ public:
+ // The location assignment command is specified by the 'location_mapping_cmd'
+ // parameter (the command might be a script or an executable). The
+ // 'metric_entity' is used to register standard cache counters: total number
+ // of queries and number of cache hits during the cache's lifetime.
+ explicit LocationCache(std::string location_mapping_cmd,
+ MetricEntity* metric_entity = nullptr);
+ ~LocationCache() = default;
+
+ // Get the location for the specified key. The key is treated as an opaque
+ // identifier.
+ //
+ // If no cached location is found, the location mapping command is run,
+ // caching the result for the lifetime of the cache.
+ //
+ // This method returns an error if there was an issue running the location
+ // assignment command.
+ Status GetLocation(const std::string& key, std::string* location);
+
+ private:
+ // Resolves an opaque 'key' into a location using the command 'cmd'.
+ // The result will be stored in 'location', which must not be null. If there
+ // is an error running the command or the output is invalid, an error Status
+ // will be returned.
+ //
+ // TODO(wdberkeley): Eventually we may want to get multiple locations at once
+ // by giving the location mapping command multiple arguments (like Hadoop).
+ static Status GetLocationFromLocationMappingCmd(const std::string& cmd,
+ const std::string& key,
+ std::string* location);
+
+ // The executable to run when assigning locations for keys which are not yet
+ // in the cache.
+ const std::string location_mapping_cmd_;
+
+ // Counter to track cache hits, i.e. when it was not necessary to run
+ // the location assignment command.
+ scoped_refptr<Counter> location_mapping_cache_hits_;
+
+ // Counter to track overall cache queries, i.e. hits plus misses. Every miss
+ // results in the location assignment command being run.
+ scoped_refptr<Counter> location_mapping_cache_queries_;
+
+ // Spinlock to protect the location assignment map (location_map_).
+ rw_spinlock location_map_lock_;
+
+ // The location assignment map: dictionary of key --> location.
+ std::unordered_map<std::string, std::string> location_map_;
+
+ DISALLOW_COPY_AND_ASSIGN(LocationCache);
+};
+
+} // namespace master
+} // namespace kudu
[kudu] 03/03: [master] use cache for assigned locations
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 4ace91713ad81e72135cca7679e8a6e63b4382b5
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue Feb 26 18:14:30 2019 -0800
[master] use cache for assigned locations
This changelist contains changes to incorporate LocationCache
into master. With this patch, master caches once assigned location
to tablet servers and clients.
Corresponding tests have been updated accordingly and a few additional
tests are added with this changelist.
Change-Id: I12c8952c43a8ad352acd46c8006824b2ad9d1204
Reviewed-on: http://gerrit.cloudera.org:8080/12619
Tested-by: Kudu Jenkins
Reviewed-by: Will Berkeley <wd...@gmail.com>
---
src/kudu/client/client-test.cc | 44 ++++++++++++-
.../integration-tests/location_assignment-itest.cc | 70 +++++++++++++++++++++
src/kudu/master/master-test.cc | 25 ++++++--
src/kudu/master/master.cc | 9 ++-
src/kudu/master/master.h | 11 +++-
src/kudu/master/master_service.cc | 10 +--
src/kudu/master/ts_descriptor-test.cc | 24 ++++----
src/kudu/master/ts_descriptor.cc | 72 +++-------------------
src/kudu/master/ts_descriptor.h | 16 ++---
src/kudu/master/ts_manager.cc | 9 ++-
src/kudu/master/ts_manager.h | 10 ++-
src/kudu/tools/ksck_remote-test.cc | 37 ++++++-----
12 files changed, 210 insertions(+), 127 deletions(-)
diff --git a/src/kudu/client/client-test.cc b/src/kudu/client/client-test.cc
index db42248..7404f4a 100644
--- a/src/kudu/client/client-test.cc
+++ b/src/kudu/client/client-test.cc
@@ -115,6 +115,7 @@
DECLARE_bool(allow_unsafe_replication_factor);
DECLARE_bool(fail_dns_resolution);
+DECLARE_bool(location_mapping_by_uuid);
DECLARE_bool(log_inject_latency);
DECLARE_bool(master_support_connect_to_master_rpc);
DECLARE_bool(rpc_trace_negotiation);
@@ -141,6 +142,8 @@ DEFINE_int32(test_scan_num_rows, 1000, "Number of rows to insert and scan");
METRIC_DECLARE_counter(block_manager_total_bytes_read);
METRIC_DECLARE_counter(rpcs_queue_overflow);
+METRIC_DECLARE_counter(location_mapping_cache_hits);
+METRIC_DECLARE_counter(location_mapping_cache_queries);
METRIC_DECLARE_histogram(handler_latency_kudu_master_MasterService_GetMasterRegistration);
METRIC_DECLARE_histogram(handler_latency_kudu_master_MasterService_GetTabletLocations);
METRIC_DECLARE_histogram(handler_latency_kudu_master_MasterService_GetTableLocations);
@@ -5962,6 +5965,10 @@ TEST_F(ClientTest, WritingRowsWithUnsetNonNullableColumns) {
}
}
+TEST_F(ClientTest, TestClientLocationNoLocationMappingCmd) {
+ ASSERT_TRUE(client_->location().empty());
+}
+
// Client test that assigns locations to clients and tablet servers.
// For now, assigns a uniform location to all clients and tablet servers.
class ClientWithLocationTest : public ClientTest {
@@ -5972,15 +5979,46 @@ class ClientWithLocationTest : public ClientTest {
const string location = "/foo";
FLAGS_location_mapping_cmd = strings::Substitute("$0 $1",
location_cmd_path, location);
+ FLAGS_location_mapping_by_uuid = true;
}
};
-TEST_F(ClientTest, TestClientLocationNoLocationMappingCmd) {
- ASSERT_TRUE(client_->location().empty());
+TEST_F(ClientWithLocationTest, TestClientLocation) {
+ ASSERT_EQ("/foo", client_->location());
}
-TEST_F(ClientWithLocationTest, TestClientLocation) {
+TEST_F(ClientWithLocationTest, LocationCacheMetricsOnClientConnectToCluster) {
ASSERT_EQ("/foo", client_->location());
+
+ auto& metric_entity = cluster_->mini_master()->master()->metric_entity();
+ scoped_refptr<Counter> counter_hits(
+ METRIC_location_mapping_cache_hits.Instantiate(metric_entity));
+ const auto hits_before = counter_hits->value();
+ ASSERT_EQ(0, hits_before);
+ scoped_refptr<Counter> counter_queries(
+ METRIC_location_mapping_cache_queries.Instantiate(metric_entity));
+ const auto queries_before = counter_queries->value();
+ // Expecting location assignment queries from all tablet servers and
+ // the client.
+ ASSERT_EQ(cluster_->num_tablet_servers() + 1, queries_before);
+
+ static constexpr int kIterNum = 10;
+ for (auto iter = 0; iter < kIterNum; ++iter) {
+ shared_ptr<KuduClient> client;
+ ASSERT_OK(KuduClientBuilder()
+ .add_master_server_addr(cluster_->mini_master()->bound_rpc_addr().ToString())
+ .Build(&client));
+ ASSERT_EQ("/foo", client->location());
+ }
+
+ // The location mapping cache should be hit every time a client is connecting
+ // from the same host as the former client. Nothing else should be touching
+ // the location assignment logic but ConnectToCluster() requests coming from
+ // the clients instantiated above.
+ const auto queries_after = counter_queries->value();
+ ASSERT_EQ(queries_before + kIterNum, queries_after);
+ const auto hits_after = counter_hits->value();
+ ASSERT_EQ(hits_before + kIterNum, hits_after);
}
} // namespace client
diff --git a/src/kudu/integration-tests/location_assignment-itest.cc b/src/kudu/integration-tests/location_assignment-itest.cc
index cae1fa8..ce1d891 100644
--- a/src/kudu/integration-tests/location_assignment-itest.cc
+++ b/src/kudu/integration-tests/location_assignment-itest.cc
@@ -49,6 +49,8 @@
DECLARE_int32(num_replicas);
DECLARE_int32(num_tablet_servers);
+METRIC_DECLARE_counter(location_mapping_cache_hits);
+METRIC_DECLARE_counter(location_mapping_cache_queries);
METRIC_DECLARE_counter(scans_started);
METRIC_DECLARE_entity(tablet);
@@ -225,6 +227,74 @@ TEST_P(TsLocationAssignmentITest, Basic) {
NO_FATALS(cluster_->AssertNoCrashes());
}
+// Verify the behavior of the location mapping cache upon tablet server
+// registrations.
+TEST_P(TsLocationAssignmentITest, LocationMappingCacheOnTabletServerRestart) {
+ if (!AllowSlowTests()) {
+ LOG(WARNING) << "test is skipped; set KUDU_ALLOW_SLOW_TESTS=1 to run";
+ return;
+ }
+
+ NO_FATALS(StartCluster());
+ NO_FATALS(CheckLocationInfo());
+ NO_FATALS(cluster_->AssertNoCrashes());
+
+ const auto num_tablet_servers = cluster_->num_tablet_servers();
+
+ int64_t hits_before;
+ ASSERT_OK(itest::GetInt64Metric(
+ cluster_->leader_master()->bound_http_hostport(),
+ &METRIC_ENTITY_server,
+ nullptr,
+ &METRIC_location_mapping_cache_hits,
+ "value",
+ &hits_before));
+ ASSERT_EQ(0, hits_before);
+
+ int64_t queries_before;
+ ASSERT_OK(itest::GetInt64Metric(
+ cluster_->leader_master()->bound_http_hostport(),
+ &METRIC_ENTITY_server,
+ nullptr,
+ &METRIC_location_mapping_cache_queries,
+ "value",
+ &queries_before));
+ ASSERT_EQ(num_tablet_servers, queries_before);
+
+ for (auto idx = 0; idx < num_tablet_servers; ++idx) {
+ auto* ts = cluster_->tablet_server(idx);
+ ts->Shutdown();
+ ASSERT_OK(ts->Restart());
+ }
+
+ NO_FATALS(CheckLocationInfo());
+ NO_FATALS(cluster_->AssertNoCrashes());
+
+ ASSERT_EVENTUALLY([&]() {
+ int64_t hits_after;
+ ASSERT_OK(itest::GetInt64Metric(
+ cluster_->leader_master()->bound_http_hostport(),
+ &METRIC_ENTITY_server,
+ nullptr,
+ &METRIC_location_mapping_cache_hits,
+ "value",
+ &hits_after));
+ ASSERT_EQ(hits_before + num_tablet_servers, hits_after);
+ });
+
+ ASSERT_EVENTUALLY([&]() {
+ int64_t queries_after;
+ ASSERT_OK(itest::GetInt64Metric(
+ cluster_->leader_master()->bound_http_hostport(),
+ &METRIC_ENTITY_server,
+ nullptr,
+ &METRIC_location_mapping_cache_queries,
+ "value",
+ &queries_after));
+ ASSERT_EQ(queries_before + num_tablet_servers, queries_after);
+ });
+}
+
INSTANTIATE_TEST_CASE_P(, TsLocationAssignmentITest,
::testing::Combine(::testing::Values(1, 3),
::testing::Values(1, 8, 16, 32)));
diff --git a/src/kudu/master/master-test.cc b/src/kudu/master/master-test.cc
index 552d1ed..72f2b62 100644
--- a/src/kudu/master/master-test.cc
+++ b/src/kudu/master/master-test.cc
@@ -473,6 +473,11 @@ TEST_F(MasterTest, TestRegisterAndHeartbeat) {
// Set a command that always fails.
FLAGS_location_mapping_cmd = "false";
+ // Restarting the master to take into account the new setting for the
+ // --location_mapping_cmd flag.
+ mini_master_->Shutdown();
+ ASSERT_OK(mini_master_->Restart());
+
// Set a new UUID so registration is for the first time.
auto new_common = common;
new_common.mutable_ts_instance()->set_permanent_uuid("lmc-fail-ts");
@@ -486,7 +491,7 @@ TEST_F(MasterTest, TestRegisterAndHeartbeat) {
// Registration should fail.
Status s = proxy_->TSHeartbeat(hb_req, &hb_resp, &rpc);
- ASSERT_TRUE(s.IsRemoteError());
+ ASSERT_TRUE(s.IsRemoteError()) << s.ToString();
ASSERT_STR_CONTAINS(s.ToString(), "failed to run location mapping command");
// Make sure the tablet server isn't returned to clients.
@@ -497,11 +502,7 @@ TEST_F(MasterTest, TestRegisterAndHeartbeat) {
LOG(INFO) << SecureDebugString(list_ts_resp);
ASSERT_FALSE(list_ts_resp.has_error());
- ASSERT_EQ(1, list_ts_resp.servers_size());
- ASSERT_EQ("my-ts-uuid", list_ts_resp.servers(0).instance_id().permanent_uuid());
-
- // Reset the flag.
- FLAGS_location_mapping_cmd = "";
+ ASSERT_EQ(0, list_ts_resp.servers_size());
}
}
@@ -1656,6 +1657,10 @@ TEST_F(MasterTest, TestConnectToMasterAndAssignLocation) {
const string location = "/foo";
FLAGS_location_mapping_cmd = Substitute("$0 $1", kLocationCmdPath, location);
{
+ // Restarting the master to take into account the new setting for the
+ // --location_mapping_cmd flag.
+ mini_master_->Shutdown();
+ ASSERT_OK(mini_master_->Restart());
ConnectToMasterRequestPB req;
ConnectToMasterResponsePB resp;
RpcController rpc;
@@ -1668,6 +1673,10 @@ TEST_F(MasterTest, TestConnectToMasterAndAssignLocation) {
// location should be assigned.
FLAGS_location_mapping_cmd = "false";
{
+ // Restarting the master to take into account the new setting for the
+ // --location_mapping_cmd flag.
+ mini_master_->Shutdown();
+ ASSERT_OK(mini_master_->Restart());
ConnectToMasterRequestPB req;
ConnectToMasterResponsePB resp;
RpcController rpc;
@@ -1680,6 +1689,10 @@ TEST_F(MasterTest, TestConnectToMasterAndAssignLocation) {
const string new_location = "/bar";
FLAGS_location_mapping_cmd = Substitute("$0 $1", kLocationCmdPath, new_location);
{
+ // Restarting the master to take into account the new setting for the
+ // --location_mapping_cmd flag.
+ mini_master_->Shutdown();
+ ASSERT_OK(mini_master_->Restart());
ConnectToMasterRequestPB req;
ConnectToMasterResponsePB resp;
RpcController rpc;
diff --git a/src/kudu/master/master.cc b/src/kudu/master/master.cc
index bc7f9ae..4f0a7e0 100644
--- a/src/kudu/master/master.cc
+++ b/src/kudu/master/master.cc
@@ -37,8 +37,10 @@
#include "kudu/fs/fs_manager.h"
#include "kudu/gutil/bind.h"
#include "kudu/gutil/bind_helpers.h"
+#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/master/catalog_manager.h"
+#include "kudu/master/location_cache.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/master.proxy.h"
#include "kudu/master/master_cert_authority.h"
@@ -87,6 +89,7 @@ TAG_FLAG(authz_token_validity_seconds, experimental);
DECLARE_bool(hive_metastore_sasl_enabled);
DECLARE_string(keytab_file);
+DECLARE_string(location_mapping_cmd);
using std::min;
using std::shared_ptr;
@@ -124,11 +127,15 @@ GROUP_FLAG_VALIDATOR(hive_metastore_sasl_enabled, ValidateHiveMetastoreSaslEnabl
Master::Master(const MasterOptions& opts)
: KuduServer("Master", opts, "kudu.master"),
state_(kStopped),
- ts_manager_(new TSManager(metric_entity_)),
catalog_manager_(new CatalogManager(this)),
path_handlers_(new MasterPathHandlers(this)),
opts_(opts),
registration_initialized_(false) {
+ const auto& location_cmd = FLAGS_location_mapping_cmd;
+ if (!location_cmd.empty()) {
+ location_cache_.reset(new LocationCache(location_cmd, metric_entity_.get()));
+ }
+ ts_manager_.reset(new TSManager(location_cache_.get(), metric_entity_));
}
Master::~Master() {
diff --git a/src/kudu/master/master.h b/src/kudu/master/master.h
index fda75a1..077aabd 100644
--- a/src/kudu/master/master.h
+++ b/src/kudu/master/master.h
@@ -38,6 +38,9 @@ class HostPortPB;
class MaintenanceManager;
class MonoDelta;
class ThreadPool;
+namespace master {
+class LocationCache;
+} // namespace master
namespace security {
class TokenSigner;
@@ -83,6 +86,8 @@ class Master : public kserver::KuduServer {
const MasterOptions& opts() { return opts_; }
+ LocationCache* location_cache() { return location_cache_.get(); }
+
// Get the RPC and HTTP addresses for this master instance.
Status GetMasterRegistration(ServerRegistrationPB* registration) const;
@@ -131,7 +136,6 @@ class Master : public kserver::KuduServer {
std::unique_ptr<MasterCertAuthority> cert_authority_;
std::unique_ptr<security::TokenSigner> token_signer_;
- gscoped_ptr<TSManager> ts_manager_;
gscoped_ptr<CatalogManager> catalog_manager_;
gscoped_ptr<MasterPathHandlers> path_handlers_;
@@ -151,6 +155,11 @@ class Master : public kserver::KuduServer {
// The maintenance manager for this master.
std::shared_ptr<MaintenanceManager> maintenance_manager_;
+ // A simplistic cache to track already assigned locations.
+ std::unique_ptr<LocationCache> location_cache_;
+
+ gscoped_ptr<TSManager> ts_manager_;
+
DISALLOW_COPY_AND_ASSIGN(Master);
};
diff --git a/src/kudu/master/master_service.cc b/src/kudu/master/master_service.cc
index 7daefae..b8f4f4c 100644
--- a/src/kudu/master/master_service.cc
+++ b/src/kudu/master/master_service.cc
@@ -36,6 +36,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/hms/hms_catalog.h"
#include "kudu/master/catalog_manager.h"
+#include "kudu/master/location_cache.h"
#include "kudu/master/master.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/master_cert_authority.h"
@@ -58,7 +59,6 @@
DECLARE_bool(hive_metastore_sasl_enabled);
DECLARE_bool(raft_prepare_replacement_before_eviction);
DECLARE_string(hive_metastore_uris);
-DECLARE_string(location_mapping_cmd);
DEFINE_int32(master_inject_latency_on_tablet_lookups_ms, 0,
"Number of milliseconds that the master will sleep before responding to "
@@ -568,12 +568,12 @@ void MasterServiceImpl::ConnectToMaster(const ConnectToMasterRequestPB* /*req*/,
}
// Assign a location to the client if needed.
- if (!FLAGS_location_mapping_cmd.empty() &&
+ auto* location_cache = server_->location_cache();
+ if (location_cache != nullptr &&
PREDICT_TRUE(FLAGS_master_client_location_assignment_enabled)) {
string location;
- Status s = GetLocationFromLocationMappingCmd(FLAGS_location_mapping_cmd,
- rpc->remote_address().host(),
- &location);
+ const auto s = location_cache->GetLocation(
+ rpc->remote_address().host(), &location);
if (s.ok()) {
resp->set_client_location(location);
} else {
diff --git a/src/kudu/master/ts_descriptor-test.cc b/src/kudu/master/ts_descriptor-test.cc
index 92e3fec..85baea5 100644
--- a/src/kudu/master/ts_descriptor-test.cc
+++ b/src/kudu/master/ts_descriptor-test.cc
@@ -22,20 +22,18 @@
#include <vector>
#include <boost/optional/optional.hpp>
-#include <gflags/gflags_declare.h>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include "kudu/common/common.pb.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/master/location_cache.h"
#include "kudu/util/path_util.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
-DECLARE_string(location_mapping_cmd);
-
using std::shared_ptr;
using std::string;
using std::vector;
@@ -76,7 +74,7 @@ TEST(TSDescriptorTest, TestRegistration) {
ServerRegistrationPB registration;
SetupBasicRegistrationInfo(uuid, &instance, ®istration);
shared_ptr<TSDescriptor> desc;
- ASSERT_OK(TSDescriptor::RegisterNew(instance, registration, &desc));
+ ASSERT_OK(TSDescriptor::RegisterNew(instance, registration, nullptr, &desc));
// Spot check some fields and the ToString value.
ASSERT_EQ(uuid, desc->permanent_uuid());
@@ -91,14 +89,15 @@ TEST(TSDescriptorTest, TestLocationCmd) {
"testdata/first_argument.sh");
// A happy case, using all allowed special characters.
const string location = "/foo-bar0/BAAZ._9-quux";
- FLAGS_location_mapping_cmd = Substitute("$0 $1", kLocationCmdPath, location);
+ const string location_cmd = Substitute("$0 $1", kLocationCmdPath, location);
+ LocationCache cache(location_cmd, nullptr);
const string uuid = "test";
NodeInstancePB instance;
ServerRegistrationPB registration;
SetupBasicRegistrationInfo(uuid, &instance, ®istration);
shared_ptr<TSDescriptor> desc;
- ASSERT_OK(TSDescriptor::RegisterNew(instance, registration, &desc));
+ ASSERT_OK(TSDescriptor::RegisterNew(instance, registration, &cache, &desc));
ASSERT_EQ(location, desc->location());
@@ -109,9 +108,10 @@ TEST(TSDescriptorTest, TestLocationCmd) {
"foo", // Doesn't begin with /.
"/foo$", // Contains the illegal character '$'.
};
- for (const auto& bad_location : bad_locations) {
- FLAGS_location_mapping_cmd = Substitute("$0 $1", kLocationCmdPath, bad_location);
- ASSERT_TRUE(desc->Register(instance, registration).IsRuntimeError());
+ for (const auto& location : bad_locations) {
+ const auto location_cmd = Substitute("$0 $1", kLocationCmdPath, location);
+ LocationCache cache(location_cmd, nullptr);
+ ASSERT_TRUE(desc->Register(instance, registration, &cache).IsRuntimeError());
}
// Bad cases where the script is invalid.
@@ -127,9 +127,9 @@ TEST(TSDescriptorTest, TestLocationCmd) {
// Command returns too many locations (i.e. contains illegal ' ' character).
Substitute("echo $0 $1", "/foo", "/bar"),
};
- for (const auto& bad_cmd : bad_cmds) {
- FLAGS_location_mapping_cmd = bad_cmd;
- ASSERT_TRUE(desc->Register(instance, registration).IsRuntimeError());
+ for (const auto& cmd : bad_cmds) {
+ LocationCache cache(cmd, nullptr);
+ ASSERT_TRUE(desc->Register(instance, registration, &cache).IsRuntimeError());
}
}
} // namespace master
diff --git a/src/kudu/master/ts_descriptor.cc b/src/kudu/master/ts_descriptor.cc
index 619f927..208adac 100644
--- a/src/kudu/master/ts_descriptor.cc
+++ b/src/kudu/master/ts_descriptor.cc
@@ -18,7 +18,6 @@
#include "kudu/master/ts_descriptor.h"
#include <cmath>
-#include <cstdio>
#include <mutex>
#include <ostream>
#include <unordered_set>
@@ -32,17 +31,15 @@
#include "kudu/common/wire_protocol.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/consensus.proxy.h"
-#include "kudu/gutil/strings/charset.h"
-#include "kudu/gutil/strings/split.h"
-#include "kudu/gutil/strings/strip.h"
+#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/master/location_cache.h"
#include "kudu/tserver/tserver_admin.proxy.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/logging.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/pb_util.h"
-#include "kudu/util/subprocess.h"
#include "kudu/util/trace.h"
DEFINE_int32(tserver_unresponsive_timeout_ms, 60 * 1000,
@@ -75,63 +72,12 @@ using strings::Substitute;
namespace kudu {
namespace master {
-namespace {
-// Returns if 'location' is a valid location string, i.e. it begins with /
-// and consists of /-separated tokens each of which contains only characters
-// from the set [a-zA-Z0-9_-.].
-bool IsValidLocation(const string& location) {
- if (location.empty() || location[0] != '/') {
- return false;
- }
- const strings::CharSet charset("abcdefghijklmnopqrstuvwxyz"
- "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
- "0123456789"
- "_-./");
- for (const auto c : location) {
- if (!charset.Test(c)) {
- return false;
- }
- }
- return true;
-}
-} // anonymous namespace
-
-Status GetLocationFromLocationMappingCmd(const string& cmd,
- const string& host,
- string* location) {
- DCHECK(location);
- vector<string> argv = strings::Split(cmd, " ", strings::SkipEmpty());
- if (argv.empty()) {
- return Status::RuntimeError("invalid empty location mapping command");
- }
- argv.push_back(host);
- string stderr, location_temp;
- Status s = Subprocess::Call(argv, /*stdin=*/"", &location_temp, &stderr);
- if (!s.ok()) {
- return Status::RuntimeError(
- Substitute("failed to run location mapping command: $0", s.ToString()),
- stderr);
- }
- StripWhiteSpace(&location_temp);
- // Special case an empty location for a better error.
- if (location_temp.empty()) {
- return Status::RuntimeError(
- "location mapping command returned invalid empty location");
- }
- if (!IsValidLocation(location_temp)) {
- return Status::RuntimeError(
- "location mapping command returned invalid location",
- location_temp);
- }
- *location = std::move(location_temp);
- return Status::OK();
-}
-
Status TSDescriptor::RegisterNew(const NodeInstancePB& instance,
const ServerRegistrationPB& registration,
+ LocationCache* location_cache,
shared_ptr<TSDescriptor>* desc) {
shared_ptr<TSDescriptor> ret(TSDescriptor::make_shared(instance.permanent_uuid()));
- RETURN_NOT_OK(ret->Register(instance, registration));
+ RETURN_NOT_OK(ret->Register(instance, registration, location_cache));
desc->swap(ret);
return Status::OK();
}
@@ -210,7 +156,8 @@ Status TSDescriptor::RegisterUnlocked(const NodeInstancePB& instance,
}
Status TSDescriptor::Register(const NodeInstancePB& instance,
- const ServerRegistrationPB& registration) {
+ const ServerRegistrationPB& registration,
+ LocationCache* location_cache) {
// Do basic registration work under the lock.
{
std::lock_guard<simple_spinlock> l(lock_);
@@ -219,8 +166,7 @@ Status TSDescriptor::Register(const NodeInstancePB& instance,
// Resolve the location outside the lock. This involves calling the location
// mapping script.
- const string& location_mapping_cmd = FLAGS_location_mapping_cmd;
- if (!location_mapping_cmd.empty()) {
+ if (PREDICT_TRUE(location_cache != nullptr)) {
// In some test scenarios the location is assigned per tablet server UUID.
// That's the case when multiple (or even all) tablet servers have the same
// IP address for their RPC endpoint.
@@ -228,9 +174,7 @@ Status TSDescriptor::Register(const NodeInstancePB& instance,
? permanent_uuid() : registration_->rpc_addresses(0).host();
TRACE(Substitute("tablet server $0: assigning location", permanent_uuid()));
string location;
- Status s = GetLocationFromLocationMappingCmd(location_mapping_cmd,
- cmd_arg,
- &location);
+ const auto s = location_cache->GetLocation(cmd_arg, &location);
TRACE(Substitute(
"tablet server $0: assigned location '$1'", permanent_uuid(), location));
diff --git a/src/kudu/master/ts_descriptor.h b/src/kudu/master/ts_descriptor.h
index 65724fe..76a5fa8 100644
--- a/src/kudu/master/ts_descriptor.h
+++ b/src/kudu/master/ts_descriptor.h
@@ -53,17 +53,7 @@ class TabletServerAdminServiceProxy;
namespace master {
-// Resolves 'host', which is the IP address or hostname of a tablet server or
-// client, into a location using the command 'cmd'. The result will be stored
-// in 'location', which must not be null. If there is an error running the
-// command or the output is invalid, an error Status will be returned.
-// TODO(wdberkeley): Refactor into a separate class and implement a caching
-// policy.
-// TODO(wdberkeley): Eventually we may want to get multiple locations at once
-// by giving the script multiple arguments (like Hadoop).
-Status GetLocationFromLocationMappingCmd(const std::string& cmd,
- const std::string& host,
- std::string* location);
+class LocationCache;
// Master-side view of a single tablet server.
//
@@ -73,6 +63,7 @@ class TSDescriptor : public enable_make_shared<TSDescriptor> {
public:
static Status RegisterNew(const NodeInstancePB& instance,
const ServerRegistrationPB& registration,
+ LocationCache* location_cache,
std::shared_ptr<TSDescriptor>* desc);
virtual ~TSDescriptor() = default;
@@ -89,7 +80,8 @@ class TSDescriptor : public enable_make_shared<TSDescriptor> {
// Register this tablet server.
Status Register(const NodeInstancePB& instance,
- const ServerRegistrationPB& registration);
+ const ServerRegistrationPB& registration,
+ LocationCache* location_cache);
const std::string &permanent_uuid() const { return permanent_uuid_; }
int64_t latest_seqno() const;
diff --git a/src/kudu/master/ts_manager.cc b/src/kudu/master/ts_manager.cc
index 4e4bebf..6bd5939 100644
--- a/src/kudu/master/ts_manager.cc
+++ b/src/kudu/master/ts_manager.cc
@@ -49,7 +49,9 @@ using strings::Substitute;
namespace kudu {
namespace master {
-TSManager::TSManager(const scoped_refptr<MetricEntity>& metric_entity) {
+TSManager::TSManager(LocationCache* location_cache,
+ const scoped_refptr<MetricEntity>& metric_entity)
+ : location_cache_(location_cache) {
METRIC_cluster_replica_skew.InstantiateFunctionGauge(
metric_entity,
Bind(&TSManager::ClusterSkew, Unretained(this)))
@@ -93,14 +95,15 @@ Status TSManager::RegisterTS(const NodeInstancePB& instance,
if (!ContainsKey(servers_by_id_, uuid)) {
shared_ptr<TSDescriptor> new_desc;
- RETURN_NOT_OK(TSDescriptor::RegisterNew(instance, registration, &new_desc));
+ RETURN_NOT_OK(TSDescriptor::RegisterNew(
+ instance, registration, location_cache_, &new_desc));
InsertOrDie(&servers_by_id_, uuid, new_desc);
LOG(INFO) << Substitute("Registered new tserver with Master: $0",
new_desc->ToString());
desc->swap(new_desc);
} else {
shared_ptr<TSDescriptor> found(FindOrDie(servers_by_id_, uuid));
- RETURN_NOT_OK(found->Register(instance, registration));
+ RETURN_NOT_OK(found->Register(instance, registration, location_cache_));
LOG(INFO) << Substitute("Re-registered known tserver with Master: $0",
found->ToString());
desc->swap(found);
diff --git a/src/kudu/master/ts_manager.h b/src/kudu/master/ts_manager.h
index b6dc306..bf52a9c 100644
--- a/src/kudu/master/ts_manager.h
+++ b/src/kudu/master/ts_manager.h
@@ -35,6 +35,8 @@ class ServerRegistrationPB;
namespace master {
+class LocationCache;
+
// Tracks the servers that the master has heard from, along with their
// last heartbeat, etc.
//
@@ -47,7 +49,11 @@ namespace master {
// This class is thread-safe.
class TSManager {
public:
- explicit TSManager(const scoped_refptr<MetricEntity>& metric_entity);
+ // 'location_cache' is a pointer to location mapping cache to use when
+ // registering tablet servers. The location cache should outlive the
+ // TSManager. 'metric_entity' is used to register metrics used by TSManager.
+ TSManager(LocationCache* location_cache,
+ const scoped_refptr<MetricEntity>& metric_entity);
virtual ~TSManager();
// Lookup the tablet server descriptor for the given instance identifier.
@@ -92,6 +98,8 @@ class TSManager {
std::string, std::shared_ptr<TSDescriptor>> TSDescriptorMap;
TSDescriptorMap servers_by_id_;
+ LocationCache* location_cache_;
+
DISALLOW_COPY_AND_ASSIGN(TSManager);
};
diff --git a/src/kudu/tools/ksck_remote-test.cc b/src/kudu/tools/ksck_remote-test.cc
index 7b7e2e3..b02c0e8 100644
--- a/src/kudu/tools/ksck_remote-test.cc
+++ b/src/kudu/tools/ksck_remote-test.cc
@@ -511,13 +511,19 @@ TEST_F(RemoteKsckTest, TestLeaderMasterDown) {
}
TEST_F(RemoteKsckTest, TestClusterWithLocation) {
- // There is no location assigned for the existing three tablet servers.
- // With the flag set, the newly added server will be assiged with location '/foo'.
const string location_cmd_path = JoinPathSegments(GetTestExecutableDirectory(),
"testdata/first_argument.sh");
const string location = "/foo";
FLAGS_location_mapping_cmd = Substitute("$0 $1", location_cmd_path, location);
+ // After setting the --location_mapping_cmd flag it's necessary to restart
+ // the masters to pick up the new value.
+ for (auto idx = 0; idx < mini_cluster_->num_masters(); ++idx) {
+ auto* master = mini_cluster_->mini_master(idx);
+ master->Shutdown();
+ ASSERT_OK(master->Start());
+ }
+
ASSERT_OK(mini_cluster_->AddTabletServer());
ASSERT_EQ(4, mini_cluster_->num_tablet_servers());
@@ -529,8 +535,8 @@ TEST_F(RemoteKsckTest, TestClusterWithLocation) {
// a connection takes much longer. To avoid flakiness of this test scenario,
// few calls below are wrapped into ASSERT_EVENTUALLY().
//
- // TODO(KUDU-2704): remove ASSERT_EVENTUALLY around CheckMasterConsensus
- // when KUDU-2704 is addressed.
+ // TODO(KUDU-2709): remove ASSERT_EVENTUALLY around CheckMasterConsensus
+ // when KUDU-2709 is addressed.
ASSERT_EVENTUALLY([&]() {
ASSERT_OK(ksck_->CheckMasterConsensus());
});
@@ -547,28 +553,21 @@ TEST_F(RemoteKsckTest, TestClusterWithLocation) {
ASSERT_OK(ksck_->PrintResults());
const string& err_string = err_stream_.str();
- // The existing tablet servers should have location '<none>' displayed.
- for (int i = 0; i < 3; i++) {
- auto *ts = mini_cluster_->mini_tablet_server(i);
- ASSERT_STR_CONTAINS(err_string, Substitute("$0 | $1 | HEALTHY | <none>",
+ // With the flag set and masters restarted, all tablet servers are assigned
+ // with location '/foo': both the existing ones and the newly added.
+ for (int idx = 0; idx < mini_cluster_->num_tablet_servers(); ++idx) {
+ auto *ts = mini_cluster_->mini_tablet_server(idx);
+ ASSERT_STR_CONTAINS(err_string, Substitute("$0 | $1 | HEALTHY | $2",
ts->uuid(),
- ts->bound_rpc_addr().ToString()));
+ ts->bound_rpc_addr().ToString(),
+ location));
}
-
- // The newly added tablet server should have the assigned location displayed.
- auto *ts = mini_cluster_->mini_tablet_server(3);
- ASSERT_STR_CONTAINS(err_string, Substitute("$0 | $1 | HEALTHY | $2",
- ts->uuid(),
- ts->bound_rpc_addr().ToString(),
- location));
ASSERT_STR_CONTAINS(err_string,
"Tablet Server Location Summary\n"
" Location | Count\n"
"----------+---------\n");
ASSERT_STR_CONTAINS(err_string,
- " <none> | 3\n");
- ASSERT_STR_CONTAINS(err_string,
- " /foo | 1\n");
+ " /foo | 4\n");
}
} // namespace tools
[kudu] 01/03: Rename DeadlineTracker to TimeoutTracker
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 8fbf1cc2bbae3b9967e3c89bd20a254364d25edc
Author: Will Berkeley <wd...@gmail.org>
AuthorDate: Tue Feb 5 13:26:32 2019 -0800
Rename DeadlineTracker to TimeoutTracker
A deadline is the latest time by which something should be completed.
It's an instant, like "next Tuesday at noon". A timeout is an interval
of time allowed for some event to occur or be completed. It's a delta,
like "15 minutes". The DeadlineTracker tracked a "relative deadline",
relative to when the instance's "deadline" was set. That's actually a
timeout. This patch harmonizes the names to the concepts.
Change-Id: I3f465c925856390ecf4747e84bdd5a67c51c69eb
Reviewed-on: http://gerrit.cloudera.org:8080/12373
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Will Berkeley <wd...@gmail.com>
---
.../org/apache/kudu/client/AlterTableRequest.java | 2 +-
.../org/apache/kudu/client/AsyncKuduClient.java | 38 ++---
.../org/apache/kudu/client/AsyncKuduScanner.java | 2 +-
.../main/java/org/apache/kudu/client/Batch.java | 6 +-
.../java/org/apache/kudu/client/BatchResponse.java | 2 +-
.../org/apache/kudu/client/ConnectToCluster.java | 2 +-
.../org/apache/kudu/client/CreateTableRequest.java | 2 +-
.../org/apache/kudu/client/DeadlineTracker.java | 159 ---------------------
.../org/apache/kudu/client/DeleteTableRequest.java | 2 +-
.../apache/kudu/client/GetTableSchemaRequest.java | 2 +-
.../kudu/client/IsAlterTableDoneRequest.java | 2 +-
.../kudu/client/IsCreateTableDoneRequest.java | 2 +-
.../org/apache/kudu/client/KuduPartitioner.java | 6 +-
.../main/java/org/apache/kudu/client/KuduRpc.java | 10 +-
.../org/apache/kudu/client/ListTablesRequest.java | 2 +-
.../kudu/client/ListTabletServersRequest.java | 2 +-
.../org/apache/kudu/client/ListTabletsRequest.java | 2 +-
.../java/org/apache/kudu/client/Operation.java | 6 +-
.../java/org/apache/kudu/client/PingRequest.java | 2 +-
.../main/java/org/apache/kudu/client/RpcProxy.java | 4 +-
.../org/apache/kudu/client/TimeoutTracker.java | 159 +++++++++++++++++++++
.../test/java/org/apache/kudu/client/ITClient.java | 8 +-
.../apache/kudu/client/TestConnectionCache.java | 6 +-
...eadlineTracker.java => TestTimeoutTracker.java} | 22 +--
.../java/org/apache/kudu/test/KuduTestHarness.java | 11 +-
.../org/apache/kudu/test/TestMiniKuduCluster.java | 4 +-
26 files changed, 232 insertions(+), 233 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
index 9e2dfd0..5734c67 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableRequest.java
@@ -80,7 +80,7 @@ class AlterTableRequest extends KuduRpc<AlterTableResponse> {
final AlterTableResponsePB.Builder respBuilder = AlterTableResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
AlterTableResponse response = new AlterTableResponse(
- deadlineTracker.getElapsedMillis(),
+ timeoutTracker.getElapsedMillis(),
tsUUID,
respBuilder.hasTableId() ? respBuilder.getTableId().toStringUtf8() : null);
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 20949a3..cf22d0a 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
@@ -978,7 +978,7 @@ public class AsyncKuduClient implements AutoCloseable {
RecoverableException ex = (RecoverableException)arg;
long sleepTime = getSleepTimeForRpcMillis(fakeRpc);
if (cannotRetryRequest(fakeRpc) ||
- fakeRpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTime)) {
+ fakeRpc.timeoutTracker.wouldSleepingTimeoutMillis(sleepTime)) {
tooManyAttemptsOrTimeout(fakeRpc, ex); // invokes fakeRpc.Deferred
return null;
}
@@ -1579,7 +1579,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
long sleepTimeMillis = getSleepTimeForRpcMillis(rpc);
- if (rpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTimeMillis)) {
+ if (rpc.timeoutTracker.wouldSleepingTimeoutMillis(sleepTimeMillis)) {
tooManyAttemptsOrTimeout(rpc, null);
return;
}
@@ -1609,7 +1609,7 @@ public class AsyncKuduClient implements AutoCloseable {
}
long sleepTimeMillis = getSleepTimeForRpcMillis(rpc);
- if (rpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTimeMillis)) {
+ if (rpc.timeoutTracker.wouldSleepingTimeoutMillis(sleepTimeMillis)) {
tooManyAttemptsOrTimeout(rpc, null);
return;
}
@@ -1664,7 +1664,7 @@ public class AsyncKuduClient implements AutoCloseable {
* {@code false} otherwise (in which case it's OK to retry once more)
*/
private static boolean cannotRetryRequest(final KuduRpc<?> rpc) {
- return rpc.deadlineTracker.timedOut() || rpc.attempt > MAX_RPC_ATTEMPTS;
+ return rpc.timeoutTracker.timedOut() || rpc.attempt > MAX_RPC_ATTEMPTS;
}
/**
@@ -1722,7 +1722,7 @@ public class AsyncKuduClient implements AutoCloseable {
d = getMasterTableLocationsPB(parentRpc);
} else {
long timeoutMillis = parentRpc == null ? defaultAdminOperationTimeoutMs :
- parentRpc.deadlineTracker.getMillisBeforeDeadline();
+ parentRpc.timeoutTracker.getMillisBeforeTimeout();
// Leave the end of the partition key range empty in order to pre-fetch tablet locations.
GetTableLocationsRequest rpc =
new GetTableLocationsRequest(masterTable,
@@ -1818,7 +1818,7 @@ public class AsyncKuduClient implements AutoCloseable {
final byte[] endPartitionKey,
final int fetchBatchSize,
final List<LocatedTablet> ret,
- final DeadlineTracker deadlineTracker) {
+ final TimeoutTracker timeoutTracker) {
// We rely on the keys initially not being empty.
Preconditions.checkArgument(startPartitionKey == null || startPartitionKey.length > 0,
"use null for unbounded start partition key");
@@ -1846,9 +1846,9 @@ public class AsyncKuduClient implements AutoCloseable {
continue;
}
- if (deadlineTracker.timedOut()) {
+ if (timeoutTracker.timedOut()) {
Status statusTimedOut = Status.TimedOut("Took too long getting the list of tablets, " +
- deadlineTracker);
+ timeoutTracker);
return Deferred.fromError(new NonRecoverableException(statusTimedOut));
}
@@ -1861,7 +1861,7 @@ public class AsyncKuduClient implements AutoCloseable {
// Build a fake RPC to encapsulate and propagate the timeout. There's no actual "RPC" to send.
KuduRpc fakeRpc = buildFakeRpc("loopLocateTable",
null,
- deadlineTracker.getMillisBeforeDeadline());
+ timeoutTracker.getMillisBeforeTimeout());
return locateTablet(table, key, fetchBatchSize, fakeRpc).addCallbackDeferring(
new Callback<Deferred<List<LocatedTablet>>, GetTableLocationsResponsePB>() {
@@ -1872,7 +1872,7 @@ public class AsyncKuduClient implements AutoCloseable {
endPartitionKey,
fetchBatchSize,
ret,
- deadlineTracker);
+ timeoutTracker);
}
@Override
@@ -1904,14 +1904,14 @@ public class AsyncKuduClient implements AutoCloseable {
int fetchBatchSize,
long deadline) {
final List<LocatedTablet> ret = Lists.newArrayList();
- final DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(deadline);
+ final TimeoutTracker timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(deadline);
return loopLocateTable(table,
startPartitionKey,
endPartitionKey,
fetchBatchSize,
ret,
- deadlineTracker);
+ timeoutTracker);
}
/**
@@ -1998,7 +1998,7 @@ public class AsyncKuduClient implements AutoCloseable {
long sleepTime = getSleepTimeForRpcMillis(rpc);
if (cannotRetryRequest(rpc) ||
- rpc.deadlineTracker.wouldSleepingTimeoutMillis(sleepTime)) {
+ rpc.timeoutTracker.wouldSleepingTimeoutMillis(sleepTime)) {
// Don't let it retry.
return tooManyAttemptsOrTimeout(rpc, ex);
}
@@ -2215,7 +2215,7 @@ public class AsyncKuduClient implements AutoCloseable {
Deferred<LocatedTablet> getTabletLocation(final KuduTable table,
final byte[] partitionKey,
final LookupType lookupType,
- long deadline) {
+ long timeout) {
// Locate the tablet at the partition key by locating tablets between
// the partition key (inclusive), and the incremented partition key (exclusive).
@@ -2230,10 +2230,10 @@ public class AsyncKuduClient implements AutoCloseable {
endPartitionKey = Arrays.copyOf(partitionKey, partitionKey.length + 1);
}
- final DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(deadline);
+ final TimeoutTracker timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(timeout);
Deferred<List<LocatedTablet>> locatedTablets = locateTable(
- table, startPartitionKey, endPartitionKey, FETCH_TABLETS_PER_POINT_LOOKUP, deadline);
+ table, startPartitionKey, endPartitionKey, FETCH_TABLETS_PER_POINT_LOOKUP, timeout);
// Then pick out the single tablet result from the list.
return locatedTablets.addCallbackDeferring(
@@ -2265,7 +2265,7 @@ public class AsyncKuduClient implements AutoCloseable {
// This is a LOWER_BOUND lookup, get the tablet location from the upper bound key
// of the non-covered range to return the next valid tablet location.
return getTabletLocation(table, entry.getUpperBoundPartitionKey(),
- LookupType.POINT, deadlineTracker.getMillisBeforeDeadline());
+ LookupType.POINT, timeoutTracker.getMillisBeforeTimeout());
}
return Deferred.fromResult(new LocatedTablet(entry.getTablet()));
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
index 15668d1..93bafb8 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduScanner.java
@@ -1021,7 +1021,7 @@ public final class AsyncKuduScanner {
}
}
RowResultIterator iterator = RowResultIterator.makeRowResultIterator(
- deadlineTracker.getElapsedMillis(), tsUUID, schema, resp.getData(), callResponse);
+ timeoutTracker.getElapsedMillis(), tsUUID, schema, resp.getData(), callResponse);
boolean hasMore = resp.getHasMoreResults();
if (id.length != 0 && scannerId != null && !Bytes.equals(scannerId, id)) {
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java b/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
index a89374e..b08e246 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Batch.java
@@ -73,8 +73,8 @@ class Batch extends KuduRpc<BatchResponse> {
* @param timeoutMillis the new timeout of the batch in milliseconds
*/
void resetTimeoutMillis(Timer timer, long timeoutMillis) {
- deadlineTracker.reset();
- deadlineTracker.setDeadline(timeoutMillis);
+ timeoutTracker.reset();
+ timeoutTracker.setTimeout(timeoutMillis);
if (timeoutTask != null) {
timeoutTask.cancel();
}
@@ -145,7 +145,7 @@ class Batch extends KuduRpc<BatchResponse> {
}
}
- BatchResponse response = new BatchResponse(deadlineTracker.getElapsedMillis(),
+ BatchResponse response = new BatchResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
builder.getTimestamp(),
errorsPB,
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/BatchResponse.java b/java/kudu-client/src/main/java/org/apache/kudu/client/BatchResponse.java
index 408c4cd..24078eb 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/BatchResponse.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/BatchResponse.java
@@ -78,7 +78,7 @@ public class BatchResponse extends KuduRpcResponse {
currentErrorIndex++;
}
individualResponses.add(
- new OperationResponse(currentOperation.deadlineTracker.getElapsedMillis(),
+ new OperationResponse(currentOperation.timeoutTracker.getElapsedMillis(),
tsUUID,
writeTimestamp,
currentOperation,
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectToCluster.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectToCluster.java
index 1ee9767..0557b56 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectToCluster.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ConnectToCluster.java
@@ -100,7 +100,7 @@ final class ConnectToCluster {
// TODO: Handle the situation when multiple in-flight RPCs all want to query the masters,
// basically reuse in some way the master permits.
long timeoutMillis = parentRpc == null ? defaultTimeoutMs :
- parentRpc.deadlineTracker.getMillisBeforeDeadline();
+ parentRpc.timeoutTracker.getMillisBeforeTimeout();
final ConnectToMasterRequest rpc = new ConnectToMasterRequest(masterTable, timer, timeoutMillis);
rpc.setParentRpc(parentRpc);
Deferred<ConnectToMasterResponsePB> d = rpc.getDeferred();
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
index f1d5f20..c270f82 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableRequest.java
@@ -78,7 +78,7 @@ class CreateTableRequest extends KuduRpc<CreateTableResponse> {
readProtobuf(callResponse.getPBMessage(), builder);
CreateTableResponse response =
new CreateTableResponse(
- deadlineTracker.getElapsedMillis(),
+ timeoutTracker.getElapsedMillis(),
tsUUID,
builder.getTableId().toStringUtf8());
return new Pair<CreateTableResponse, Object>(
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java b/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
deleted file mode 100644
index ec2d6e9..0000000
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/DeadlineTracker.java
+++ /dev/null
@@ -1,159 +0,0 @@
-// 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.util.concurrent.TimeUnit;
-
-import com.google.common.base.Stopwatch;
-
-/**
- * This is a wrapper class around {@link com.google.common.base.Stopwatch} used to track a relative
- * deadline in the future.
- * <p>
- * The watch starts as soon as this object is created with a deadline of 0,
- * meaning that there's no deadline.
- * The deadline has been reached once the stopwatch's elapsed time is equal or greater than the
- * provided deadline.
- */
-public class DeadlineTracker {
- private final Stopwatch stopwatch;
- /** relative deadline in milliseconds **/
- private long deadline = 0;
-
- /**
- * Creates a new tracker, which starts the stopwatch right now.
- */
- public DeadlineTracker() {
- this(Stopwatch.createUnstarted());
- }
-
- /**
- * Creates a new tracker, using the specified stopwatch, and starts it right now.
- * The stopwatch is reset if it was already running.
- * @param stopwatch Specific Stopwatch to use
- */
- public DeadlineTracker(Stopwatch stopwatch) {
- if (stopwatch.isRunning()) {
- stopwatch.reset();
- }
- this.stopwatch = stopwatch.start();
- }
-
- /**
- * Check if we're already past the deadline.
- * @return true if we're past the deadline, otherwise false. Also returns false if no deadline
- * was specified
- */
- public boolean timedOut() {
- if (!hasDeadline()) {
- return false;
- }
- return deadline - stopwatch.elapsed(TimeUnit.MILLISECONDS) <= 0;
- }
-
- /**
- * Get the number of milliseconds before the deadline is reached.
- * <p>
- * This method is used to pass down the remaining deadline to the RPCs, so has special semantics.
- * A deadline of 0 is used to indicate an infinite deadline, and negative deadlines are invalid.
- * Thus, if the deadline has passed (i.e. <tt>deadline - stopwatch.elapsedMillis() <= 0</tt>),
- * the returned value is floored at <tt>1</tt>.
- * <p>
- * Callers who care about this behavior should first check {@link #timedOut()}.
- *
- * @return the remaining millis before the deadline is reached, or 1 if the remaining time is
- * lesser or equal to 0, or Long.MAX_VALUE if no deadline was specified (in which case it
- * should never be called).
- * @throws IllegalStateException if this method is called and no deadline was set
- */
- public long getMillisBeforeDeadline() {
- if (!hasDeadline()) {
- throw new IllegalStateException("This tracker doesn't have a deadline set so it cannot " +
- "answer getMillisBeforeDeadline()");
- }
- long millisBeforeDeadline = deadline - stopwatch.elapsed(TimeUnit.MILLISECONDS);
- millisBeforeDeadline = millisBeforeDeadline <= 0 ? 1 : millisBeforeDeadline;
- return millisBeforeDeadline;
- }
-
- public long getElapsedMillis() {
- return this.stopwatch.elapsed(TimeUnit.MILLISECONDS);
- }
-
- /**
- * Tells if a non-zero deadline was set.
- * @return true if the deadline is greater than 0, false otherwise.
- */
- public boolean hasDeadline() {
- return deadline != 0;
- }
-
- /**
- * Utility method to check if sleeping for a specified amount of time would put us past the
- * deadline.
- * @param plannedSleepTimeMillis number of milliseconds for a planned sleep
- * @return if the planned sleeps goes past the deadline.
- */
- public boolean wouldSleepingTimeoutMillis(long plannedSleepTimeMillis) {
- if (!hasDeadline()) {
- return false;
- }
- return getMillisBeforeDeadline() - plannedSleepTimeMillis <= 0;
- }
-
- /**
- * Sets the deadline to 0 (no deadline) and restarts the stopwatch from scratch.
- */
- public void reset() {
- deadline = 0;
- stopwatch.reset();
- stopwatch.start();
- }
-
- /**
- * Get the deadline (in milliseconds).
- * @return the current deadline
- */
- public long getDeadline() {
- return deadline;
- }
-
- /**
- * Set a new deadline for this tracker. It cannot be smaller than 0,
- * and if it is 0 then it means that there is no deadline (which is the default behavior).
- * This method won't call reset().
- * @param deadline a number of milliseconds greater or equal to 0
- * @throws IllegalArgumentException if the deadline is lesser than 0
- */
- public void setDeadline(long deadline) {
- if (deadline < 0) {
- throw new IllegalArgumentException("The deadline must be greater or equal to 0, " +
- "the passed value is " + deadline);
- }
- this.deadline = deadline;
- }
-
- @Override
- public String toString() {
- StringBuilder buf = new StringBuilder("DeadlineTracker(timeout=");
- buf.append(deadline);
- buf.append(", elapsed=").append(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- buf.append(")");
- return buf.toString();
- }
-}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
index 80c207d..dd8f1fc 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/DeleteTableRequest.java
@@ -67,7 +67,7 @@ class DeleteTableRequest extends KuduRpc<DeleteTableResponse> {
final Master.DeleteTableResponsePB.Builder builder = Master.DeleteTableResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
DeleteTableResponse response =
- new DeleteTableResponse(deadlineTracker.getElapsedMillis(), tsUUID);
+ new DeleteTableResponse(timeoutTracker.getElapsedMillis(), tsUUID);
return new Pair<DeleteTableResponse, Object>(
response, builder.hasError() ? builder.getError() : null);
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
index 93671bf..d3e48e6 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java
@@ -84,7 +84,7 @@ public class GetTableSchemaRequest extends KuduRpc<GetTableSchemaResponse> {
readProtobuf(callResponse.getPBMessage(), respBuilder);
Schema schema = ProtobufHelper.pbToSchema(respBuilder.getSchema());
GetTableSchemaResponse response = new GetTableSchemaResponse(
- deadlineTracker.getElapsedMillis(),
+ timeoutTracker.getElapsedMillis(),
tsUUID,
schema,
respBuilder.getTableId().toStringUtf8(),
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
index 2866faf..2bce8a1 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/IsAlterTableDoneRequest.java
@@ -65,7 +65,7 @@ class IsAlterTableDoneRequest extends KuduRpc<IsAlterTableDoneResponse> {
String tsUUID) throws KuduException {
final IsAlterTableDoneResponsePB.Builder respBuilder = IsAlterTableDoneResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
- IsAlterTableDoneResponse resp = new IsAlterTableDoneResponse(deadlineTracker.getElapsedMillis(),
+ IsAlterTableDoneResponse resp = new IsAlterTableDoneResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
respBuilder.getDone());
return new Pair<IsAlterTableDoneResponse, Object>(
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
index 2fd0290..76ca70a 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/IsCreateTableDoneRequest.java
@@ -58,7 +58,7 @@ class IsCreateTableDoneRequest extends KuduRpc<IsCreateTableDoneResponse> {
IsCreateTableDoneResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), builder);
IsCreateTableDoneResponse resp =
- new IsCreateTableDoneResponse(deadlineTracker.getElapsedMillis(),
+ new IsCreateTableDoneResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
builder.getDone());
return new Pair<IsCreateTableDoneResponse, Object>(
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPartitioner.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPartitioner.java
index bdadea1..9dba46c 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPartitioner.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduPartitioner.java
@@ -143,8 +143,8 @@ public class KuduPartitioner {
* @return a new {@link KuduPartitioner}
*/
public KuduPartitioner build() throws KuduException {
- final DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(timeoutMillis);
+ final TimeoutTracker timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(timeoutMillis);
NavigableMap<BytesKey, Integer> partitionByStartKey = new TreeMap<>();
// Insert a sentinel for the beginning of the table, in case a user
// queries for any row which falls before the first partition.
@@ -157,7 +157,7 @@ public class KuduPartitioner {
tablet = KuduClient.joinAndHandleException(
table.getAsyncClient().getTabletLocation(table,
nextPartKey.bytes, AsyncKuduClient.LookupType.LOWER_BOUND,
- deadlineTracker.getMillisBeforeDeadline()));
+ timeoutTracker.getMillisBeforeTimeout()));
} catch (NonCoveredRangeException ncr) {
// No more tablets
break;
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
index 3d212ee..eb33c80 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/KuduRpc.java
@@ -130,7 +130,7 @@ public abstract class KuduRpc<R> {
final KuduTable table;
- final DeadlineTracker deadlineTracker;
+ final TimeoutTracker timeoutTracker;
// 'timeoutTask' is a handle to the timer task that will time out the RPC. It is
// null if and only if the task has no timeout.
@@ -155,8 +155,8 @@ public abstract class KuduRpc<R> {
KuduRpc(KuduTable table, Timer timer, long timeoutMillis) {
this.table = table;
- this.deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(timeoutMillis);
+ this.timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(timeoutMillis);
if (timer != null) {
this.timeoutTask = AsyncKuduClient.newTimeout(timer,
new RpcTimeoutTask(),
@@ -257,7 +257,7 @@ public abstract class KuduRpc<R> {
if (timeoutTask != null) {
timeoutTask.cancel();
}
- deadlineTracker.reset();
+ timeoutTracker.reset();
traces.clear();
parentRpc = null;
d.callback(result);
@@ -383,7 +383,7 @@ public abstract class KuduRpc<R> {
buf.append(tablet.getTabletId());
}
buf.append(", attempt=").append(attempt);
- buf.append(", ").append(deadlineTracker);
+ buf.append(", ").append(timeoutTracker);
buf.append(", ").append(RpcTraceFrame.getHumanReadableStringForTraces(traces));
// Cheating a bit, we're not actually logging but we'll augment the information provided by
// this method if DEBUG is enabled.
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
index e7416df..a0e43a7 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTablesRequest.java
@@ -71,7 +71,7 @@ class ListTablesRequest extends KuduRpc<ListTablesResponse> {
for (Master.ListTablesResponsePB.TableInfo info : respBuilder.getTablesList()) {
tables.add(info.getName());
}
- ListTablesResponse response = new ListTablesResponse(deadlineTracker.getElapsedMillis(),
+ ListTablesResponse response = new ListTablesResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
tables);
return new Pair<ListTablesResponse, Object>(
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
index 75e62fb..53d8277 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletServersRequest.java
@@ -65,7 +65,7 @@ public class ListTabletServersRequest extends KuduRpc<ListTabletServersResponse>
servers.add(entry.getRegistration().getRpcAddresses(0).getHost());
}
ListTabletServersResponse response =
- new ListTabletServersResponse(deadlineTracker.getElapsedMillis(),
+ new ListTabletServersResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
serversCount,
servers);
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
index a6d4ff3..0645549 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/ListTabletsRequest.java
@@ -61,7 +61,7 @@ class ListTabletsRequest extends KuduRpc<ListTabletsResponse> {
: respBuilder.getStatusAndSchemaList()) {
tablets.add(info.getTabletStatus().getTabletId());
}
- ListTabletsResponse response = new ListTabletsResponse(deadlineTracker.getElapsedMillis(),
+ ListTabletsResponse response = new ListTabletsResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
tablets);
return new Pair<ListTabletsResponse, Object>(
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
index a585c72..38ba205 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/Operation.java
@@ -107,8 +107,8 @@ public abstract class Operation extends KuduRpc<OperationResponse> {
* @param timeoutMillis the new timeout of the batch in milliseconds
*/
void resetTimeoutMillis(Timer timer, long timeoutMillis) {
- deadlineTracker.reset();
- deadlineTracker.setDeadline(timeoutMillis);
+ timeoutTracker.reset();
+ timeoutTracker.setTimeout(timeoutMillis);
if (timeoutTask != null) {
timeoutTask.cancel();
}
@@ -175,7 +175,7 @@ public abstract class Operation extends KuduRpc<OperationResponse> {
error = null;
}
}
- OperationResponse response = new OperationResponse(deadlineTracker.getElapsedMillis(),
+ OperationResponse response = new OperationResponse(timeoutTracker.getElapsedMillis(),
tsUUID,
builder.getTimestamp(),
this,
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java b/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
index d536998..877a571 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/PingRequest.java
@@ -68,7 +68,7 @@ class PingRequest extends KuduRpc<PingResponse> {
final Master.PingResponsePB.Builder respBuilder =
Master.PingResponsePB.newBuilder();
readProtobuf(callResponse.getPBMessage(), respBuilder);
- PingResponse response = new PingResponse(deadlineTracker.getElapsedMillis(), tsUUID);
+ PingResponse response = new PingResponse(timeoutTracker.getElapsedMillis(), tsUUID);
return new Pair<>(response, null);
}
}
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/RpcProxy.java b/java/kudu-client/src/main/java/org/apache/kudu/client/RpcProxy.java
index d95a27c..8419437 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/RpcProxy.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/RpcProxy.java
@@ -183,8 +183,8 @@ class RpcProxy {
.setMethodName(rpc.method()));
final Message reqPB = rpc.createRequestPB();
// TODO(wdberkeley): We should enforce that every RPC has a timeout.
- if (rpc.deadlineTracker.hasDeadline()) {
- headerBuilder.setTimeoutMillis((int) rpc.deadlineTracker.getMillisBeforeDeadline());
+ if (rpc.timeoutTracker.hasTimeout()) {
+ headerBuilder.setTimeoutMillis((int) rpc.timeoutTracker.getMillisBeforeTimeout());
}
if (rpc.isRequestTracked()) {
RpcHeader.RequestIdPB.Builder requestIdBuilder = RpcHeader.RequestIdPB.newBuilder();
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/TimeoutTracker.java b/java/kudu-client/src/main/java/org/apache/kudu/client/TimeoutTracker.java
new file mode 100644
index 0000000..0e99c0f
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/TimeoutTracker.java
@@ -0,0 +1,159 @@
+// 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.util.concurrent.TimeUnit;
+
+import com.google.common.base.Stopwatch;
+
+/**
+ * This is a wrapper class around {@link com.google.common.base.Stopwatch} used to track a timeout
+ * in the future.
+ * <p>
+ * The watch starts as soon as this object is created with a timeout of 0, which means that
+ * there's no timeout.
+ * The timeout has been reached once the stopwatch's elapsed time is equal or greater than the
+ * provided timeout.
+ */
+public class TimeoutTracker {
+ private final Stopwatch stopwatch;
+ /** timeout in milliseconds **/
+ private long timeout = 0;
+
+ /**
+ * Creates a new tracker, which starts the stopwatch right now.
+ */
+ public TimeoutTracker() {
+ this(Stopwatch.createUnstarted());
+ }
+
+ /**
+ * Creates a new tracker, using the specified stopwatch, and starts it right now.
+ * The stopwatch is reset if it was already running.
+ * @param stopwatch Specific Stopwatch to use
+ */
+ public TimeoutTracker(Stopwatch stopwatch) {
+ if (stopwatch.isRunning()) {
+ stopwatch.reset();
+ }
+ this.stopwatch = stopwatch.start();
+ }
+
+ /**
+ * Check if we're already past the timeout.
+ * @return true if we're past the timeout, otherwise false. Also returns false if no timeout
+ * was specified
+ */
+ public boolean timedOut() {
+ if (!hasTimeout()) {
+ return false;
+ }
+ return timeout - stopwatch.elapsed(TimeUnit.MILLISECONDS) <= 0;
+ }
+
+ /**
+ * Get the number of milliseconds before the timeout is reached.
+ * <p>
+ * This method is used to pass down the remaining timeout to the RPCs, so has special semantics.
+ * A timeout of 0 is used to indicate an infinite timeout, and negative timeouts are invalid.
+ * Thus, if the timeout has passed (i.e. <tt>timeout - stopwatch.elapsedMillis() <= 0</tt>),
+ * the returned value is floored at <tt>1</tt>.
+ * <p>
+ * Callers who care about this behavior should first check {@link #timedOut()}.
+ *
+ * @return the remaining millis before the timeout is reached, or 1 if the remaining time is
+ * lesser or equal to 0, or Long.MAX_VALUE if no timeout was specified (in which case it
+ * should never be called).
+ * @throws IllegalStateException if this method is called and no timeout was set
+ */
+ public long getMillisBeforeTimeout() {
+ if (!hasTimeout()) {
+ throw new IllegalStateException("This tracker doesn't have a timeout set so it cannot " +
+ "answer getMillisBeforeTimeout()");
+ }
+ long millisBeforeTimeout = timeout - stopwatch.elapsed(TimeUnit.MILLISECONDS);
+ millisBeforeTimeout = millisBeforeTimeout <= 0 ? 1 : millisBeforeTimeout;
+ return millisBeforeTimeout;
+ }
+
+ public long getElapsedMillis() {
+ return this.stopwatch.elapsed(TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Tells if a non-zero timeout was set.
+ * @return true if the timeout is greater than 0, false otherwise.
+ */
+ public boolean hasTimeout() {
+ return timeout != 0;
+ }
+
+ /**
+ * Utility method to check if sleeping for a specified amount of time would put us past the
+ * timeout.
+ * @param plannedSleepTimeMillis number of milliseconds for a planned sleep
+ * @return if the planned sleeps goes past the timeout.
+ */
+ public boolean wouldSleepingTimeoutMillis(long plannedSleepTimeMillis) {
+ if (!hasTimeout()) {
+ return false;
+ }
+ return getMillisBeforeTimeout() - plannedSleepTimeMillis <= 0;
+ }
+
+ /**
+ * Sets the timeout to 0 (no timeout) and restarts the stopwatch from scratch.
+ */
+ public void reset() {
+ timeout = 0;
+ stopwatch.reset();
+ stopwatch.start();
+ }
+
+ /**
+ * Get the timeout (in milliseconds).
+ * @return the current timeout
+ */
+ public long getTimeout() {
+ return timeout;
+ }
+
+ /**
+ * Set a new timeout for this tracker. It cannot be smaller than 0,
+ * and if it is 0 then it means that there is no timeout (which is the default behavior).
+ * This method won't call reset().
+ * @param timeout a number of milliseconds greater or equal to 0
+ * @throws IllegalArgumentException if the timeout is lesser than 0
+ */
+ public void setTimeout(long timeout) {
+ if (timeout < 0) {
+ throw new IllegalArgumentException("The timeout must be greater or equal to 0, " +
+ "the passed value is " + timeout);
+ }
+ this.timeout = timeout;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder buf = new StringBuilder("TimeoutTracker(timeout=");
+ buf.append(timeout);
+ buf.append(", elapsed=").append(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ buf.append(")");
+ return buf.toString();
+ }
+}
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
index e03ed4f..fa85efc 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
@@ -380,10 +380,10 @@ public class ITClient {
*/
private boolean fullScan() {
int rowCount;
- DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(DEFAULT_SLEEP);
+ TimeoutTracker timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(DEFAULT_SLEEP);
- while (keepRunningLatch.getCount() > 0 && !deadlineTracker.timedOut()) {
+ while (keepRunningLatch.getCount() > 0 && !timeoutTracker.timedOut()) {
KuduScanner scanner = getScannerBuilder().build();
try {
@@ -410,7 +410,7 @@ public class ITClient {
// No need to do anything, we'll exit the loop once we test getCount() in the condition.
}
}
- return !deadlineTracker.timedOut();
+ return !timeoutTracker.timedOut();
}
private KuduScanner.KuduScannerBuilder getScannerBuilder() {
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 64b6ed2..8750d41 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
@@ -106,9 +106,9 @@ public class TestConnectionCache {
}
private void waitForConnectionToTerminate(Connection c) throws InterruptedException {
- DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(5000);
- while (!c.isTerminated() && !deadlineTracker.timedOut()) {
+ TimeoutTracker timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(5000);
+ while (!c.isTerminated() && !timeoutTracker.timedOut()) {
Thread.sleep(250);
}
}
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeoutTracker.java
similarity index 82%
rename from java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
rename to java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeoutTracker.java
index e5d8adf..b43c8d7 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestDeadlineTracker.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeoutTracker.java
@@ -26,7 +26,7 @@ import com.google.common.base.Stopwatch;
import com.google.common.base.Ticker;
import org.junit.Test;
-public class TestDeadlineTracker {
+public class TestTimeoutTracker {
@Test
public void testTimeout() {
@@ -40,37 +40,37 @@ public class TestDeadlineTracker {
Stopwatch stopwatch = Stopwatch.createUnstarted(ticker);
// no timeout set
- DeadlineTracker tracker = new DeadlineTracker(stopwatch);
- tracker.setDeadline(0);
- assertFalse(tracker.hasDeadline());
+ TimeoutTracker tracker = new TimeoutTracker(stopwatch);
+ tracker.setTimeout(0);
+ assertFalse(tracker.hasTimeout());
assertFalse(tracker.timedOut());
// 500ms timeout set
tracker.reset();
- tracker.setDeadline(500);
- assertTrue(tracker.hasDeadline());
+ tracker.setTimeout(500);
+ assertTrue(tracker.hasTimeout());
assertFalse(tracker.timedOut());
assertFalse(tracker.wouldSleepingTimeoutMillis(499));
assertTrue(tracker.wouldSleepingTimeoutMillis(500));
assertTrue(tracker.wouldSleepingTimeoutMillis(501));
- assertEquals(500, tracker.getMillisBeforeDeadline());
+ assertEquals(500, tracker.getMillisBeforeTimeout());
// fast forward 200ms
timeToReturn.set(200 * 1000000);
- assertTrue(tracker.hasDeadline());
+ assertTrue(tracker.hasTimeout());
assertFalse(tracker.timedOut());
assertFalse(tracker.wouldSleepingTimeoutMillis(299));
assertTrue(tracker.wouldSleepingTimeoutMillis(300));
assertTrue(tracker.wouldSleepingTimeoutMillis(301));
- assertEquals(300, tracker.getMillisBeforeDeadline());
+ assertEquals(300, tracker.getMillisBeforeTimeout());
// fast forward another 400ms, so the RPC timed out
timeToReturn.set(600 * 1000000);
- assertTrue(tracker.hasDeadline());
+ assertTrue(tracker.hasTimeout());
assertTrue(tracker.timedOut());
assertTrue(tracker.wouldSleepingTimeoutMillis(299));
assertTrue(tracker.wouldSleepingTimeoutMillis(300));
assertTrue(tracker.wouldSleepingTimeoutMillis(301));
- assertEquals(1, tracker.getMillisBeforeDeadline());
+ assertEquals(1, tracker.getMillisBeforeTimeout());
}
}
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 10a1271..7b2834a 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
@@ -18,13 +18,13 @@ package org.apache.kudu.test;
import org.apache.kudu.client.AsyncKuduClient;
import org.apache.kudu.client.AsyncKuduClient.AsyncKuduClientBuilder;
-import org.apache.kudu.client.DeadlineTracker;
import org.apache.kudu.client.HostAndPort;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduTable;
import org.apache.kudu.client.LocatedTablet;
import org.apache.kudu.client.RemoteTablet;
+import org.apache.kudu.client.TimeoutTracker;
import org.apache.kudu.test.cluster.MiniKuduCluster;
import org.apache.kudu.test.cluster.MiniKuduCluster.MiniKuduClusterBuilder;
import org.apache.kudu.test.cluster.FakeDNS;
@@ -42,7 +42,6 @@ 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;
@@ -231,17 +230,17 @@ public class KuduTestHarness extends ExternalResource {
public HostAndPort findLeaderTabletServer(LocatedTablet tablet)
throws Exception {
LocatedTablet.Replica leader = null;
- DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(DEFAULT_SLEEP);
+ TimeoutTracker timeoutTracker = new TimeoutTracker();
+ timeoutTracker.setTimeout(DEFAULT_SLEEP);
while (leader == null) {
- if (deadlineTracker.timedOut()) {
+ if (timeoutTracker.timedOut()) {
fail("Timed out while trying to find a leader for this table");
}
leader = tablet.getLeaderReplica();
if (leader == null) {
LOG.info("Sleeping while waiting for a tablet LEADER to arise, currently slept {} ms",
- deadlineTracker.getElapsedMillis());
+ timeoutTracker.getElapsedMillis());
Thread.sleep(50);
}
}
diff --git a/java/kudu-test-utils/src/test/java/org/apache/kudu/test/TestMiniKuduCluster.java b/java/kudu-test-utils/src/test/java/org/apache/kudu/test/TestMiniKuduCluster.java
index 1722a57..68b3933 100644
--- a/java/kudu-test-utils/src/test/java/org/apache/kudu/test/TestMiniKuduCluster.java
+++ b/java/kudu-test-utils/src/test/java/org/apache/kudu/test/TestMiniKuduCluster.java
@@ -22,11 +22,11 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.net.Socket;
-import org.apache.kudu.client.DeadlineTracker;
import org.apache.kudu.client.HostAndPort;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduClient.KuduClientBuilder;
import org.apache.kudu.client.ListTablesResponse;
+import org.apache.kudu.client.TimeoutTracker;
import org.apache.kudu.test.cluster.MiniKuduCluster;
import org.apache.kudu.test.junit.RetryRule;
import org.apache.kudu.test.cluster.FakeDNS;
@@ -116,7 +116,7 @@ public class TestMiniKuduCluster {
*/
private static void testHostPort(HostAndPort hp,
boolean testIsOpen) throws InterruptedException {
- DeadlineTracker tracker = new DeadlineTracker();
+ TimeoutTracker tracker = new TimeoutTracker();
while (tracker.getElapsedMillis() < SLEEP_TIME_MS) {
try {
Socket socket = new Socket(hp.getHost(), hp.getPort());