You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by br...@apache.org on 2018/09/10 20:32:34 UTC
hive git commit: HIVE-20506 - HOS times out when cluster is full
while Hive-on-MR waits
Repository: hive
Updated Branches:
refs/heads/master afb61aebf -> 346c0ce44
HIVE-20506 - HOS times out when cluster is full while Hive-on-MR waits
(Brock Noland reviewed by Sahil Takiar)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/346c0ce4
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/346c0ce4
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/346c0ce4
Branch: refs/heads/master
Commit: 346c0ce44d89ab030400e017f3e8ec6000344a6e
Parents: afb61ae
Author: Brock Noland <br...@phdata.io>
Authored: Mon Sep 10 15:31:22 2018 -0500
Committer: Brock Noland <br...@phdata.io>
Committed: Mon Sep 10 15:31:57 2018 -0500
----------------------------------------------------------------------
.../spark/session/SparkSessionManagerImpl.java | 2 +-
.../hive/spark/client/SparkClientFactory.java | 4 +-
.../spark/client/SparkSubmitSparkClient.java | 27 ++++-
.../apache/hive/spark/client/rpc/RpcServer.java | 116 ++++++++++++++++++-
.../hive/spark/client/TestSparkClient.java | 2 +-
.../apache/hive/spark/client/rpc/TestRpc.java | 85 ++++++++++----
6 files changed, 207 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/346c0ce4/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/session/SparkSessionManagerImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/session/SparkSessionManagerImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/session/SparkSessionManagerImpl.java
index 79a56bd..8dae54d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/session/SparkSessionManagerImpl.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/spark/session/SparkSessionManagerImpl.java
@@ -96,7 +96,7 @@ public class SparkSessionManagerImpl implements SparkSessionManager {
startTimeoutThread();
Map<String, String> sparkConf = HiveSparkClientFactory.initiateSparkConf(hiveConf, null);
try {
- SparkClientFactory.initialize(sparkConf);
+ SparkClientFactory.initialize(sparkConf, hiveConf);
inited = true;
} catch (IOException e) {
throw new HiveException("Error initializing SparkClientFactory", e);
http://git-wip-us.apache.org/repos/asf/hive/blob/346c0ce4/spark-client/src/main/java/org/apache/hive/spark/client/SparkClientFactory.java
----------------------------------------------------------------------
diff --git a/spark-client/src/main/java/org/apache/hive/spark/client/SparkClientFactory.java b/spark-client/src/main/java/org/apache/hive/spark/client/SparkClientFactory.java
index 1974e88..54ecdf0 100644
--- a/spark-client/src/main/java/org/apache/hive/spark/client/SparkClientFactory.java
+++ b/spark-client/src/main/java/org/apache/hive/spark/client/SparkClientFactory.java
@@ -47,12 +47,12 @@ public final class SparkClientFactory {
*
* @param conf Map containing configuration parameters for the client library.
*/
- public static void initialize(Map<String, String> conf) throws IOException {
+ public static void initialize(Map<String, String> conf, HiveConf hiveConf) throws IOException {
if (server == null) {
synchronized (serverLock) {
if (server == null) {
try {
- server = new RpcServer(conf);
+ server = new RpcServer(conf, hiveConf);
} catch (InterruptedException ie) {
throw Throwables.propagate(ie);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/346c0ce4/spark-client/src/main/java/org/apache/hive/spark/client/SparkSubmitSparkClient.java
----------------------------------------------------------------------
diff --git a/spark-client/src/main/java/org/apache/hive/spark/client/SparkSubmitSparkClient.java b/spark-client/src/main/java/org/apache/hive/spark/client/SparkSubmitSparkClient.java
index 7a6e77b..1879829 100644
--- a/spark-client/src/main/java/org/apache/hive/spark/client/SparkSubmitSparkClient.java
+++ b/spark-client/src/main/java/org/apache/hive/spark/client/SparkSubmitSparkClient.java
@@ -31,6 +31,8 @@ import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
+import java.util.regex.Pattern;
+import java.util.regex.Matcher;
import org.apache.commons.lang3.StringUtils;
@@ -51,6 +53,7 @@ class SparkSubmitSparkClient extends AbstractSparkClient {
private static final Logger LOG = LoggerFactory.getLogger(SparkSubmitSparkClient.class);
+ private static final Pattern YARN_APPLICATION_ID_REGEX = Pattern.compile("\\s(application_[0-9]+_[0-9]+)(\\s|$)");
private static final String SPARK_HOME_ENV = "SPARK_HOME";
private static final String SPARK_HOME_KEY = "spark.home";
@@ -191,17 +194,37 @@ class SparkSubmitSparkClient extends AbstractSparkClient {
final Process child = pb.start();
String threadName = Thread.currentThread().getName();
final List<String> childErrorLog = Collections.synchronizedList(new ArrayList<String>());
+ final List<String> childOutLog = Collections.synchronizedList(new ArrayList<String>());
final LogRedirector.LogSourceCallback callback = () -> isAlive;
LogRedirector.redirect("spark-submit-stdout-redir-" + threadName,
- new LogRedirector(child.getInputStream(), LOG, callback));
+ new LogRedirector(child.getInputStream(), LOG, childOutLog, callback));
LogRedirector.redirect("spark-submit-stderr-redir-" + threadName,
new LogRedirector(child.getErrorStream(), LOG, childErrorLog, callback));
runnable = () -> {
try {
int exitCode = child.waitFor();
- if (exitCode != 0) {
+ if (exitCode == 0) {
+ synchronized (childOutLog) {
+ for (String line : childOutLog) {
+ Matcher m = YARN_APPLICATION_ID_REGEX.matcher(line);
+ if (m.find()) {
+ LOG.info("Found application id " + m.group(1));
+ rpcServer.setApplicationId(m.group(1));
+ }
+ }
+ }
+ synchronized (childErrorLog) {
+ for (String line : childErrorLog) {
+ Matcher m = YARN_APPLICATION_ID_REGEX.matcher(line);
+ if (m.find()) {
+ LOG.info("Found application id " + m.group(1));
+ rpcServer.setApplicationId(m.group(1));
+ }
+ }
+ }
+ } else {
List<String> errorMessages = new ArrayList<>();
synchronized (childErrorLog) {
for (String line : childErrorLog) {
http://git-wip-us.apache.org/repos/asf/hive/blob/346c0ce4/spark-client/src/main/java/org/apache/hive/spark/client/rpc/RpcServer.java
----------------------------------------------------------------------
diff --git a/spark-client/src/main/java/org/apache/hive/spark/client/rpc/RpcServer.java b/spark-client/src/main/java/org/apache/hive/spark/client/rpc/RpcServer.java
index 0c67ffd..1f6114a 100644
--- a/spark-client/src/main/java/org/apache/hive/spark/client/rpc/RpcServer.java
+++ b/spark-client/src/main/java/org/apache/hive/spark/client/rpc/RpcServer.java
@@ -61,6 +61,11 @@ import io.netty.util.concurrent.ScheduledFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.hive.common.classification.InterfaceAudience;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+import org.apache.hadoop.yarn.client.api.YarnClient;
/**
* An RPC server. The server matches remote clients based on a secret that is generated on
@@ -79,9 +84,12 @@ public class RpcServer implements Closeable {
private final int port;
private final ConcurrentMap<String, ClientInfo> pendingClients;
private final RpcConfiguration config;
+ private String applicationId;
+ private final HiveConf hiveConf;
- public RpcServer(Map<String, String> mapConf) throws IOException, InterruptedException {
+ public RpcServer(Map<String, String> mapConf, HiveConf hiveConf) throws IOException, InterruptedException {
this.config = new RpcConfiguration(mapConf);
+ this.hiveConf = hiveConf;
this.group = new NioEventLoopGroup(
this.config.getRpcThreadCount(),
new ThreadFactoryBuilder()
@@ -166,14 +174,116 @@ public class RpcServer implements Closeable {
return registerClient(clientId, secret, serverDispatcher, config.getServerConnectTimeoutMs());
}
+ public void setApplicationId(String applicationId) {
+ this.applicationId = applicationId;
+ }
+
+ /**
+ * This function converts an application in form of a String into a ApplicationId.
+ *
+ * @param appIDStr The application id in form of a string
+ * @return the application id as an instance of ApplicationId class.
+ */
+ private static ApplicationId getApplicationIDFromString(String appIDStr) {
+ String[] parts = appIDStr.split("_");
+ if (parts.length < 3) {
+ throw new IllegalStateException("the application id found is not valid. application id: " + appIDStr);
+ }
+ long timestamp = Long.parseLong(parts[1]);
+ int id = Integer.parseInt(parts[2]);
+ return ApplicationId.newInstance(timestamp, id);
+ }
+
+ public static boolean isApplicationAccepted(HiveConf conf, String applicationId) {
+ if (applicationId == null) {
+ return false;
+ }
+ YarnClient yarnClient = null;
+ try {
+ ApplicationId appId = getApplicationIDFromString(applicationId);
+ yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ ApplicationReport appReport = yarnClient.getApplicationReport(appId);
+ return appReport != null && appReport.getYarnApplicationState() == YarnApplicationState.ACCEPTED;
+ } catch (Exception ex) {
+ LOG.error("Failed getting application status for: " + applicationId + ": " + ex, ex);
+ return false;
+ } finally {
+ if (yarnClient != null) {
+ try {
+ yarnClient.stop();
+ } catch (Exception ex) {
+ LOG.error("Failed to stop yarn client: " + ex, ex);
+ }
+ }
+ }
+ }
+
+ static class YarnApplicationStateFinder {
+ public boolean isApplicationAccepted(HiveConf conf, String applicationId) {
+ if (applicationId == null) {
+ return false;
+ }
+ YarnClient yarnClient = null;
+ try {
+ LOG.info("Trying to find " + applicationId);
+ ApplicationId appId = getApplicationIDFromString(applicationId);
+ yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ ApplicationReport appReport = yarnClient.getApplicationReport(appId);
+ return appReport != null && appReport.getYarnApplicationState() == YarnApplicationState.ACCEPTED;
+ } catch (Exception ex) {
+ LOG.error("Failed getting application status for: " + applicationId + ": " + ex, ex);
+ return false;
+ } finally {
+ if (yarnClient != null) {
+ try {
+ yarnClient.stop();
+ } catch (Exception ex) {
+ LOG.error("Failed to stop yarn client: " + ex, ex);
+ }
+ }
+ }
+ }
+ }
+
@VisibleForTesting
Future<Rpc> registerClient(final String clientId, String secret,
- RpcDispatcher serverDispatcher, long clientTimeoutMs) {
+ RpcDispatcher serverDispatcher, final long clientTimeoutMs) {
+ return registerClient(clientId, secret, serverDispatcher, clientTimeoutMs, new YarnApplicationStateFinder());
+ }
+
+ @VisibleForTesting
+ Future<Rpc> registerClient(final String clientId, String secret,
+ RpcDispatcher serverDispatcher, long clientTimeoutMs,
+ YarnApplicationStateFinder yarnApplicationStateFinder) {
final Promise<Rpc> promise = group.next().newPromise();
Runnable timeout = new Runnable() {
@Override
public void run() {
+ // check to see if application is in ACCEPTED state, if so, don't set failure
+ // if applicationId is not null
+ // do yarn application -status $applicationId
+ // if state == ACCEPTED
+ // reschedule timeout runnable
+ // else
+ // set failure as below
+ LOG.info("Trying to find " + applicationId);
+ if (yarnApplicationStateFinder.isApplicationAccepted(hiveConf, applicationId)) {
+ final ClientInfo client = pendingClients.get(clientId);
+ if (client != null) {
+ LOG.info("Extending timeout for client " + clientId);
+ ScheduledFuture<?> oldTimeoutFuture = client.timeoutFuture;
+ client.timeoutFuture = group.schedule(this,
+ clientTimeoutMs,
+ TimeUnit.MILLISECONDS);
+ oldTimeoutFuture.cancel(true);
+ return;
+ }
+ }
promise.setFailure(new TimeoutException(
String.format("Client '%s' timed out waiting for connection from the Remote Spark" +
" Driver", clientId)));
@@ -369,7 +479,7 @@ public class RpcServer implements Closeable {
final Promise<Rpc> promise;
final String secret;
final RpcDispatcher dispatcher;
- final ScheduledFuture<?> timeoutFuture;
+ ScheduledFuture<?> timeoutFuture;
private ClientInfo(String id, Promise<Rpc> promise, String secret, RpcDispatcher dispatcher,
ScheduledFuture<?> timeoutFuture) {
http://git-wip-us.apache.org/repos/asf/hive/blob/346c0ce4/spark-client/src/test/java/org/apache/hive/spark/client/TestSparkClient.java
----------------------------------------------------------------------
diff --git a/spark-client/src/test/java/org/apache/hive/spark/client/TestSparkClient.java b/spark-client/src/test/java/org/apache/hive/spark/client/TestSparkClient.java
index d738003..996b24e 100644
--- a/spark-client/src/test/java/org/apache/hive/spark/client/TestSparkClient.java
+++ b/spark-client/src/test/java/org/apache/hive/spark/client/TestSparkClient.java
@@ -339,7 +339,7 @@ public class TestSparkClient {
private void runTest(TestFunction test) throws Exception {
Map<String, String> conf = createConf();
- SparkClientFactory.initialize(conf);
+ SparkClientFactory.initialize(conf, HIVECONF);
SparkClient client = null;
try {
test.config(conf);
http://git-wip-us.apache.org/repos/asf/hive/blob/346c0ce4/spark-client/src/test/java/org/apache/hive/spark/client/rpc/TestRpc.java
----------------------------------------------------------------------
diff --git a/spark-client/src/test/java/org/apache/hive/spark/client/rpc/TestRpc.java b/spark-client/src/test/java/org/apache/hive/spark/client/rpc/TestRpc.java
index 013bcff..7da33fe 100644
--- a/spark-client/src/test/java/org/apache/hive/spark/client/rpc/TestRpc.java
+++ b/spark-client/src/test/java/org/apache/hive/spark/client/rpc/TestRpc.java
@@ -19,7 +19,6 @@ package org.apache.hive.spark.client.rpc;
import java.io.Closeable;
import java.net.InetAddress;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -64,10 +63,12 @@ public class TestRpc {
private static final Map<String, String> emptyConfig =
ImmutableMap.of(HiveConf.ConfVars.SPARK_RPC_CHANNEL_LOG_LEVEL.varname, "DEBUG");
private static final int RETRY_ACQUIRE_PORT_COUNT = 10;
+ private HiveConf hiveConf;
@Before
public void setUp() {
closeables = Lists.newArrayList();
+ hiveConf = new HiveConf();
}
@After
@@ -99,7 +100,7 @@ public class TestRpc {
@Test
public void testClientServer() throws Exception {
- RpcServer server = autoClose(new RpcServer(emptyConfig));
+ RpcServer server = autoClose(new RpcServer(emptyConfig, hiveConf));
Rpc[] rpcs = createRpcConnection(server);
Rpc serverRpc = rpcs[0];
Rpc client = rpcs[1];
@@ -136,25 +137,25 @@ public class TestRpc {
// Test if rpc_server_address is configured
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_ADDRESS.varname, hostAddress);
- RpcServer server1 = autoClose(new RpcServer(config));
+ RpcServer server1 = autoClose(new RpcServer(config, hiveConf));
assertTrue("Host address should match the expected one", server1.getAddress() == hostAddress);
// Test if rpc_server_address is not configured but HS2 server host is configured
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_ADDRESS.varname, "");
config.put(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_BIND_HOST.varname, hostAddress);
- RpcServer server2 = autoClose(new RpcServer(config));
+ RpcServer server2 = autoClose(new RpcServer(config, hiveConf));
assertTrue("Host address should match the expected one", server2.getAddress() == hostAddress);
// Test if both are not configured
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_ADDRESS.varname, "");
config.put(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_BIND_HOST.varname, "");
- RpcServer server3 = autoClose(new RpcServer(config));
+ RpcServer server3 = autoClose(new RpcServer(config, hiveConf));
assertTrue("Host address should match the expected one", server3.getAddress() == InetAddress.getLocalHost().getHostName());
}
@Test
public void testBadHello() throws Exception {
- RpcServer server = autoClose(new RpcServer(emptyConfig));
+ RpcServer server = autoClose(new RpcServer(emptyConfig, hiveConf));
Future<Rpc> serverRpcFuture = server.registerClient("client", "newClient",
new TestDispatcher());
@@ -170,7 +171,7 @@ public class TestRpc {
// On failure, the SASL handler will throw an exception indicating that the SASL
// negotiation failed.
assertTrue("Unexpected exception: " + ee.getCause(),
- ee.getCause() instanceof SaslException);
+ ee.getCause() instanceof SaslException);
}
serverRpcFuture.cancel(true);
@@ -180,12 +181,12 @@ public class TestRpc {
public void testServerPort() throws Exception {
Map<String, String> config = new HashMap<String, String>();
- RpcServer server0 = new RpcServer(config);
+ RpcServer server0 = new RpcServer(config, hiveConf);
assertTrue("Empty port range should return a random valid port: " + server0.getPort(), server0.getPort() >= 0);
IOUtils.closeQuietly(server0);
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_PORT.varname, "49152-49222,49223,49224-49333");
- RpcServer server1 = new RpcServer(config);
+ RpcServer server1 = new RpcServer(config, hiveConf);
assertTrue("Port should be within configured port range:" + server1.getPort(), server1.getPort() >= 49152 && server1.getPort() <= 49333);
IOUtils.closeQuietly(server1);
@@ -194,7 +195,7 @@ public class TestRpc {
for (int i = 0; i < RETRY_ACQUIRE_PORT_COUNT; i++) {
try {
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_PORT.varname, String.valueOf(expectedPort));
- server2 = new RpcServer(config);
+ server2 = new RpcServer(config, hiveConf);
break;
} catch (Exception e) {
LOG.debug("Error while connecting to port " + expectedPort + " retrying: " + e.getMessage());
@@ -208,7 +209,7 @@ public class TestRpc {
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_PORT.varname, "49552-49222,49223,49224-49333");
try {
- autoClose(new RpcServer(config));
+ autoClose(new RpcServer(config, hiveConf));
assertTrue("Invalid port range should throw an exception", false); // Should not reach here
} catch(IllegalArgumentException e) {
assertEquals(
@@ -222,7 +223,7 @@ public class TestRpc {
for (int i = 0; i < RETRY_ACQUIRE_PORT_COUNT; i++) {
try {
config.put(HiveConf.ConfVars.SPARK_RPC_SERVER_PORT.varname, String.valueOf(expectedPort) + ",21-23");
- server3 = new RpcServer(config);
+ server3 = new RpcServer(config, hiveConf);
break;
} catch (Exception e) {
LOG.debug("Error while connecting to port " + expectedPort + " retrying");
@@ -236,7 +237,7 @@ public class TestRpc {
@Test
public void testCloseListener() throws Exception {
- RpcServer server = autoClose(new RpcServer(emptyConfig));
+ RpcServer server = autoClose(new RpcServer(emptyConfig, hiveConf));
Rpc[] rpcs = createRpcConnection(server);
Rpc client = rpcs[1];
@@ -255,7 +256,7 @@ public class TestRpc {
@Test
public void testNotDeserializableRpc() throws Exception {
- RpcServer server = autoClose(new RpcServer(emptyConfig));
+ RpcServer server = autoClose(new RpcServer(emptyConfig, hiveConf));
Rpc[] rpcs = createRpcConnection(server);
Rpc client = rpcs[1];
@@ -270,10 +271,10 @@ public class TestRpc {
@Test
public void testEncryption() throws Exception {
Map<String, String> eConf = ImmutableMap.<String,String>builder()
- .putAll(emptyConfig)
- .put(RpcConfiguration.RPC_SASL_OPT_PREFIX + "qop", Rpc.SASL_AUTH_CONF)
- .build();
- RpcServer server = autoClose(new RpcServer(eConf));
+ .putAll(emptyConfig)
+ .put(RpcConfiguration.RPC_SASL_OPT_PREFIX + "qop", Rpc.SASL_AUTH_CONF)
+ .build();
+ RpcServer server = autoClose(new RpcServer(eConf, hiveConf));
Rpc[] rpcs = createRpcConnection(server, eConf, null);
Rpc client = rpcs[1];
@@ -288,7 +289,7 @@ public class TestRpc {
Map<String, String> conf = ImmutableMap.<String,String>builder()
.putAll(emptyConfig)
.build();
- RpcServer server = autoClose(new RpcServer(conf));
+ RpcServer server = autoClose(new RpcServer(conf, hiveConf));
String secret = server.createSecret();
try {
@@ -310,9 +311,53 @@ public class TestRpc {
}
}
+ static class MockYarnApplicationStateFinder extends RpcServer.YarnApplicationStateFinder {
+ private int count = 0;
+ public boolean isApplicationAccepted(HiveConf conf, String applicationId) {
+ return count++ < 10;
+ }
+ }
+
+
+ /**
+ * Tests that we don't timeout with a short timeout but the spark application isn't running.
+ */
+ @Test
+ public void testExtendClientTimeout() throws Exception {
+ Map<String, String> conf = ImmutableMap.<String, String>builder()
+ .putAll(emptyConfig)
+ .build();
+ RpcServer server = autoClose(new RpcServer(conf, hiveConf));
+ String secret = server.createSecret();
+ MockYarnApplicationStateFinder yarnApplicationStateFinder = new MockYarnApplicationStateFinder();
+ Future<Rpc> promise = server.registerClient("client", secret, new TestDispatcher(), 2L,
+ yarnApplicationStateFinder);
+ assertFalse(promise.isDone());
+ Thread.sleep(50);
+ try {
+ promise.get();
+ fail("Server should have timed out client.");
+ } catch (ExecutionException ee) {
+ assertTrue(ee.getCause() instanceof TimeoutException);
+ }
+
+ NioEventLoopGroup eloop = new NioEventLoopGroup();
+ Future<Rpc> clientRpcFuture = Rpc.createClient(conf, eloop,
+ "localhost", server.getPort(), "client", secret, new TestDispatcher());
+ try {
+ autoClose(clientRpcFuture.get());
+ fail("Client should have failed to connect to server.");
+ } catch (ExecutionException ee) {
+ // Error should not be a timeout.
+ assertFalse(ee.getCause() instanceof TimeoutException);
+ }
+ }
+
+
+
@Test
public void testRpcServerMultiThread() throws Exception {
- final RpcServer server = autoClose(new RpcServer(emptyConfig));
+ final RpcServer server = autoClose(new RpcServer(emptyConfig, hiveConf));
final String msg = "Hello World!";
Callable<String> callable = () -> {
Rpc[] rpcs = createRpcConnection(server, emptyConfig, UUID.randomUUID().toString());