diff --git a/be/src/rpc/impala-service-pool.cc b/be/src/rpc/impala-service-pool.cc
index 9db137a..74fe2df 100644
--- a/be/src/rpc/impala-service-pool.cc
+++ b/be/src/rpc/impala-service-pool.cc
@@ -24,6 +24,7 @@
 #include <vector>
 
 #include "exec/kudu-util.h"
+#include "gutil/strings/numbers.h"
 #include "kudu/gutil/gscoped_ptr.h"
 #include "kudu/gutil/ref_counted.h"
 #include "kudu/gutil/strings/join.h"
@@ -58,11 +59,13 @@
 
 ImpalaServicePool::ImpalaServicePool(const scoped_refptr<kudu::MetricEntity>& entity,
     int service_queue_length, kudu::rpc::GeneratedServiceIf* service,
-    MemTracker* service_mem_tracker)
+    MemTracker* service_mem_tracker, const TNetworkAddress& address)
   : service_mem_tracker_(service_mem_tracker),
     service_(service),
     service_queue_(service_queue_length),
-    incoming_queue_time_(METRIC_impala_incoming_queue_time.Instantiate(entity)) {
+    incoming_queue_time_(METRIC_impala_incoming_queue_time.Instantiate(entity)),
+    hostname_(address.hostname),
+    port_(SimpleItoa(address.port)) {
   DCHECK(service_mem_tracker_ != nullptr);
   const TMetricDef& overflow_metric_def =
       MetricDefs::Get(RPC_QUEUE_OVERFLOW_METRIC_KEY, service_->service_name());
@@ -188,10 +191,22 @@
     service_mem_tracker_->Consume(transfer_size);
   }
 
-  Status debug_status = DebugAction(FLAGS_debug_actions, "SERVICE_POOL_SERVER_BUSY");
+  // Debug action for simulating rpc errors. To use, specify:
+  // --debug_actions=IMPALA_SERVICE_POOL:<hostname>:<port>:<rpc>:<action>
+  // where <hostname> and <port> represent the impalad receiving the rpc, <port> is the BE
+  // krpc port (default 27000), <rpc> is the name of an rpc in the service, eg.
+  // 'TransmitData' or 'ReportExecStatus', and <action> is any of the debug actions, eg.
+  // FAIL or SLEEP.
+  Status debug_status = DebugAction(FLAGS_debug_actions, "IMPALA_SERVICE_POOL",
+      {hostname_, port_, c->remote_method().method_name()});
   if (UNLIKELY(!debug_status.ok())) {
-    // Simulate the service being too busy.
-    RejectTooBusy(c);
+    if (debug_status.msg().msg() == "REJECT_TOO_BUSY") {
+      // Simulate the service being too busy.
+      RejectTooBusy(c);
+    } else {
+      FailAndReleaseRpc(kudu::rpc::ErrorStatusPB::FATAL_UNKNOWN,
+          kudu::Status::RuntimeError(debug_status.msg().msg()), c);
+    }
     return kudu::Status::OK();
   }
 
diff --git a/be/src/rpc/impala-service-pool.h b/be/src/rpc/impala-service-pool.h
index dd56436..7771358 100644
--- a/be/src/rpc/impala-service-pool.h
+++ b/be/src/rpc/impala-service-pool.h
@@ -45,9 +45,11 @@
   ///
   /// 'service_mem_tracker' is the MemTracker for tracking the memory usage of RPC
   /// payloads in the service queue.
+  ///
+  /// 'address' is the ip address and port that 'service' runs on.
   ImpalaServicePool(const scoped_refptr<kudu::MetricEntity>& entity,
       int service_queue_length, kudu::rpc::GeneratedServiceIf* service,
-      MemTracker* service_mem_tracker);
+      MemTracker* service_mem_tracker, const TNetworkAddress& address);
 
   virtual ~ImpalaServicePool();
 
@@ -111,6 +113,10 @@
   boost::mutex shutdown_lock_;
   bool closing_ = false;
 
+  /// The address this service is running on.
+  const std::string hostname_;
+  const std::string port_;
+
   DISALLOW_COPY_AND_ASSIGN(ImpalaServicePool);
 };
 
diff --git a/be/src/rpc/rpc-mgr-kerberized-test.cc b/be/src/rpc/rpc-mgr-kerberized-test.cc
index 86c9eaa..62b15e4 100644
--- a/be/src/rpc/rpc-mgr-kerberized-test.cc
+++ b/be/src/rpc/rpc-mgr-kerberized-test.cc
@@ -74,7 +74,7 @@
 
   // Enable TLS.
   ScopedSetTlsFlags s(SERVER_CERT, PRIVATE_KEY, SERVER_CERT);
-  ASSERT_OK(tls_rpc_mgr.Init());
+  ASSERT_OK(tls_rpc_mgr.Init(tls_krpc_address));
 
   ASSERT_OK(RunMultipleServicesTest(&tls_rpc_mgr, tls_krpc_address));
   tls_rpc_mgr.Shutdown();
@@ -95,7 +95,7 @@
       static_cast<ScanMemServiceImpl*>(scan_mem_impl)->mem_tracker()));
   FLAGS_num_acceptor_threads = 2;
   FLAGS_num_reactor_threads = 10;
-  ASSERT_OK(rpc_mgr_.StartServices(krpc_address_));
+  ASSERT_OK(rpc_mgr_.StartServices());
 
   // Switch over to a credentials cache which only contains the dummy credential.
   // Kinit done in InitAuth() uses a different credentials cache.
diff --git a/be/src/rpc/rpc-mgr-test.cc b/be/src/rpc/rpc-mgr-test.cc
index c0d4fbd..cb698fa 100644
--- a/be/src/rpc/rpc-mgr-test.cc
+++ b/be/src/rpc/rpc-mgr-test.cc
@@ -51,7 +51,7 @@
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
   ScopedSetTlsFlags s(SERVER_CERT, PRIVATE_KEY, SERVER_CERT);
-  ASSERT_OK(tls_rpc_mgr.Init());
+  ASSERT_OK(tls_rpc_mgr.Init(tls_krpc_address));
 
   ASSERT_OK(RunMultipleServicesTest(&tls_rpc_mgr, tls_krpc_address));
   tls_rpc_mgr.Shutdown();
@@ -74,7 +74,7 @@
   int32_t tls_service_port = FindUnusedEphemeralPort();
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
-  ASSERT_FALSE(tls_rpc_mgr.Init().ok());
+  ASSERT_FALSE(tls_rpc_mgr.Init(tls_krpc_address).ok());
   tls_rpc_mgr.Shutdown();
 }
 
@@ -91,7 +91,7 @@
   int32_t tls_service_port = FindUnusedEphemeralPort();
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
-  ASSERT_FALSE(tls_rpc_mgr.Init().ok());
+  ASSERT_FALSE(tls_rpc_mgr.Init(tls_krpc_address).ok());
   tls_rpc_mgr.Shutdown();
 }
 
@@ -108,7 +108,7 @@
   int32_t tls_service_port = FindUnusedEphemeralPort();
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
-  ASSERT_OK(tls_rpc_mgr.Init());
+  ASSERT_OK(tls_rpc_mgr.Init(tls_krpc_address));
   ASSERT_OK(RunMultipleServicesTest(&tls_rpc_mgr, tls_krpc_address));
   tls_rpc_mgr.Shutdown();
 }
@@ -125,7 +125,7 @@
   int32_t tls_service_port = FindUnusedEphemeralPort();
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
-  ASSERT_FALSE(tls_rpc_mgr.Init().ok());
+  ASSERT_FALSE(tls_rpc_mgr.Init(tls_krpc_address).ok());
   tls_rpc_mgr.Shutdown();
 }
 
@@ -142,7 +142,7 @@
   int32_t tls_service_port = FindUnusedEphemeralPort();
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
-  ASSERT_OK(tls_rpc_mgr.Init());
+  ASSERT_OK(tls_rpc_mgr.Init(tls_krpc_address));
   ASSERT_OK(RunMultipleServicesTest(&tls_rpc_mgr, tls_krpc_address));
   tls_rpc_mgr.Shutdown();
 }
@@ -161,7 +161,7 @@
   int32_t tls_service_port = FindUnusedEphemeralPort();
   tls_krpc_address = MakeNetworkAddress(ip, tls_service_port);
 
-  ASSERT_OK(tls_rpc_mgr.Init());
+  ASSERT_OK(tls_rpc_mgr.Init(tls_krpc_address));
   ASSERT_OK(RunMultipleServicesTest(&tls_rpc_mgr, tls_krpc_address));
   tls_rpc_mgr.Shutdown();
 }
@@ -186,7 +186,7 @@
 
   FLAGS_num_acceptor_threads = 2;
   FLAGS_num_reactor_threads = 10;
-  ASSERT_OK(rpc_mgr_.StartServices(krpc_address_));
+  ASSERT_OK(rpc_mgr_.StartServices());
 
   unique_ptr<PingServiceProxy> proxy;
   ASSERT_OK(static_cast<PingServiceImpl*>(ping_impl)->GetProxy(krpc_address_,
@@ -216,7 +216,7 @@
 
   FLAGS_num_acceptor_threads = 2;
   FLAGS_num_reactor_threads = 10;
-  ASSERT_OK(rpc_mgr_.StartServices(krpc_address_));
+  ASSERT_OK(rpc_mgr_.StartServices());
 
   RpcController controller;
   srand(0);
@@ -251,7 +251,7 @@
   int32_t secondary_service_port = FindUnusedEphemeralPort();
   secondary_krpc_address = MakeNetworkAddress(ip, secondary_service_port);
 
-  ASSERT_OK(secondary_rpc_mgr.Init());
+  ASSERT_OK(secondary_rpc_mgr.Init(secondary_krpc_address));
   ASSERT_FALSE(RunMultipleServicesTest(&secondary_rpc_mgr, secondary_krpc_address).ok());
   secondary_rpc_mgr.Shutdown();
 }
@@ -299,7 +299,7 @@
       static_cast<PingServiceImpl*>(ping_impl)->mem_tracker()));
   FLAGS_num_acceptor_threads = 2;
   FLAGS_num_reactor_threads = 10;
-  ASSERT_OK(rpc_mgr_.StartServices(krpc_address_));
+  ASSERT_OK(rpc_mgr_.StartServices());
 
   // Find the counter which tracks the number of times the service queue is too full.
   const string& overflow_count = Substitute(
@@ -318,8 +318,9 @@
 
   // Use DebugAction to make the Impala Service Pool reject 50% of Krpc calls as if the
   // service is too busy.
-  auto s = ScopedFlagSetter<string>::Make(
-      &FLAGS_debug_actions, "SERVICE_POOL_SERVER_BUSY:FAIL@0.5");
+  auto s = ScopedFlagSetter<string>::Make(&FLAGS_debug_actions,
+      Substitute("IMPALA_SERVICE_POOL:$0:$1:Ping:FAIL@0.5@REJECT_TOO_BUSY",
+          krpc_address_.hostname, krpc_address_.port));
   PingRequestPB request;
   PingResponsePB response;
   const int64_t timeout_ms = 10 * MILLIS_PER_SEC;
diff --git a/be/src/rpc/rpc-mgr-test.h b/be/src/rpc/rpc-mgr-test.h
index ad428a4..f559409 100644
--- a/be/src/rpc/rpc-mgr-test.h
+++ b/be/src/rpc/rpc-mgr-test.h
@@ -139,7 +139,7 @@
     ASSERT_OK(HostnameToIpAddr(FLAGS_hostname, &ip));
     krpc_address_ = MakeNetworkAddress(ip, SERVICE_PORT);
     exec_env_.reset(new ExecEnv());
-    ASSERT_OK(rpc_mgr_.Init());
+    ASSERT_OK(rpc_mgr_.Init(krpc_address_));
   }
 
   virtual void TearDown() {
@@ -303,7 +303,7 @@
 
   FLAGS_num_acceptor_threads = 2;
   FLAGS_num_reactor_threads = 10;
-  RETURN_IF_ERROR(rpc_mgr->StartServices(krpc_address));
+  RETURN_IF_ERROR(rpc_mgr->StartServices());
 
   unique_ptr<PingServiceProxy> ping_proxy;
   RETURN_IF_ERROR(static_cast<PingServiceImpl*>(ping_impl)->GetProxy(krpc_address,
diff --git a/be/src/rpc/rpc-mgr.cc b/be/src/rpc/rpc-mgr.cc
index 3b407d7..09893e0 100644
--- a/be/src/rpc/rpc-mgr.cc
+++ b/be/src/rpc/rpc-mgr.cc
@@ -90,7 +90,10 @@
 
 namespace impala {
 
-Status RpcMgr::Init() {
+Status RpcMgr::Init(const TNetworkAddress& address) {
+  DCHECK(IsResolvedAddress(address));
+  address_ = address;
+
   // Log any RPCs which take longer than this threshold on the server.
   FLAGS_rpc_duration_too_long_ms = FLAGS_impala_slow_rpc_threshold_ms;
 
@@ -147,8 +150,9 @@
     GeneratedServiceIf* service_ptr, MemTracker* service_mem_tracker) {
   DCHECK(is_inited()) << "Must call Init() before RegisterService()";
   DCHECK(!services_started_) << "Cannot call RegisterService() after StartServices()";
-  scoped_refptr<ImpalaServicePool> service_pool = new ImpalaServicePool(
-      messenger_->metric_entity(), service_queue_depth, service_ptr, service_mem_tracker);
+  scoped_refptr<ImpalaServicePool> service_pool =
+      new ImpalaServicePool(messenger_->metric_entity(), service_queue_depth, service_ptr,
+          service_mem_tracker, address_);
   // Start the thread pool first before registering the service in case the startup fails.
   RETURN_IF_ERROR(service_pool->Init(num_service_threads));
   KUDU_RETURN_IF_ERROR(
@@ -185,20 +189,19 @@
   return true;
 }
 
-Status RpcMgr::StartServices(const TNetworkAddress& address) {
+Status RpcMgr::StartServices() {
   DCHECK(is_inited()) << "Must call Init() before StartServices()";
   DCHECK(!services_started_) << "May not call StartServices() twice";
 
-  // Convert 'address' to Kudu's Sockaddr
-  DCHECK(IsResolvedAddress(address));
+  // Convert 'address_' to Kudu's Sockaddr
   Sockaddr sockaddr;
   if (FLAGS_rpc_use_loopback) {
     // Listen on all addresses, including loopback.
-    sockaddr.set_port(address.port);
+    sockaddr.set_port(address_.port);
     DCHECK(sockaddr.IsWildcard()) << sockaddr.ToString();
   } else {
     // Only listen on the canonical address for KRPC.
-    RETURN_IF_ERROR(TNetworkAddressToSockaddr(address, &sockaddr));
+    RETURN_IF_ERROR(TNetworkAddressToSockaddr(address_, &sockaddr));
   }
 
   // Call the messenger to create an AcceptorPool for us.
diff --git a/be/src/rpc/rpc-mgr.h b/be/src/rpc/rpc-mgr.h
index e88579f..a141dba 100644
--- a/be/src/rpc/rpc-mgr.h
+++ b/be/src/rpc/rpc-mgr.h
@@ -102,16 +102,16 @@
  public:
   RpcMgr(bool use_tls = false) : use_tls_(use_tls) {}
 
-  /// Initializes the reactor threads, and prepares for sending outbound RPC requests.
-  Status Init() WARN_UNUSED_RESULT;
+  /// Initializes the reactor threads, and prepares for sending outbound RPC requests. All
+  /// services will be started on 'address', which must be a resolved IP address.
+  Status Init(const TNetworkAddress& address) WARN_UNUSED_RESULT;
 
   bool is_inited() const { return messenger_.get() != nullptr; }
 
-  /// Start the acceptor threads which listen on 'address', making KRPC services
-  /// available. 'address' has to be a resolved IP address. Before this method is called,
-  /// remote clients will get a 'connection refused' error when trying to invoke an RPC
-  /// on this host.
-  Status StartServices(const TNetworkAddress& address) WARN_UNUSED_RESULT;
+  /// Start the acceptor threads which listen on 'address_', making KRPC services
+  /// available. Before this method is called, remote clients will get a 'connection
+  /// refused' error when trying to invoke an RPC on this host.
+  Status StartServices() WARN_UNUSED_RESULT;
 
   /// Register a new service.
   ///
@@ -223,6 +223,9 @@
   /// True if TLS is configured for communication between Impala backends. messenger_ will
   /// be configured to use TLS if this is set.
   const bool use_tls_;
+
+  /// The host/port the rpc services are run on.
+  TNetworkAddress address_;
 };
 
 } // namespace impala
diff --git a/be/src/runtime/backend-client.h b/be/src/runtime/backend-client.h
index 92279fc..fee4d39 100644
--- a/be/src/runtime/backend-client.h
+++ b/be/src/runtime/backend-client.h
@@ -19,7 +19,6 @@
 #define IMPALA_BACKEND_CLIENT_H
 
 #include "runtime/client-cache.h"
-#include "testutil/fault-injection-util.h"
 #include "util/runtime-profile-counters.h"
 
 #include "gen-cpp/ImpalaInternalService.h"
diff --git a/be/src/runtime/data-stream-test.cc b/be/src/runtime/data-stream-test.cc
index 5d39c56..3121ee9 100644
--- a/be/src/runtime/data-stream-test.cc
+++ b/be/src/runtime/data-stream-test.cc
@@ -492,12 +492,12 @@
   void StartKrpcBackend() {
     RpcMgr* rpc_mgr = exec_env_->rpc_mgr();
     KrpcDataStreamMgr* krpc_stream_mgr = exec_env_->stream_mgr();
-    ASSERT_OK(rpc_mgr->Init());
+    ASSERT_OK(rpc_mgr->Init(krpc_address_));
     test_service_.reset(new ImpalaKRPCTestBackend(rpc_mgr, krpc_stream_mgr,
         exec_env_->process_mem_tracker()));
     ASSERT_OK(test_service_->Init());
     ASSERT_OK(krpc_stream_mgr->Init(test_service_->mem_tracker()));
-    ASSERT_OK(rpc_mgr->StartServices(krpc_address_));
+    ASSERT_OK(rpc_mgr->StartServices());
   }
 
   void StopKrpcBackend() {
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index e80a2a7..22049b1 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -355,7 +355,7 @@
   // Initializes the RPCMgr, ControlServices and DataStreamServices.
   // Initialization needs to happen in the following order due to dependencies:
   // - RPC manager, DataStreamService and DataStreamManager.
-  RETURN_IF_ERROR(rpc_mgr_->Init());
+  RETURN_IF_ERROR(rpc_mgr_->Init(krpc_address_));
   control_svc_.reset(new ControlService(rpc_metrics_));
   RETURN_IF_ERROR(control_svc_->Init());
   data_svc_.reset(new DataStreamService(rpc_metrics_));
@@ -508,7 +508,7 @@
 
 Status ExecEnv::StartKrpcService() {
   LOG(INFO) << "Starting KRPC service";
-  RETURN_IF_ERROR(rpc_mgr_->StartServices(krpc_address_));
+  RETURN_IF_ERROR(rpc_mgr_->StartServices());
   return Status::OK();
 }
 
diff --git a/be/src/runtime/test-env.cc b/be/src/runtime/test-env.cc
index dad65e1..6e95174 100644
--- a/be/src/runtime/test-env.cc
+++ b/be/src/runtime/test-env.cc
@@ -79,7 +79,7 @@
   IpAddr ip_address;
   RETURN_IF_ERROR(HostnameToIpAddr(FLAGS_hostname, &ip_address));
   exec_env_->krpc_address_.__set_hostname(ip_address);
-  RETURN_IF_ERROR(exec_env_->rpc_mgr_->Init());
+  RETURN_IF_ERROR(exec_env_->rpc_mgr_->Init(exec_env_->krpc_address_));
   exec_env_->control_svc_.reset(new ControlService(exec_env_->rpc_metrics_));
   RETURN_IF_ERROR(exec_env_->control_svc_->Init());
 
diff --git a/be/src/service/control-service.cc b/be/src/service/control-service.cc
index 08f20c2..0ba51bc 100644
--- a/be/src/service/control-service.cc
+++ b/be/src/service/control-service.cc
@@ -31,7 +31,6 @@
 #include "runtime/query-state.h"
 #include "service/client-request-state.h"
 #include "service/impala-server.h"
-#include "testutil/fault-injection-util.h"
 #include "util/debug-util.h"
 #include "util/memory-metrics.h"
 #include "util/parse-util.h"
diff --git a/be/src/service/data-stream-service.cc b/be/src/service/data-stream-service.cc
index 890ceec..42b14f9 100644
--- a/be/src/service/data-stream-service.cc
+++ b/be/src/service/data-stream-service.cc
@@ -32,7 +32,6 @@
 #include "runtime/row-batch.h"
 #include "util/memory-metrics.h"
 #include "util/parse-util.h"
-#include "testutil/fault-injection-util.h"
 
 #include "gen-cpp/data_stream_service.pb.h"
 #include "gen-cpp/data_stream_service.proxy.h"
diff --git a/be/src/service/impala-internal-service.cc b/be/src/service/impala-internal-service.cc
index ccd00a8..22a8243 100644
--- a/be/src/service/impala-internal-service.cc
+++ b/be/src/service/impala-internal-service.cc
@@ -25,7 +25,6 @@
 #include "runtime/query-state.h"
 #include "runtime/fragment-instance-state.h"
 #include "runtime/exec-env.h"
-#include "testutil/fault-injection-util.h"
 
 #include "common/names.h"
 
diff --git a/be/src/testutil/CMakeLists.txt b/be/src/testutil/CMakeLists.txt
index 828cf6a..17885fa 100644
--- a/be/src/testutil/CMakeLists.txt
+++ b/be/src/testutil/CMakeLists.txt
@@ -26,7 +26,6 @@
 
 add_library(TestUtil
   desc-tbl-builder.cc
-  fault-injection-util.cc
   impalad-query-executor.cc
   in-process-servers.cc
   mini-kdc-wrapper.cc
diff --git a/be/src/testutil/fault-injection-util.cc b/be/src/testutil/fault-injection-util.cc
deleted file mode 100644
index 48d3a9e..0000000
--- a/be/src/testutil/fault-injection-util.cc
+++ /dev/null
@@ -1,88 +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.
-
-#ifndef NDEBUG
-
-#include "testutil/fault-injection-util.h"
-
-#include <random>
-
-#include <thrift/transport/TSSLSocket.h>
-#include <thrift/transport/TTransportException.h>
-
-#include "common/atomic.h"
-
-#include "common/names.h"
-
-DECLARE_int32(fault_injection_rpc_exception_type);
-
-namespace impala {
-
-using apache::thrift::transport::TTransportException;
-using apache::thrift::transport::TSSLException;
-
-void FaultInjectionUtil::InjectRpcException(bool is_send, int freq) {
-  static AtomicInt32 send_count(-1);
-  static AtomicInt32 recv_count(-1);
-  int32_t xcp_type = FLAGS_fault_injection_rpc_exception_type;
-  if (xcp_type == RPC_EXCEPTION_NONE) return;
-
-  // We currently support injecting exception at some RPCs only.
-  if (is_send) {
-    if (send_count.Add(1) % freq == 0) {
-      switch (xcp_type) {
-        case RPC_EXCEPTION_SEND_CLOSED_CONNECTION:
-          throw TTransportException(TTransportException::NOT_OPEN,
-              "Called write on non-open socket");
-        case RPC_EXCEPTION_SEND_TIMEDOUT:
-          throw TTransportException(TTransportException::TIMED_OUT,
-              "send timeout expired");
-        case RPC_EXCEPTION_SSL_SEND_CLOSED_CONNECTION:
-          throw TTransportException(TTransportException::NOT_OPEN);
-        case RPC_EXCEPTION_SSL_SEND_TIMEDOUT:
-          throw TSSLException("SSL_write: Resource temporarily unavailable");
-        // Simulate half-opened connections.
-        case RPC_EXCEPTION_SEND_STALE_CONNECTION:
-          throw TTransportException(TTransportException::END_OF_FILE,
-             "No more data to read.");
-        case RPC_EXCEPTION_SSL_SEND_STALE_CONNECTION:
-          throw TSSLException("SSL_read: Connection reset by peer");
-        // fall through for the default case.
-      }
-    }
-  } else {
-    if (recv_count.Add(1) % freq == 0) {
-      switch (xcp_type) {
-        case RPC_EXCEPTION_RECV_CLOSED_CONNECTION:
-          throw TTransportException(TTransportException::NOT_OPEN,
-              "Called read on non-open socket");
-        case RPC_EXCEPTION_RECV_TIMEDOUT:
-          throw TTransportException(TTransportException::TIMED_OUT,
-              "EAGAIN (timed out)");
-        case RPC_EXCEPTION_SSL_RECV_CLOSED_CONNECTION:
-          throw TTransportException(TTransportException::NOT_OPEN);
-        case RPC_EXCEPTION_SSL_RECV_TIMEDOUT:
-          throw TSSLException("SSL_read: Resource temporarily unavailable");
-        // fall through for the default case.
-      }
-    }
-  }
-}
-
-}
-
-#endif
diff --git a/be/src/testutil/fault-injection-util.h b/be/src/testutil/fault-injection-util.h
deleted file mode 100644
index 0ab77d0..0000000
--- a/be/src/testutil/fault-injection-util.h
+++ /dev/null
@@ -1,65 +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.
-
-#ifndef IMPALA_TESTUTIL_FAULT_INJECTION_UTIL_H
-#define IMPALA_TESTUTIL_FAULT_INJECTION_UTIL_H
-
-#include "util/time.h"
-
-namespace impala {
-
-#ifndef NDEBUG
-
-class FaultInjectionUtil {
- public:
-
-  enum RpcExceptionType {
-    RPC_EXCEPTION_NONE = 0,
-    RPC_EXCEPTION_SEND_CLOSED_CONNECTION,
-    RPC_EXCEPTION_SEND_STALE_CONNECTION,
-    RPC_EXCEPTION_SEND_TIMEDOUT,
-    RPC_EXCEPTION_RECV_CLOSED_CONNECTION,
-    RPC_EXCEPTION_RECV_TIMEDOUT,
-    RPC_EXCEPTION_SSL_SEND_CLOSED_CONNECTION,
-    RPC_EXCEPTION_SSL_SEND_STALE_CONNECTION,
-    RPC_EXCEPTION_SSL_SEND_TIMEDOUT,
-    RPC_EXCEPTION_SSL_RECV_CLOSED_CONNECTION,
-    RPC_EXCEPTION_SSL_RECV_TIMEDOUT,
-  };
-
-  /// Test util function that injects exceptions to RPC client functions.
-  /// 'is_send' indicates whether injected fault is at the send() or recv() of an RPC.
-  /// The exception specified in 'FLAGS_fault_injection_rpc_exception_type' is injected
-  /// on every 'freq' invocations of this function.
-  static void InjectRpcException(bool is_send, int freq);
-
-};
-
-#define FAULT_INJECTION_SEND_RPC_EXCEPTION(freq)                 \
-    FaultInjectionUtil::InjectRpcException(true, freq)
-#define FAULT_INJECTION_RECV_RPC_EXCEPTION(freq)                 \
-    FaultInjectionUtil::InjectRpcException(false, freq)
-
-#else // NDEBUG
-
-#define FAULT_INJECTION_SEND_RPC_EXCEPTION(freq)
-#define FAULT_INJECTION_RECV_RPC_EXCEPTION(freq)
-
-#endif
-
-}
-#endif // IMPALA_TESTUTIL_FAULT_INJECTION_UTIL_H
diff --git a/be/src/util/debug-util.cc b/be/src/util/debug-util.cc
index 1bad6fc..822098e 100644
--- a/be/src/util/debug-util.cc
+++ b/be/src/util/debug-util.cc
@@ -26,10 +26,13 @@
 #include "common/version.h"
 #include "runtime/collection-value.h"
 #include "runtime/descriptors.h"
+#include "runtime/exec-env.h"
 #include "runtime/raw-value.inline.h"
 #include "runtime/tuple-row.h"
 #include "runtime/row-batch.h"
 #include "util/cpu-info.h"
+#include "util/impalad-metrics.h"
+#include "util/metrics.h"
 #include "util/string-parser.h"
 #include "util/uid-util.h"
 #include "util/time.h"
@@ -329,63 +332,82 @@
   return true;
 }
 
-Status DebugActionImpl(const string& debug_action, const char* label) {
+Status DebugActionImpl(
+    const string& debug_action, const char* label, const std::vector<string>& args) {
   const DebugActionTokens& action_list = TokenizeDebugActions(debug_action);
   static const char ERROR_MSG[] = "Invalid debug_action $0:$1 ($2)";
   for (const vector<string>& components : action_list) {
-    // size() != 2 check filters out ExecNode debug actions.
-    if (components.size() != 2 || !iequals(components[0], label)) continue;
+    // 'components' should be of the form {label, arg_0, ..., arg_n, action}
+    if (components.size() != 2 + args.size() || !iequals(components[0], label)) {
+      continue;
+    }
+    // Check if the arguments match.
+    bool matches = true;
+    for (int i = 0; i < args.size(); ++i) {
+      if (!iequals(components[i + 1], args[i])) {
+        matches = false;
+        break;
+      }
+    }
+    if (!matches) continue;
+    const string& action_str = components[args.size() + 1];
     // 'tokens' becomes {command, param0, param1, ... }
-    vector<string> tokens = TokenizeDebugActionParams(components[1]);
+    vector<string> tokens = TokenizeDebugActionParams(action_str);
     DCHECK_GE(tokens.size(), 1);
     const string& cmd = tokens[0];
     int sleep_millis = 0;
     if (iequals(cmd, "SLEEP")) {
       // SLEEP@<millis>
       if (tokens.size() != 2) {
-        return Status(Substitute(ERROR_MSG, components[0], components[1],
-                "expected SLEEP@<ms>"));
+        return Status(
+            Substitute(ERROR_MSG, components[0], action_str, "expected SLEEP@<ms>"));
       }
       sleep_millis = atoi(tokens[1].c_str());
     } else if (iequals(cmd, "JITTER")) {
       // JITTER@<millis>[@<probability>}
       if (tokens.size() < 2 || tokens.size() > 3) {
-        return Status(Substitute(ERROR_MSG, components[0], components[1],
-                "expected JITTER@<ms>[@<probability>]"));
+        return Status(Substitute(ERROR_MSG, components[0], action_str,
+            "expected JITTER@<ms>[@<probability>]"));
       }
       int max_millis = atoi(tokens[1].c_str());
       if (tokens.size() == 3) {
         bool should_execute = true;
         if (!ParseProbability(tokens[2], &should_execute)) {
-          return Status(Substitute(ERROR_MSG, components[0], components[1],
-                  "invalid probability"));
+          return Status(
+              Substitute(ERROR_MSG, components[0], action_str, "invalid probability"));
         }
         if (!should_execute) continue;
       }
       sleep_millis = rand() % (max_millis + 1);
     } else if (iequals(cmd, "FAIL")) {
-      // FAIL[@<probability>]
-      if (tokens.size() > 2) {
-        return Status(Substitute(ERROR_MSG, components[0], components[1],
-                "expected FAIL[@<probability>]"));
+      // FAIL[@<probability>][@<error message>]
+      if (tokens.size() > 3) {
+        return Status(Substitute(ERROR_MSG, components[0], action_str,
+            "expected FAIL[@<probability>][@<error message>]"));
       }
-      if (tokens.size() == 2) {
+      if (tokens.size() >= 2) {
         bool should_execute = true;
         if (!ParseProbability(tokens[1], &should_execute)) {
-          return Status(Substitute(ERROR_MSG, components[0], components[1],
-                  "invalid probability"));
+          return Status(
+              Substitute(ERROR_MSG, components[0], action_str, "invalid probability"));
         }
         if (!should_execute) continue;
       }
-      return Status(TErrorCode::INTERNAL_ERROR, Substitute("Debug Action: $0:$1",
-              components[0], components[1]));
+      string error_msg = tokens.size() == 3 ?
+          tokens[2] :
+          Substitute("Debug Action: $0:$1", components[0], action_str);
+
+      if (ImpaladMetrics::DEBUG_ACTION_NUM_FAIL != nullptr) {
+        ImpaladMetrics::DEBUG_ACTION_NUM_FAIL->Increment(1l);
+      }
+      return Status(TErrorCode::INTERNAL_ERROR, error_msg);
     } else {
-      return Status(Substitute(ERROR_MSG, components[0], components[1],
-              "invalid command"));
+      DCHECK(false) << "Invalid debug action";
+      return Status(Substitute(ERROR_MSG, components[0], action_str, "invalid command"));
     }
     if (sleep_millis > 0) {
-      VLOG(1) << Substitute("Debug Action: $0:$1 sleeping for $2 ms",
-          components[0], components[1], sleep_millis);
+      VLOG(1) << Substitute("Debug Action: $0:$1 sleeping for $2 ms", components[0],
+          action_str, sleep_millis);
       SleepForMs(sleep_millis);
     }
   }
diff --git a/be/src/util/debug-util.h b/be/src/util/debug-util.h
index f88495b..d0fcad8 100644
--- a/be/src/util/debug-util.h
+++ b/be/src/util/debug-util.h
@@ -142,16 +142,18 @@
 std::vector<std::string> TokenizeDebugActionParams(const string& action);
 
 /// Slow path implementing DebugAction() for the case where 'debug_action' is non-empty.
-Status DebugActionImpl(const string& debug_action, const char* label) WARN_UNUSED_RESULT;
+Status DebugActionImpl(const string& debug_action, const char* label,
+    const std::vector<string>& args) WARN_UNUSED_RESULT;
 
 /// If debug_action query option has a "global action" (i.e. not exec-node specific)
-/// and matches the given 'label', apply the the action. See ImpalaService.thrift for
-/// details of the format and available global actions. For ExecNode code, use
-/// ExecNode::ExecDebugAction() instead.
-WARN_UNUSED_RESULT static inline Status DebugAction(
-    const string& debug_action, const char* label) {
+/// and matches the given 'label' and 'args', apply the the action. See
+/// ImpalaService.thrift for details of the format and available global actions. For
+/// ExecNode code, use ExecNode::ExecDebugAction() instead. Will return OK unless either
+/// an invalid debug action is specified or the FAIL action is executed.
+WARN_UNUSED_RESULT static inline Status DebugAction(const string& debug_action,
+    const char* label, const std::vector<string>& args = std::vector<string>()) {
   if (LIKELY(debug_action.empty())) return Status::OK();
-  return DebugActionImpl(debug_action, label);
+  return DebugActionImpl(debug_action, label, args);
 }
 
 WARN_UNUSED_RESULT static inline Status DebugAction(
diff --git a/be/src/util/impalad-metrics.cc b/be/src/util/impalad-metrics.cc
index d190564..7238284 100644
--- a/be/src/util/impalad-metrics.cc
+++ b/be/src/util/impalad-metrics.cc
@@ -24,6 +24,7 @@
 
 #include "common/names.h"
 
+DECLARE_string(debug_actions);
 DECLARE_bool(use_local_catalog);
 
 namespace impala {
@@ -135,6 +136,7 @@
     "impala-server.hedged-read-ops";
 const char* ImpaladMetricKeys::HEDGED_READ_OPS_WIN =
     "impala-server.hedged-read-ops-win";
+const char* ImpaladMetricKeys::DEBUG_ACTION_NUM_FAIL = "impala.debug_action.fail";
 
 // These are created by impala-server during startup.
 // =======
@@ -168,6 +170,7 @@
 IntCounter* ImpaladMetrics::CATALOG_CACHE_MISS_COUNT = nullptr;
 IntCounter* ImpaladMetrics::CATALOG_CACHE_REQUEST_COUNT = nullptr;
 IntCounter* ImpaladMetrics::CATALOG_CACHE_TOTAL_LOAD_TIME = nullptr;
+IntCounter* ImpaladMetrics::DEBUG_ACTION_NUM_FAIL = nullptr;
 
 // Gauges
 IntGauge* ImpaladMetrics::CATALOG_NUM_DBS = nullptr;
@@ -354,6 +357,9 @@
   HEDGED_READ_OPS = m->AddCounter(ImpaladMetricKeys::HEDGED_READ_OPS, 0);
   HEDGED_READ_OPS_WIN = m->AddCounter(ImpaladMetricKeys::HEDGED_READ_OPS_WIN, 0);
 
+  if (!FLAGS_debug_actions.empty()) {
+    DEBUG_ACTION_NUM_FAIL = m->AddCounter(ImpaladMetricKeys::DEBUG_ACTION_NUM_FAIL, 0);
+  }
 }
 
 }
diff --git a/be/src/util/impalad-metrics.h b/be/src/util/impalad-metrics.h
index fafbb5f..7f20ccc 100644
--- a/be/src/util/impalad-metrics.h
+++ b/be/src/util/impalad-metrics.h
@@ -209,6 +209,10 @@
   /// (i.e. returned faster than original read).
   static const char* HEDGED_READ_OPS_WIN;
 
+  /// Total number of times the FAIL debug action is hit. The counter is only created if
+  /// --debug_actions is set.
+  static const char* DEBUG_ACTION_NUM_FAIL;
+
 };
 
 /// Global impalad-wide metrics.  This is useful for objects that want to update metrics
@@ -251,6 +255,7 @@
   static IntCounter* CATALOG_CACHE_MISS_COUNT;
   static IntCounter* CATALOG_CACHE_REQUEST_COUNT;
   static IntCounter* CATALOG_CACHE_TOTAL_LOAD_TIME;
+  static IntCounter* DEBUG_ACTION_NUM_FAIL;
 
   // Gauges
   static IntGauge* CATALOG_NUM_DBS;
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 33d50ba..3a733c5 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -89,15 +89,28 @@
   //  invalid, the option is ignored.
   //
   // 2. Global actions
-  //  "<global label>:<command>@<param0>@<param1>@...<paramN>",
-  //  global labels are marked in the code with DEBUG_ACTION*() macros.
+  //  "<global label>:<arg0>:...:<argN>:<command>@<param0>@<param1>@...<paramN>",
+  //  Used with the DebugAction() call, the action will be performed if the label and
+  //  optional arguments all match. The arguments can be used to make the debug action
+  //  context dependent, for example to only fail rpcs when a particular hostname matches.
+  //  Note that some debug actions must be specified as a query option while others must
+  //  be passed in with the startup flag.
   //  Available global actions:
   //  - SLEEP@<ms> will sleep for the 'ms' milliseconds.
   //  - JITTER@<ms>[@<probability>] will sleep for a random amount of time between 0
   //    and 'ms' milliseconds with the given probability. If <probability> is omitted,
   //    it is 1.0.
-  //  - FAIL[@<probability>] returns an INTERNAL_ERROR status with the given
-  //    probability. If <probability> is omitted, it is 1.0.
+  //  - FAIL[@<probability>][@<error>] returns an INTERNAL_ERROR status with the given
+  //    probability and error. If <probability> is omitted, it is 1.0. If 'error' is
+  //    omitted, a generic error of the form: 'Debug Action: <label>:<action>' is used.
+  //    The code executing the debug action may respond to different error messages by
+  //    exercising different error paths.
+  // Examples:
+  // - CRS_BEFORE_ADMISSION:SLEEP@1000
+  //   Causes a 1 second sleep before queries are submitted to the admission controller.
+  // - IMPALA_SERVICE_POOL:127.0.0.1:27002:TransmitData:FAIL@0.1@REJECT_TOO_BUSY
+  //   Causes TransmitData rpcs to the third minicluster impalad to fail 10% of the time
+  //   with a "service too busy" error.
   //
   // Only a single ExecNode action is allowed, but multiple global actions can be
   // specified. To specify multiple actions, separate them with "|".
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index fc883a8..de1a8e1 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -2618,5 +2618,15 @@
     "units": "NONE",
     "kind": "COUNTER",
     "key": "impala.webserver.total-cookie-auth-failure"
+  },
+  {
+    "description": "The number of times the FAIL debug action returned an error. For testing only.",
+    "contexts": [
+      "IMPALAD"
+    ],
+    "label": "FAIL debug action hits",
+    "units": "NONE",
+    "kind": "COUNTER",
+    "key": "impala.debug_action.fail"
   }
 ]
diff --git a/tests/common/impala_cluster.py b/tests/common/impala_cluster.py
index 694c631..bbed25e 100644
--- a/tests/common/impala_cluster.py
+++ b/tests/common/impala_cluster.py
@@ -457,7 +457,7 @@
     super(ImpaladProcess, self).__init__(cmd, container_id, port_map)
     self.service = ImpaladService(self.hostname, self.webserver_interface,
         self.get_webserver_port(), self.__get_beeswax_port(), self.__get_be_port(),
-        self.__get_hs2_port(), self.__get_hs2_http_port(),
+        self.__get_krpc_port(), self.__get_hs2_port(), self.__get_hs2_http_port(),
         self._get_webserver_certificate_file())
 
   def _get_default_webserver_port(self):
@@ -469,6 +469,9 @@
   def __get_be_port(self):
     return int(self._get_port('be_port', DEFAULT_BE_PORT))
 
+  def __get_krpc_port(self):
+    return int(self._get_port('krpc_port', DEFAULT_KRPC_PORT))
+
   def __get_hs2_port(self):
     return int(self._get_port('hs2_port', DEFAULT_HS2_PORT))
 
diff --git a/tests/common/impala_service.py b/tests/common/impala_service.py
index fb6beb2..9cb7f50 100644
--- a/tests/common/impala_service.py
+++ b/tests/common/impala_service.py
@@ -162,12 +162,13 @@
 # new connections or accessing the debug webpage.
 class ImpaladService(BaseImpalaService):
   def __init__(self, hostname, webserver_interface="", webserver_port=25000,
-      beeswax_port=21000, be_port=22000, hs2_port=21050, hs2_http_port=28000,
-      webserver_certificate_file=""):
+      beeswax_port=21000, be_port=22000, krpc_port=27000, hs2_port=21050,
+      hs2_http_port=28000, webserver_certificate_file=""):
     super(ImpaladService, self).__init__(
         hostname, webserver_interface, webserver_port, webserver_certificate_file)
     self.beeswax_port = beeswax_port
     self.be_port = be_port
+    self.krpc_port = krpc_port
     self.hs2_port = hs2_port
     self.hs2_http_port = hs2_http_port
 
diff --git a/tests/custom_cluster/test_rpc_exception.py b/tests/custom_cluster/test_rpc_exception.py
index fc60c1a..07310b4 100644
--- a/tests/custom_cluster/test_rpc_exception.py
+++ b/tests/custom_cluster/test_rpc_exception.py
@@ -24,6 +24,16 @@
 class TestRPCException(CustomClusterTestSuite):
   """Tests Impala exception handling in TransmitData() RPC to make sure no
      duplicated row batches are sent. """
+  # DataStreamService rpc names
+  TRANSMIT_DATA_RPC = "TransmitData"
+  END_DATA_STREAM_RPC = "EndDataStream"
+
+  # Error to specify for ImpalaServicePool to reject rpcs with a 'server too busy' error.
+  REJECT_TOO_BUSY_MSG = "REJECT_TOO_BUSY"
+
+  # The BE krpc port of the impalad these tests simulate rpc errors at.
+  KRPC_PORT = 27002
+
   # This query ends up calling TransmitData() more than 2048 times to ensure
   # proper test coverage.
   TEST_QUERY = "select count(*) from tpch_parquet.lineitem t1, tpch_parquet.lineitem t2 \
@@ -40,55 +50,74 @@
       pytest.skip('runs only in exhaustive')
     super(TestRPCException, cls).setup_class()
 
-  # Execute TEST_QUERY. If 'exception_string' is None, it's expected to complete
-  # sucessfully with result matching EXPECTED_RESULT. Otherwise, it's expected
-  # to fail with 'exception_string'.
+  def _get_num_fails(self, impalad):
+    num_fails = impalad.service.get_metric_value("impala.debug_action.fail")
+    if num_fails is None:
+      return 0
+    return num_fails
+
+  # Execute TEST_QUERY repeatedly until the FAIL debug action has been hit. If
+  # 'exception_string' is None, it's expected to always complete sucessfully with result
+  # matching EXPECTED_RESULT. Otherwise, it's expected to fail with 'exception_string' if
+  # the debug action has been hit.
   def execute_test_query(self, exception_string):
-    try:
-      result = self.client.execute(self.TEST_QUERY)
-      assert result.data == self.EXPECTED_RESULT
-      assert not exception_string
-    except ImpalaBeeswaxException as e:
-      if exception_string is None:
-        raise e
-      assert exception_string in str(e)
+    impalad = self.cluster.impalads[2]
+    assert impalad.service.krpc_port == self.KRPC_PORT
+    # Re-run the query until the metrics show that we hit the debug action or we've run 10
+    # times. Each test in this file has at least a 50% chance of hitting the action per
+    # run, so there's at most a (1/2)^10 chance that this loop will fail spuriously.
+    i = 0
+    while self._get_num_fails(impalad) == 0 and i < 10:
+      i += 1
+      try:
+        result = self.client.execute(self.TEST_QUERY)
+        assert result.data == self.EXPECTED_RESULT
+        assert not exception_string or self._get_num_fails(impalad) == 0
+      except ImpalaBeeswaxException as e:
+        if exception_string is None:
+          raise e
+        assert exception_string in str(e)
+    assert self._get_num_fails(impalad) > 0
+
+  def _get_fail_action(rpc, error=None, port=KRPC_PORT, p=0.1):
+    """Returns a debug action that causes rpcs with the name 'rpc' that are sent to the
+    impalad at 'port' to FAIL with probability 'p' and return 'error' if specified."""
+    debug_action = "IMPALA_SERVICE_POOL:127.0.0.1:{port}:{rpc}:FAIL@{probability}" \
+        .format(rpc=rpc, probability=p, port=port)
+    if error is not None:
+      debug_action += "@" + error
+    return debug_action
 
   @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=1")
-  def test_rpc_send_closed_connection(self, vector):
+  @CustomClusterTestSuite.with_args("--debug_actions=" +
+      _get_fail_action(rpc=TRANSMIT_DATA_RPC, error=REJECT_TOO_BUSY_MSG))
+  def test_transmit_data_retry(self):
+    """Run a query where TransmitData may fail with a "server too busy" error. We should
+    always retry in this case, so the query should always eventually succeed."""
     self.execute_test_query(None)
 
   @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=2")
-  def test_rpc_send_stale_connection(self, vector):
+  @CustomClusterTestSuite.with_args("--debug_actions=" +
+      _get_fail_action(rpc=TRANSMIT_DATA_RPC, error=REJECT_TOO_BUSY_MSG) +
+      "|" + _get_fail_action(rpc=TRANSMIT_DATA_RPC))
+  def test_transmit_data_error(self):
+    """Run a query where TransmitData may fail with a "server too busy" or with a generic
+    error. The query should either succeed or fail with the given error."""
+    self.execute_test_query("Debug Action: IMPALA_SERVICE_POOL:FAIL@0.1")
+
+  @pytest.mark.execute_serially
+  @CustomClusterTestSuite.with_args("--debug_actions=" +
+      _get_fail_action(rpc=END_DATA_STREAM_RPC, error=REJECT_TOO_BUSY_MSG, p=0.5))
+  def test_end_data_stream_retry(self):
+    """Run a query where EndDataStream may fail with a "server too busy" error. We should
+    always retry in this case, so the query should always eventually succeed."""
     self.execute_test_query(None)
 
   @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=3")
-  def test_rpc_send_timed_out(self, vector):
-    self.execute_test_query(None)
-
-  @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=5")
-  def test_rpc_recv_timed_out(self, vector):
-    self.execute_test_query(None)
-
-  @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=6")
-  def test_rpc_secure_send_closed_connection(self, vector):
-    self.execute_test_query(None)
-
-  @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=7")
-  def test_rpc_secure_send_stale_connection(self, vector):
-    self.execute_test_query(None)
-
-  @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=8")
-  def test_rpc_secure_send_timed_out(self, vector):
-    self.execute_test_query(None)
-
-  @pytest.mark.execute_serially
-  @CustomClusterTestSuite.with_args("--fault_injection_rpc_exception_type=10")
-  def test_rpc_secure_recv_timed_out(self, vector):
-    self.execute_test_query(None)
+  @CustomClusterTestSuite.with_args("--debug_actions=" +
+      _get_fail_action(rpc=END_DATA_STREAM_RPC, error=REJECT_TOO_BUSY_MSG, p=0.5) +
+      "|" + _get_fail_action(rpc=END_DATA_STREAM_RPC, p=0.5))
+  def test_end_data_stream_error(self):
+    """Run a query where EndDataStream may fail with a "server too busy" or with a generic
+    error. The query should either succeed or fail with the given error."""
+    self.execute_test_query("Debug Action: IMPALA_SERVICE_POOL:FAIL@0.5")