You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/03/01 16:15:36 UTC
[1/3] impala git commit: IMPALA-6432: Default
rpc_negotiation_timeout_ms may cause queries to fail on large clusters
Repository: impala
Updated Branches:
refs/heads/2.x 9e2b77c08 -> 258938ad9
IMPALA-6432: Default rpc_negotiation_timeout_ms may cause queries to fail on large clusters
For large secure clusters, it was empirically found by Michael Ho
that the default negotiation timeout of 300000 ms and using 64
negotiation threads works best.
Future patches may attempt to improve negitation times at which point
we can revisit bringing down the defaults.
Change-Id: I4f3198ceae82ac92f37c981e75134ce948fc4e72
Reviewed-on: http://gerrit.cloudera.org:8080/9434
Reviewed-by: Sailesh Mukil <sa...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/37586a8d
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/37586a8d
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/37586a8d
Branch: refs/heads/2.x
Commit: 37586a8d0b55fc17f63d1247c004527e7b90715c
Parents: 9e2b77c
Author: Sailesh Mukil <sa...@apache.org>
Authored: Fri Feb 23 09:54:42 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Wed Feb 28 22:10:16 2018 +0000
----------------------------------------------------------------------
be/src/rpc/authentication.cc | 1 -
be/src/rpc/rpc-mgr.cc | 4 ++--
2 files changed, 2 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/37586a8d/be/src/rpc/authentication.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/authentication.cc b/be/src/rpc/authentication.cc
index b333054..fc41723 100644
--- a/be/src/rpc/authentication.cc
+++ b/be/src/rpc/authentication.cc
@@ -832,7 +832,6 @@ Status AuthManager::InitKerberosEnv() {
<< FLAGS_krb5_debug_file;
}
}
-
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/37586a8d/be/src/rpc/rpc-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr.cc b/be/src/rpc/rpc-mgr.cc
index ed9614e..4faddb7 100644
--- a/be/src/rpc/rpc-mgr.cc
+++ b/be/src/rpc/rpc-mgr.cc
@@ -72,9 +72,9 @@ DEFINE_int32(num_reactor_threads, 0,
"default value 0, it will be set to number of CPU cores.");
DEFINE_int32(rpc_retry_interval_ms, 5,
"Time in millisecond of waiting before retrying an RPC when remote is busy");
-DEFINE_int32(rpc_negotiation_timeout_ms, 60000,
+DEFINE_int32(rpc_negotiation_timeout_ms, 300000,
"Time in milliseconds of waiting for a negotiation to complete before timing out.");
-DEFINE_int32(rpc_negotiation_thread_count, 4,
+DEFINE_int32(rpc_negotiation_thread_count, 64,
"Maximum number of threads dedicated to handling RPC connection negotiations.");
namespace impala {
[2/3] impala git commit: IMPALA-6583: Wait for non-empty catalog
update at impalad startup
Posted by ta...@apache.org.
IMPALA-6583: Wait for non-empty catalog update at impalad startup
Currently an empty statestore catalog update would enable impalad to
begin compiling queries. It breaks many custom cluster tests. This patch
makes impalad wait for a non-zero catalog version.
Testing: test_single_coordinator_cluster_config is run in a loop for
several hours. Core tests are also run.
Change-Id: Ie5f09d03497543f283b39c6186ecfda2e0097c87
Reviewed-on: http://gerrit.cloudera.org:8080/9458
Reviewed-by: Alex Behm <al...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/85000c28
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/85000c28
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/85000c28
Branch: refs/heads/2.x
Commit: 85000c2866de79258fdc656f40684c190ed6244d
Parents: 37586a8
Author: Tianyi Wang <tw...@cloudera.com>
Authored: Mon Feb 26 15:00:06 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Mar 1 05:34:02 2018 +0000
----------------------------------------------------------------------
.../impala/catalog/CatalogServiceCatalog.java | 4 +++
.../apache/impala/catalog/ImpaladCatalog.java | 27 ++++++++++++--------
.../org/apache/impala/service/Frontend.java | 5 ++--
3 files changed, 23 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/85000c28/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index a3b0cb0..7ea821c 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -1120,6 +1120,10 @@ public class CatalogServiceCatalog extends Catalog {
reader.run();
versionLock_.writeLock().lock();
+ // In case of an empty new catalog, the version should still change to reflect the
+ // reset operation itself and to unblock impalads by making the catalog version >
+ // INITIAL_CATALOG_VERSION. See Frontend.waitForCatalog()
+ ++catalogVersion_;
// Assign new versions to all the loaded data sources.
for (DataSource dataSource: getDataSources()) {
dataSource.setCatalogVersion(incrementAndGetCatalogVersion());
http://git-wip-us.apache.org/repos/asf/impala/blob/85000c28/fe/src/main/java/org/apache/impala/catalog/ImpaladCatalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/ImpaladCatalog.java b/fe/src/main/java/org/apache/impala/catalog/ImpaladCatalog.java
index 1da5d0b..fce4574 100644
--- a/fe/src/main/java/org/apache/impala/catalog/ImpaladCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/ImpaladCatalog.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.fs.Path;
import org.apache.impala.analysis.TableName;
@@ -84,10 +85,8 @@ public class ImpaladCatalog extends Catalog {
// all objects in the catalog have at a minimum, this version. Because updates may
// be applied out of band of a StateStore heartbeat, it is possible the catalog
// contains some objects > than this version.
- private long lastSyncedCatalogVersion_ = Catalog.INITIAL_CATALOG_VERSION;
-
- // Flag to determine if the Catalog is ready to accept user requests. See isReady().
- private final AtomicBoolean isReady_ = new AtomicBoolean(false);
+ private AtomicLong lastSyncedCatalogVersion_ =
+ new AtomicLong(Catalog.INITIAL_CATALOG_VERSION);
// Tracks modifications to this Impalad's catalog from direct updates to the cache.
private final CatalogDeltaLog catalogDeltaLog_ = new CatalogDeltaLog();
@@ -159,7 +158,7 @@ public class ImpaladCatalog extends Catalog {
if (req.isSetCatalog_service_id()) setCatalogServiceId(req.catalog_service_id);
ArrayDeque<TCatalogObject> updatedObjects = new ArrayDeque<>();
ArrayDeque<TCatalogObject> deletedObjects = new ArrayDeque<>();
- long newCatalogVersion = lastSyncedCatalogVersion_;
+ long newCatalogVersion = lastSyncedCatalogVersion_.get();
Pair<Boolean, ByteBuffer> update;
while ((update = FeSupport.NativeGetNextCatalogObjectUpdate(req.native_iterator_ptr))
!= null) {
@@ -207,10 +206,9 @@ public class ImpaladCatalog extends Catalog {
for (TCatalogObject catalogObject: deletedObjects) removeCatalogObject(catalogObject);
- lastSyncedCatalogVersion_ = newCatalogVersion;
+ lastSyncedCatalogVersion_.set(newCatalogVersion);
// Cleanup old entries in the log.
- catalogDeltaLog_.garbageCollect(lastSyncedCatalogVersion_);
- isReady_.set(true);
+ catalogDeltaLog_.garbageCollect(newCatalogVersion);
// Notify all the threads waiting on a catalog update.
synchronized (catalogUpdateEventNotifier_) {
catalogUpdateEventNotifier_.notifyAll();
@@ -369,7 +367,7 @@ public class ImpaladCatalog extends Catalog {
"Unexpected TCatalogObjectType: " + catalogObject.getType());
}
- if (catalogObject.getCatalog_version() > lastSyncedCatalogVersion_) {
+ if (catalogObject.getCatalog_version() > lastSyncedCatalogVersion_.get()) {
catalogDeltaLog_.addRemovedObject(catalogObject);
}
}
@@ -513,10 +511,17 @@ public class ImpaladCatalog extends Catalog {
* received and processed a valid catalog topic update from the StateStore),
* false otherwise.
*/
- public boolean isReady() { return isReady_.get(); }
+ public boolean isReady() {
+ return lastSyncedCatalogVersion_.get() > INITIAL_CATALOG_VERSION;
+ }
// Only used for testing.
- public void setIsReady(boolean isReady) { isReady_.set(isReady); }
+ public void setIsReady(boolean isReady) {
+ lastSyncedCatalogVersion_.incrementAndGet();
+ synchronized (catalogUpdateEventNotifier_) {
+ catalogUpdateEventNotifier_.notifyAll();
+ }
+ }
public AuthorizationPolicy getAuthPolicy() { return authPolicy_; }
public String getDefaultKuduMasterHosts() { return defaultKuduMasterHosts_; }
http://git-wip-us.apache.org/repos/asf/impala/blob/85000c28/fe/src/main/java/org/apache/impala/service/Frontend.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index f03feb0..41395d6 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -799,9 +799,10 @@ public class Frontend {
/**
* Waits indefinitely for the local catalog to be ready. The catalog is "ready" after
- * the first catalog update is received from the statestore.
+ * the first catalog update with a version > INITIAL_CATALOG_VERSION is received from
+ * the statestore.
*
- * @see ImpaladCatalog.isReady
+ * @see ImpaladCatalog#isReady()
*/
public void waitForCatalog() {
LOG.info("Waiting for first catalog update from the statestore.");
[3/3] impala git commit: KUDU-2301: (Part-1) Add instrumentation on a
per connection level
Posted by ta...@apache.org.
KUDU-2301: (Part-1) Add instrumentation on a per connection level
This patch returns the OutboundTransfer queue size on a per
connection level and makes them accessible via the
DumpRunningRpcs() call.
A test is added in rpc-test to ensure that this metric works as
expected.
A future patch will add more metrics.
Change-Id: Iae1a5fe0066adf644a9cac41ad6696e1bbf00465
Reviewed-on: http://gerrit.cloudera.org:8080/9343
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Reviewed-on: http://gerrit.cloudera.org:8080/9383
Reviewed-by: Sailesh Mukil <sa...@cloudera.com>
Reviewed-by: Michael Ho <kw...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/258938ad
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/258938ad
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/258938ad
Branch: refs/heads/2.x
Commit: 258938ad9c0539e9f82a0cb6e142e1cbeaff377f
Parents: 85000c2
Author: Sailesh Mukil <sa...@apache.org>
Authored: Mon Feb 12 14:30:49 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Mar 1 05:34:02 2018 +0000
----------------------------------------------------------------------
be/src/kudu/rpc/connection.cc | 4 +-
be/src/kudu/rpc/connection.h | 5 +++
be/src/kudu/rpc/messenger.h | 1 +
be/src/kudu/rpc/rpc-test.cc | 67 ++++++++++++++++++++++++++++
be/src/kudu/rpc/rpc_introspection.proto | 1 +
5 files changed, 77 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/258938ad/be/src/kudu/rpc/connection.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/connection.cc b/be/src/kudu/rpc/connection.cc
index bc7446e..2cde8c8 100644
--- a/be/src/kudu/rpc/connection.cc
+++ b/be/src/kudu/rpc/connection.cc
@@ -20,6 +20,7 @@
#include <stdint.h>
#include <algorithm>
+#include <cerrno>
#include <iostream>
#include <memory>
#include <set>
@@ -50,7 +51,6 @@
#include "kudu/util/status.h"
#include "kudu/util/trace.h"
-using std::function;
using std::includes;
using std::set;
using std::shared_ptr;
@@ -746,6 +746,8 @@ Status Connection::DumpPB(const DumpRunningRpcsRequestPB& req,
c->call->DumpPB(req, resp->add_calls_in_flight());
}
}
+
+ resp->set_outbound_queue_size(num_queued_outbound_transfers());
} else if (direction_ == SERVER) {
if (negotiation_complete_) {
// It's racy to dump credentials while negotiating, since the Connection
http://git-wip-us.apache.org/repos/asf/impala/blob/258938ad/be/src/kudu/rpc/connection.h
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/connection.h b/be/src/kudu/rpc/connection.h
index 9a78c14..e40f814 100644
--- a/be/src/kudu/rpc/connection.h
+++ b/be/src/kudu/rpc/connection.h
@@ -18,6 +18,7 @@
#ifndef KUDU_RPC_CONNECTION_H
#define KUDU_RPC_CONNECTION_H
+#include <cstddef>
#include <cstdint>
#include <limits>
#include <memory>
@@ -220,6 +221,10 @@ class Connection : public RefCountedThreadSafe<Connection> {
scheduled_for_shutdown_ = true;
}
+ size_t num_queued_outbound_transfers() const {
+ return outbound_transfers_.size();
+ }
+
private:
friend struct CallAwaitingResponse;
friend class QueueTransferTask;
http://git-wip-us.apache.org/repos/asf/impala/blob/258938ad/be/src/kudu/rpc/messenger.h
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/messenger.h b/be/src/kudu/rpc/messenger.h
index b0099d3..d1a1d79 100644
--- a/be/src/kudu/rpc/messenger.h
+++ b/be/src/kudu/rpc/messenger.h
@@ -311,6 +311,7 @@ class Messenger {
private:
FRIEND_TEST(TestRpc, TestConnectionKeepalive);
FRIEND_TEST(TestRpc, TestConnectionAlwaysKeepalive);
+ FRIEND_TEST(TestRpc, TestClientConnectionsMetrics);
FRIEND_TEST(TestRpc, TestCredentialsPolicy);
FRIEND_TEST(TestRpc, TestReopenOutboundConnections);
http://git-wip-us.apache.org/repos/asf/impala/blob/258938ad/be/src/kudu/rpc/rpc-test.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/rpc-test.cc b/be/src/kudu/rpc/rpc-test.cc
index f76cc8f..dcbe5a7 100644
--- a/be/src/kudu/rpc/rpc-test.cc
+++ b/be/src/kudu/rpc/rpc-test.cc
@@ -20,10 +20,15 @@
#include <limits.h>
#include <memory>
#include <string>
+#include <unistd.h>
#include <unordered_map>
#include <vector>
#include <boost/bind.hpp>
+#include <boost/core/ref.hpp>
+#include <boost/function.hpp>
+#include <gflags/gflags_declare.h>
+#include <glog/logging.h>
#include <gtest/gtest.h>
#include "kudu/gutil/map-util.h"
@@ -31,6 +36,12 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/rpc/constants.h"
#include "kudu/rpc/outbound_call.h"
+#include "kudu/rpc/proxy.h"
+#include "kudu/rpc/reactor.h"
+#include "kudu/rpc/rpc_controller.h"
+#include "kudu/rpc/rpc_introspection.pb.h"
+#include "kudu/rpc/rpc_sidecar.h"
+#include "kudu/rpc/rtest.pb.h"
#include "kudu/rpc/serialization.h"
#include "kudu/security/test/test_certs.h"
#include "kudu/util/countdown_latch.h"
@@ -418,6 +429,62 @@ TEST_P(TestRpc, TestConnectionAlwaysKeepalive) {
ASSERT_EQ(1, metrics.num_client_connections_) << "Client should have 1 client connections";
}
+// Test that the metrics on a per connection level work accurately.
+TEST_P(TestRpc, TestClientConnectionMetrics) {
+ // Only run one reactor per messenger, so we can grab the metrics from that
+ // one without having to check all.
+ n_server_reactor_threads_ = 1;
+ keepalive_time_ms_ = -1;
+
+ // Set up server.
+ Sockaddr server_addr;
+ bool enable_ssl = GetParam();
+ StartTestServer(&server_addr, enable_ssl);
+
+ // Set up client.
+ LOG(INFO) << "Connecting to " << server_addr.ToString();
+ shared_ptr<Messenger> client_messenger(CreateMessenger("Client", 1, enable_ssl));
+ Proxy p(client_messenger, server_addr, server_addr.host(),
+ GenericCalculatorService::static_service_name());
+
+ // Cause the reactor thread to be blocked for 2 seconds.
+ server_messenger_->ScheduleOnReactor(boost::bind(sleep, 2), MonoDelta::FromSeconds(0));
+
+ RpcController controller;
+ DumpRunningRpcsRequestPB dump_req;
+ DumpRunningRpcsResponsePB dump_resp;
+ dump_req.set_include_traces(false);
+
+ // We'll send several calls asynchronously to force RPC queueing on the sender side.
+ int n_calls = 1000;
+ AddRequestPB add_req;
+ add_req.set_x(rand());
+ add_req.set_y(rand());
+ AddResponsePB add_resp;
+
+ vector<unique_ptr<RpcController>> controllers;
+ CountDownLatch latch(n_calls);
+ for (int i = 0; i < n_calls; i++) {
+ controllers.emplace_back(new RpcController());
+ p.AsyncRequest(GenericCalculatorService::kAddMethodName, add_req, &add_resp,
+ controllers.back().get(), boost::bind(&CountDownLatch::CountDown, boost::ref(latch)));
+ }
+
+ // Since we blocked the only reactor thread for sometime, we should see RPCs queued on the
+ // OutboundTransfer queue, unless the main thread is very slow.
+ ASSERT_OK(client_messenger->DumpRunningRpcs(dump_req, &dump_resp));
+ ASSERT_EQ(1, dump_resp.outbound_connections_size());
+ ASSERT_GT(dump_resp.outbound_connections(0).outbound_queue_size(), 0);
+
+ // Wait for the calls to be marked finished.
+ latch.Wait();
+
+ // Verify that all the RPCs have finished.
+ for (const auto& controller : controllers) {
+ ASSERT_TRUE(controller->finished());
+ }
+}
+
// Test that outbound connections to the same server are reopen upon every RPC
// call when the 'rpc_reopen_outbound_connections' flag is set.
TEST_P(TestRpc, TestReopenOutboundConnections) {
http://git-wip-us.apache.org/repos/asf/impala/blob/258938ad/be/src/kudu/rpc/rpc_introspection.proto
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/rpc_introspection.proto b/be/src/kudu/rpc/rpc_introspection.proto
index 5c4f4f1..7685903 100644
--- a/be/src/kudu/rpc/rpc_introspection.proto
+++ b/be/src/kudu/rpc/rpc_introspection.proto
@@ -62,6 +62,7 @@ message RpcConnectionPB {
// TODO: swap out for separate fields
optional string remote_user_credentials = 3;
repeated RpcCallInProgressPB calls_in_flight = 4;
+ optional int64 outbound_queue_size = 5;
}
message DumpRunningRpcsRequestPB {