This is an automated email from the ASF dual-hosted git repository.
zhangbutao pushed a commit to branch branch-2.3
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/branch-2.3 by this push:
new 36b32612ba0 HIVE-27172: (2.3) Add the HMS client connection timeout
config (#5115)
36b32612ba0 is described below
commit 36b32612ba0a20d3d305c77a813aca62d7129ed0
Author: Cheng Pan <[email protected]>
AuthorDate: Wed Mar 6 09:06:11 2024 +0800
HIVE-27172: (2.3) Add the HMS client connection timeout config (#5115)
---
.../hadoop/hive/common/auth/HiveAuthUtils.java | 8 ++-
.../java/org/apache/hadoop/hive/conf/HiveConf.java | 3 +
.../hadoop/hive/metastore/HiveMetaStoreClient.java | 8 ++-
.../hive/metastore/TestHiveMetaStoreTimeout.java | 73 ++++++++++++++++------
4 files changed, 69 insertions(+), 23 deletions(-)
diff --git
a/common/src/java/org/apache/hadoop/hive/common/auth/HiveAuthUtils.java
b/common/src/java/org/apache/hadoop/hive/common/auth/HiveAuthUtils.java
index 632980e7cd8..7fb2bd3ed97 100644
--- a/common/src/java/org/apache/hadoop/hive/common/auth/HiveAuthUtils.java
+++ b/common/src/java/org/apache/hadoop/hive/common/auth/HiveAuthUtils.java
@@ -55,6 +55,11 @@ public class HiveAuthUtils {
}
public static TTransport getSSLSocket(String host, int port, int
loginTimeout,
+ String trustStorePath, String trustStorePassWord) throws
TTransportException {
+ return getSSLSocket(host, port, loginTimeout, loginTimeout,
trustStorePath, trustStorePassWord);
+ }
+
+ public static TTransport getSSLSocket(String host, int port, int
socketTimeout, int connectionTimeout,
String trustStorePath, String trustStorePassWord) throws
TTransportException {
TSSLTransportFactory.TSSLTransportParameters params =
new TSSLTransportFactory.TSSLTransportParameters();
@@ -62,7 +67,8 @@ public class HiveAuthUtils {
params.requireClientAuth(true);
// The underlying SSLSocket object is bound to host:port with the given
SO_TIMEOUT and
// SSLContext created with the given params
- TSocket tSSLSocket = TSSLTransportFactory.getClientSocket(host, port,
loginTimeout, params);
+ TSocket tSSLSocket = TSSLTransportFactory.getClientSocket(host, port,
socketTimeout, params);
+ tSSLSocket.setConnectTimeout(connectionTimeout);
return getSSLSocketWithHttps(tSSLSocket);
}
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index 46a74305eed..b8e54fdd39f 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -665,6 +665,9 @@ public class HiveConf extends Configuration {
"MetaStore Client socket lifetime in seconds. After this time is
exceeded, client\n" +
"reconnects on the next MetaStore operation. A value of 0s means the
connection\n" +
"has an infinite lifetime."),
+
METASTORE_CLIENT_CONNECTION_TIMEOUT("hive.metastore.client.connection.timeout",
"600s",
+ new TimeValidator(TimeUnit.SECONDS),
+ "MetaStore Client connection timeout in seconds"),
METASTOREPWD("javax.jdo.option.ConnectionPassword", "mine",
"password to use against metastore database"),
METASTORECONNECTURLHOOK("hive.metastore.ds.connection.url.hook", "",
diff --git
a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
index c34124f410c..04ae5b0718d 100644
---
a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
+++
b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
@@ -400,6 +400,8 @@ public class HiveMetaStoreClient implements
IMetaStoreClient {
boolean useCompactProtocol =
conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_COMPACT_PROTOCOL);
int clientSocketTimeout = (int) conf.getTimeVar(
ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, TimeUnit.MILLISECONDS);
+ int connectionTimeout = (int) conf.getTimeVar(
+ ConfVars.METASTORE_CLIENT_CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS);
for (int attempt = 0; !isConnected && attempt < retries; ++attempt) {
for (URI store : metastoreUris) {
@@ -420,7 +422,7 @@ public class HiveMetaStoreClient implements
IMetaStoreClient {
String tokenSig =
conf.getVar(ConfVars.METASTORE_TOKEN_SIGNATURE);
// tokenSig could be null
tokenStrForm = Utils.getTokenStrForm(tokenSig);
- transport = new TSocket(new TConfiguration(), store.getHost(),
store.getPort(), clientSocketTimeout);
+ transport = new TSocket(new TConfiguration(), store.getHost(),
store.getPort(), clientSocketTimeout, connectionTimeout);
if(tokenStrForm != null) {
// authenticate using delegation tokens via the "DIGEST"
mechanism
@@ -451,7 +453,7 @@ public class HiveMetaStoreClient implements
IMetaStoreClient {
HiveConf.ConfVars.HIVE_METASTORE_SSL_TRUSTSTORE_PASSWORD.varname);
// Create an SSL socket and connect
- transport = HiveAuthUtils.getSSLSocket(store.getHost(),
store.getPort(), clientSocketTimeout, trustStorePath, trustStorePassword );
+ transport = HiveAuthUtils.getSSLSocket(store.getHost(),
store.getPort(), clientSocketTimeout, connectionTimeout, trustStorePath,
trustStorePassword );
LOG.info("Opened an SSL connection to metastore, current
connections: " + connCount.incrementAndGet());
} catch(IOException e) {
throw new IllegalArgumentException(e);
@@ -461,7 +463,7 @@ public class HiveMetaStoreClient implements
IMetaStoreClient {
}
} else {
try {
- transport = new TSocket(new TConfiguration(), store.getHost(),
store.getPort(), clientSocketTimeout);
+ transport = new TSocket(new TConfiguration(), store.getHost(),
store.getPort(), clientSocketTimeout, connectionTimeout);
} catch (TTransportException e) {
tte = e;
throw new MetaException(e.toString());
diff --git
a/metastore/src/test/org/apache/hadoop/hive/metastore/TestHiveMetaStoreTimeout.java
b/metastore/src/test/org/apache/hadoop/hive/metastore/TestHiveMetaStoreTimeout.java
index f8eed1820cc..da9cd7860aa 100644
---
a/metastore/src/test/org/apache/hadoop/hive/metastore/TestHiveMetaStoreTimeout.java
+++
b/metastore/src/test/org/apache/hadoop/hive/metastore/TestHiveMetaStoreTimeout.java
@@ -18,14 +18,20 @@
package org.apache.hadoop.hive.metastore;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.util.StringUtils;
+import org.apache.thrift.transport.TTransportException;
+import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
+
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -37,9 +43,10 @@ public class TestHiveMetaStoreTimeout {
protected static HiveMetaStoreClient client;
protected static HiveConf hiveConf;
protected static Warehouse warehouse;
+ protected static int port;
@BeforeClass
- public static void setUp() throws Exception {
+ public static void startMetaStoreServer() throws Exception {
HiveMetaStore.TEST_TIMEOUT_ENABLED = true;
hiveConf = new HiveConf(TestHiveMetaStoreTimeout.class);
hiveConf.setBoolean(HiveConf.ConfVars.HIVE_WAREHOUSE_SUBDIR_INHERIT_PERMS.varname,
true);
@@ -48,25 +55,25 @@ public class TestHiveMetaStoreTimeout {
hiveConf.setTimeVar(HiveConf.ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, 10
* 1000,
TimeUnit.MILLISECONDS);
warehouse = new Warehouse(hiveConf);
- try {
- client = new HiveMetaStoreClient(hiveConf);
- } catch (Throwable e) {
- System.err.println("Unable to open the metastore");
- System.err.println(StringUtils.stringifyException(e));
- throw e;
- }
+ port = MetaStoreUtils.startMetaStoreWithRetry(hiveConf);
+ hiveConf.setVar(HiveConf.ConfVars.METASTOREURIS, "thrift://localhost:" +
port);
+ hiveConf.setBoolVar(HiveConf.ConfVars.METASTORE_EXECUTE_SET_UGI, false);
}
@AfterClass
public static void tearDown() throws Exception {
HiveMetaStore.TEST_TIMEOUT_ENABLED = false;
- try {
- client.close();
- } catch (Throwable e) {
- System.err.println("Unable to close metastore");
- System.err.println(StringUtils.stringifyException(e));
- throw e;
- }
+ }
+
+ @Before
+ public void setup() throws MetaException {
+ client = new HiveMetaStoreClient(hiveConf);
+ }
+
+ @After
+ public void cleanup() {
+ client.close();
+ client = null;
}
@Test
@@ -99,9 +106,8 @@ public class TestHiveMetaStoreTimeout {
try {
client.createDatabase(db);
Assert.fail("should throw timeout exception.");
- } catch (MetaException e) {
- Assert.assertTrue("unexpected MetaException",
e.getMessage().contains("Timeout when " +
- "executing method: create_database"));
+ } catch (TTransportException e) {
+ Assert.assertTrue("unexpected Exception", e.getMessage().contains("Read
timed out"));
}
// restore
@@ -140,4 +146,33 @@ public class TestHiveMetaStoreTimeout {
client.dropDatabase(dbName, true, true);
client.setMetaConf(HiveConf.ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT.varname,
"10s");
}
+
+ @Test
+ public void testConnectionTimeout() throws Exception {
+ final HiveConf newConf = new HiveConf(hiveConf);
+ newConf.setTimeVar(HiveConf.ConfVars.METASTORE_CLIENT_CONNECTION_TIMEOUT,
1000,
+ TimeUnit.MILLISECONDS);
+ // fake host to mock connection time out
+ newConf.setVar(HiveConf.ConfVars.METASTOREURIS, "thrift://1.1.1.1:" +
port);
+ newConf.setIntVar(HiveConf.ConfVars.METASTORETHRIFTCONNECTIONRETRIES, 1);
+
+ Future<Void> future = Executors.newSingleThreadExecutor().submit(new
Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ HiveMetaStoreClient c = null;
+ try {
+ c = new HiveMetaStoreClient(newConf);
+ Assert.fail("should throw connection timeout exception.");
+ } catch (MetaException e) {
+ Assert.assertTrue("unexpected Exception",
e.getMessage().contains("connect timed out"));
+ } finally {
+ if (c != null) {
+ c.close();
+ }
+ }
+ return null;
+ }
+ });
+ future.get(5, TimeUnit.SECONDS);
+ }
}
\ No newline at end of file
