This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c293fc7af [core] Make ClientPool simple to reduce thread safe bugs
(#4256)
c293fc7af is described below
commit c293fc7afa90ac87b7742e2fb11a7a52843a75d5
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 26 15:56:33 2024 +0800
[core] Make ClientPool simple to reduce thread safe bugs (#4256)
---
.../java/org/apache/paimon/client/ClientPool.java | 183 ++++----------------
.../java/org/apache/paimon/jdbc/JdbcCatalog.java | 4 +-
.../org/apache/paimon/jdbc/JdbcClientPool.java | 34 ++--
.../apache/paimon/hive/pool/CachedClientPool.java | 12 --
.../apache/paimon/hive/pool/HiveClientPool.java | 47 +----
.../paimon/hive/pool/TestHiveClientPool.java | 190 ---------------------
6 files changed, 57 insertions(+), 413 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/client/ClientPool.java
b/paimon-common/src/main/java/org/apache/paimon/client/ClientPool.java
index b437e6893..c42a30b35 100644
--- a/paimon-common/src/main/java/org/apache/paimon/client/ClientPool.java
+++ b/paimon-common/src/main/java/org/apache/paimon/client/ClientPool.java
@@ -18,18 +18,12 @@
package org.apache.paimon.client;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.io.Closeable;
-import java.util.ArrayDeque;
-import java.util.Deque;
-
-import static org.apache.paimon.utils.Preconditions.checkState;
-
-/* This file is based on source code from the Iceberg Project
(http://iceberg.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
- * additional information regarding copyright ownership. */
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
/** Client pool for using multiple clients to execute actions. */
public interface ClientPool<C, E extends Exception> {
@@ -45,167 +39,60 @@ public interface ClientPool<C, E extends Exception> {
<R> R run(Action<R, C, E> action) throws E, InterruptedException;
- <R> R run(Action<R, C, E> action, boolean retry) throws E,
InterruptedException;
-
void execute(ExecuteAction<C, E> action) throws E, InterruptedException;
- void execute(ExecuteAction<C, E> action, boolean retry) throws E,
InterruptedException;
-
/** Default implementation for {@link ClientPool}. */
abstract class ClientPoolImpl<C, E extends Exception> implements
Closeable, ClientPool<C, E> {
- private static final Logger LOG =
LoggerFactory.getLogger(ClientPoolImpl.class);
- private final int poolSize;
- private final Deque<C> clients;
- private final Class<? extends E> reconnectExc;
- private final Object signal = new Object();
- private final boolean retryByDefault;
- private volatile int currentSize;
- private boolean closed;
+ private volatile LinkedBlockingDeque<C> clients;
- public ClientPoolImpl(
- int poolSize, Class<? extends E> reconnectExc, boolean
retryByDefault) {
- this.poolSize = poolSize;
- this.reconnectExc = reconnectExc;
- this.clients = new ArrayDeque<>(poolSize);
- this.currentSize = 0;
- this.closed = false;
- this.retryByDefault = retryByDefault;
+ protected ClientPoolImpl(int poolSize, Supplier<C> supplier) {
+ this.clients = new LinkedBlockingDeque<>();
+ for (int i = 0; i < poolSize; i++) {
+ this.clients.add(supplier.get());
+ }
}
@Override
public <R> R run(Action<R, C, E> action) throws E,
InterruptedException {
- return run(action, retryByDefault);
- }
-
- @Override
- public <R> R run(Action<R, C, E> action, boolean retry) throws E,
InterruptedException {
- C client = get();
- try {
- return action.run(client);
- } catch (Exception exc) {
- if (retry && isConnectionException(exc)) {
- try {
- client = reconnect(client);
- } catch (Exception ignored) {
- // if reconnection throws any exception, rethrow the
original failure
- throw reconnectExc.cast(exc);
- }
-
+ while (true) {
+ LinkedBlockingDeque<C> clients = this.clients;
+ if (clients == null) {
+ throw new IllegalStateException("Cannot get a client from
a closed pool");
+ }
+ C client = clients.pollFirst(10, TimeUnit.SECONDS);
+ if (client == null) {
+ continue;
+ }
+ try {
return action.run(client);
+ } finally {
+ clients.addFirst(client);
}
-
- throw exc;
-
- } finally {
- release(client);
}
}
@Override
public void execute(ExecuteAction<C, E> action) throws E,
InterruptedException {
- execute(action, retryByDefault);
- }
-
- @Override
- public void execute(ExecuteAction<C, E> action, boolean retry)
- throws E, InterruptedException {
- C client = get();
- try {
- action.run(client);
- } catch (Exception exc) {
- if (retry && isConnectionException(exc)) {
- try {
- client = reconnect(client);
- } catch (Exception ignored) {
- // if reconnection throws any exception, rethrow the
original failure
- throw reconnectExc.cast(exc);
- }
-
- action.run(client);
- }
-
- throw exc;
-
- } finally {
- release(client);
- }
- }
-
- protected abstract C newClient();
-
- protected abstract C reconnect(C client);
-
- protected boolean isConnectionException(Exception exc) {
- return reconnectExc.isInstance(exc);
+ run(
+ (Action<Void, C, E>)
+ client -> {
+ action.run(client);
+ return null;
+ });
}
protected abstract void close(C client);
@Override
public void close() {
- this.closed = true;
- try {
- while (currentSize > 0) {
- if (!clients.isEmpty()) {
- synchronized (this) {
- if (!clients.isEmpty()) {
- C client = clients.removeFirst();
- close(client);
- currentSize -= 1;
- }
- }
- }
- if (clients.isEmpty() && currentSize > 0) {
- synchronized (signal) {
- // wake every second in case this missed the signal
- signal.wait(1000);
- }
- }
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- LOG.warn(
- "Interrupted while shutting down pool. Some clients
may not be closed.", e);
+ LinkedBlockingDeque<C> clients = this.clients;
+ this.clients = null;
+ if (clients != null) {
+ List<C> drain = new ArrayList<>();
+ clients.drainTo(drain);
+ drain.forEach(this::close);
}
}
-
- private C get() throws InterruptedException {
- checkState(!closed, "Cannot get a client from a closed pool");
- while (true) {
- if (!clients.isEmpty() || currentSize < poolSize) {
- synchronized (this) {
- if (!clients.isEmpty()) {
- return clients.removeFirst();
- } else if (currentSize < poolSize) {
- C client = newClient();
- currentSize += 1;
- return client;
- }
- }
- }
- synchronized (signal) {
- // wake every second in case this missed the signal
- signal.wait(1000);
- }
- }
- }
-
- private void release(C client) {
- synchronized (this) {
- clients.addFirst(client);
- }
- synchronized (signal) {
- signal.notify();
- }
- }
-
- public int poolSize() {
- return poolSize;
- }
-
- public boolean isClosed() {
- return closed;
- }
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
index bb4f61c63..e52bc9d6b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
@@ -348,9 +348,7 @@ public class JdbcCatalog extends AbstractCatalog {
@Override
public void close() throws Exception {
- if (!connections.isClosed()) {
- connections.close();
- }
+ connections.close();
}
private SchemaManager getSchemaManager(Identifier identifier) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcClientPool.java
b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcClientPool.java
index 287af41d2..d2ef4ecf7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcClientPool.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcClientPool.java
@@ -23,9 +23,9 @@ import org.apache.paimon.client.ClientPool;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
-import java.sql.SQLNonTransientConnectionException;
import java.util.Map;
import java.util.Properties;
+import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -33,14 +33,11 @@ import java.util.regex.Pattern;
public class JdbcClientPool extends ClientPool.ClientPoolImpl<Connection,
SQLException> {
private static final Pattern PROTOCOL_PATTERN =
Pattern.compile("jdbc:([^:]+):(.*)");
- private final String dbUrl;
- private final Map<String, String> properties;
+
private final String protocol;
public JdbcClientPool(int poolSize, String dbUrl, Map<String, String>
props) {
- super(poolSize, SQLNonTransientConnectionException.class, true);
- properties = props;
- this.dbUrl = dbUrl;
+ super(poolSize, clientSupplier(dbUrl, props));
Matcher matcher = PROTOCOL_PATTERN.matcher(dbUrl);
if (matcher.matches()) {
this.protocol = matcher.group(1);
@@ -49,21 +46,16 @@ public class JdbcClientPool extends
ClientPool.ClientPoolImpl<Connection, SQLExc
}
}
- @Override
- protected Connection newClient() {
- try {
- Properties dbProps =
- JdbcUtils.extractJdbcConfiguration(properties,
JdbcCatalog.PROPERTY_PREFIX);
- return DriverManager.getConnection(dbUrl, dbProps);
- } catch (SQLException e) {
- throw new RuntimeException(String.format("Failed to connect: %s",
dbUrl), e);
- }
- }
-
- @Override
- protected Connection reconnect(Connection client) {
- close(client);
- return newClient();
+ private static Supplier<Connection> clientSupplier(String dbUrl,
Map<String, String> props) {
+ return () -> {
+ try {
+ Properties dbProps =
+ JdbcUtils.extractJdbcConfiguration(props,
JdbcCatalog.PROPERTY_PREFIX);
+ return DriverManager.getConnection(dbUrl, dbProps);
+ } catch (SQLException e) {
+ throw new RuntimeException(String.format("Failed to connect:
%s", dbUrl), e);
+ }
+ };
}
public String getProtocol() {
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/CachedClientPool.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/CachedClientPool.java
index a03098083..f1d09e297 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/CachedClientPool.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/CachedClientPool.java
@@ -133,24 +133,12 @@ public class CachedClientPool implements
ClientPool<IMetaStoreClient, TException
return clientPool().run(action);
}
- @Override
- public <R> R run(Action<R, IMetaStoreClient, TException> action, boolean
retry)
- throws TException, InterruptedException {
- return clientPool().run(action, retry);
- }
-
@Override
public void execute(ExecuteAction<IMetaStoreClient, TException> action)
throws TException, InterruptedException {
clientPool().execute(action);
}
- @Override
- public void execute(ExecuteAction<IMetaStoreClient, TException> action,
boolean retry)
- throws TException, InterruptedException {
- clientPool().execute(action, retry);
- }
-
@VisibleForTesting
static Key extractKey(String clientClassName, String cacheKeys,
Configuration conf) {
// generate key elements in a certain order, so that the Key instances
are comparable
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/HiveClientPool.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/HiveClientPool.java
index 6dd086950..687baf37b 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/HiveClientPool.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/pool/HiveClientPool.java
@@ -18,16 +18,15 @@
package org.apache.paimon.hive.pool;
-import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.client.ClientPool;
import org.apache.paimon.hive.RetryingMetaStoreClientFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.thrift.TException;
-import org.apache.thrift.transport.TTransportException;
+
+import java.util.function.Supplier;
/**
* Pool of Hive Metastore clients.
@@ -36,49 +35,19 @@ import org.apache.thrift.transport.TTransportException;
*/
public class HiveClientPool extends
ClientPool.ClientPoolImpl<IMetaStoreClient, TException> {
- private final HiveConf hiveConf;
- private final String clientClassName;
-
public HiveClientPool(int poolSize, Configuration conf, String
clientClassName) {
- // Do not allow retry by default as we rely on RetryingHiveClient
- super(poolSize, TTransportException.class, false);
- this.hiveConf = new HiveConf(conf, HiveClientPool.class);
- this.hiveConf.addResource(conf);
- this.clientClassName = clientClassName;
- }
-
- @Override
- protected IMetaStoreClient newClient() {
- return new RetryingMetaStoreClientFactory().createClient(hiveConf,
clientClassName);
+ super(poolSize, clientSupplier(conf, clientClassName));
}
- @Override
- protected IMetaStoreClient reconnect(IMetaStoreClient client) {
- try {
- client.close();
- client.reconnect();
- } catch (MetaException e) {
- throw new RuntimeException("Failed to reconnect to Hive
Metastore", e);
- }
- return client;
- }
-
- @Override
- protected boolean isConnectionException(Exception e) {
- return super.isConnectionException(e)
- || (e instanceof MetaException
- && e.getMessage()
- .contains(
- "Got exception:
org.apache.thrift.transport.TTransportException"));
+ private static Supplier<IMetaStoreClient> clientSupplier(
+ Configuration conf, String clientClassName) {
+ HiveConf hiveConf = new HiveConf(conf, HiveClientPool.class);
+ hiveConf.addResource(conf);
+ return () -> new
RetryingMetaStoreClientFactory().createClient(hiveConf, clientClassName);
}
@Override
protected void close(IMetaStoreClient client) {
client.close();
}
-
- @VisibleForTesting
- HiveConf hiveConf() {
- return hiveConf;
- }
}
diff --git
a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/pool/TestHiveClientPool.java
b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/pool/TestHiveClientPool.java
deleted file mode 100644
index a5daefae7..000000000
---
a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/pool/TestHiveClientPool.java
+++ /dev/null
@@ -1,190 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.hive.pool;
-
-import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
-import org.apache.hadoop.hive.metastore.api.Function;
-import org.apache.hadoop.hive.metastore.api.FunctionType;
-import org.apache.hadoop.hive.metastore.api.GetAllFunctionsResponse;
-import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.hive.metastore.api.PrincipalType;
-import org.apache.thrift.transport.TTransportException;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/**
- * Test Hive Client Pool.
- *
- * <p>Mostly copied from iceberg.
- */
-public class TestHiveClientPool {
-
- private static final String HIVE_SITE_CONTENT =
- "<?xml version=\"1.0\"?>\n"
- + "<?xml-stylesheet type=\"text/xsl\"
href=\"configuration.xsl\"?>\n"
- + "<configuration>\n"
- + " <property>\n"
- + " <name>hive.metastore.sasl.enabled</name>\n"
- + " <value>true</value>\n"
- + " </property>\n"
- + "</configuration>\n";
-
- HiveClientPool clients;
-
- @BeforeEach
- public void before() {
- String metastoreClientClass =
"org.apache.hadoop.hive.metastore.HiveMetaStoreClient";
- HiveClientPool clientPool =
- new HiveClientPool(2, new Configuration(),
metastoreClientClass);
- clients = Mockito.spy(clientPool);
- }
-
- @AfterEach
- public void after() {
- clients.close();
- clients = null;
- }
-
- @Test
- public void testConf() {
- HiveConf conf = createHiveConf();
- conf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname, "warehouse");
-
- String metastoreClientClass =
"org.apache.hadoop.hive.metastore.HiveMetaStoreClient";
- HiveClientPool clientPool = new HiveClientPool(10, conf,
metastoreClientClass);
- HiveConf clientConf = clientPool.hiveConf();
-
-
assertThat(clientConf.get(HiveConf.ConfVars.METASTOREWAREHOUSE.varname))
-
.isEqualTo(conf.get(HiveConf.ConfVars.METASTOREWAREHOUSE.varname));
- assertThat(clientPool.poolSize()).isEqualTo(10);
-
- // 'hive.metastore.sasl.enabled' should be 'true' as defined in xml
-
assertThat(clientConf.get(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL.varname))
-
.isEqualTo(conf.get(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL.varname));
-
assertThat(clientConf.getBoolVar(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL)).isTrue();
- }
-
- private HiveConf createHiveConf() {
- HiveConf hiveConf = new HiveConf();
- try (InputStream inputStream =
- new
ByteArrayInputStream(HIVE_SITE_CONTENT.getBytes(StandardCharsets.UTF_8))) {
- hiveConf.addResource(inputStream, "for_test");
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- return hiveConf;
- }
-
- @Test
- public void testNewClientFailure() {
- Mockito.doThrow(new RuntimeException("Connection
exception")).when(clients).newClient();
- assertThatThrownBy(() -> clients.run(Object::toString))
- .isInstanceOf(RuntimeException.class)
- .hasMessage("Connection exception");
- }
-
- @Test
- public void testGetTablesFailsForNonReconnectableException() throws
Exception {
- HiveMetaStoreClient hmsClient =
Mockito.mock(HiveMetaStoreClient.class);
- Mockito.doReturn(hmsClient).when(clients).newClient();
- Mockito.doThrow(new MetaException("Another meta exception"))
- .when(hmsClient)
- .getTables(Mockito.anyString(), Mockito.anyString());
- assertThatThrownBy(() -> clients.run(client ->
client.getTables("default", "t")))
- .isInstanceOf(MetaException.class)
- .hasMessage("Another meta exception");
- }
-
- @Test
- public void testConnectionFailureRestoreForMetaException() throws
Exception {
- HiveMetaStoreClient hmsClient = newClient();
-
- // Throwing an exception may trigger the client to reconnect.
- String metaMessage = "Got exception:
org.apache.thrift.transport.TTransportException";
- Mockito.doThrow(new
MetaException(metaMessage)).when(hmsClient).getAllDatabases();
-
- // Create a new client when the reconnect method is called.
- HiveMetaStoreClient newClient = reconnect(hmsClient);
-
- List<String> databases = Lists.newArrayList("db1", "db2");
-
- Mockito.doReturn(databases).when(newClient).getAllDatabases();
- // The return is OK when the reconnect method is called.
- assertThat((List<String>) clients.run(client ->
client.getAllDatabases(), true))
- .isEqualTo(databases);
-
- // Verify that the method is called.
- Mockito.verify(clients).reconnect(hmsClient);
- Mockito.verify(clients, Mockito.never()).reconnect(newClient);
- }
-
- @Test
- public void testConnectionFailureRestoreForTTransportException() throws
Exception {
- HiveMetaStoreClient hmsClient = newClient();
- Mockito.doThrow(new
TTransportException()).when(hmsClient).getAllFunctions();
-
- // Create a new client when getAllFunctions() failed.
- HiveMetaStoreClient newClient = reconnect(hmsClient);
-
- GetAllFunctionsResponse response = new GetAllFunctionsResponse();
- response.addToFunctions(
- new Function(
- "concat",
- "db1",
- "classname",
- "root",
- PrincipalType.USER,
- 100,
- FunctionType.JAVA,
- null));
- Mockito.doReturn(response).when(newClient).getAllFunctions();
- assertThat((GetAllFunctionsResponse) clients.run(client ->
client.getAllFunctions(), true))
- .isEqualTo(response);
-
- Mockito.verify(clients).reconnect(hmsClient);
- Mockito.verify(clients, Mockito.never()).reconnect(newClient);
- }
-
- private HiveMetaStoreClient newClient() {
- HiveMetaStoreClient hmsClient =
Mockito.mock(HiveMetaStoreClient.class);
- Mockito.doReturn(hmsClient).when(clients).newClient();
- return hmsClient;
- }
-
- private HiveMetaStoreClient reconnect(HiveMetaStoreClient obsoleteClient) {
- HiveMetaStoreClient newClient =
Mockito.mock(HiveMetaStoreClient.class);
- Mockito.doReturn(newClient).when(clients).reconnect(obsoleteClient);
- return newClient;
- }
-}
