From ae2000498a78db5f5e9b5b178900b8a05fb0087e Mon Sep 17 00:00:00 2001 From: Nick Dimiduk Date: Mon, 21 Mar 2022 12:42:18 +0100 Subject: [PATCH] HBASE-26834 Adapt ConnectionRule for both sync and async connections Signed-off-by: Duo Zhang --- .../TestShellExecEndpointCoprocessor.java | 8 +- .../apache/hadoop/hbase/ConnectionRule.java | 84 +++++++++++++++---- .../apache/hadoop/hbase/MiniClusterRule.java | 20 ++++- .../http/TestApiV1ClusterMetricsResource.java | 4 +- .../hbase/master/http/TestMetaBrowser.java | 6 +- 5 files changed, 97 insertions(+), 25 deletions(-) diff --git a/hbase-it/src/test/java/org/apache/hadoop/hbase/TestShellExecEndpointCoprocessor.java b/hbase-it/src/test/java/org/apache/hadoop/hbase/TestShellExecEndpointCoprocessor.java index 81519bc449f..462b2c51bb5 100644 --- a/hbase-it/src/test/java/org/apache/hadoop/hbase/TestShellExecEndpointCoprocessor.java +++ b/hbase-it/src/test/java/org/apache/hadoop/hbase/TestShellExecEndpointCoprocessor.java @@ -56,7 +56,7 @@ public class TestShellExecEndpointCoprocessor { @Rule public final ConnectionRule connectionRule = - new ConnectionRule(miniClusterRule::createConnection); + ConnectionRule.createAsyncConnectionRule(miniClusterRule::createAsyncConnection); @Test public void testShellExecUnspecified() { @@ -69,7 +69,7 @@ public class TestShellExecEndpointCoprocessor { } private void testShellExecForeground(final Consumer consumer) { - final AsyncConnection conn = connectionRule.getConnection(); + final AsyncConnection conn = connectionRule.getAsyncConnection(); final AsyncAdmin admin = conn.getAdmin(); final String command = "echo -n \"hello world\""; @@ -87,7 +87,7 @@ public class TestShellExecEndpointCoprocessor { @Test public void testShellExecBackground() throws IOException { - final AsyncConnection conn = connectionRule.getConnection(); + final AsyncConnection conn = connectionRule.getAsyncConnection(); final AsyncAdmin admin = conn.getAdmin(); final File testDataDir = ensureTestDataDirExists(miniClusterRule.getTestingUtility()); @@ -121,7 +121,7 @@ public class TestShellExecEndpointCoprocessor { final Path testDataDir = Optional.of(testingUtility) .map(HBaseTestingUtility::getDataTestDir) .map(Object::toString) - .map(val -> Paths.get(val)) + .map(Paths::get) .orElseThrow(() -> new RuntimeException("Unable to locate temp directory path.")); final File testDataDirFile = Files.createDirectories(testDataDir).toFile(); assertTrue(testDataDirFile.exists()); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ConnectionRule.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ConnectionRule.java index 21ca35adb4a..b029cf97e78 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ConnectionRule.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ConnectionRule.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.concurrent.CompletableFuture; import java.util.function.Supplier; import org.apache.hadoop.hbase.client.AsyncConnection; +import org.apache.hadoop.hbase.client.Connection; import org.junit.ClassRule; import org.junit.Rule; import org.junit.rules.ExternalResource; @@ -35,7 +36,7 @@ import org.junit.rules.ExternalResource; * public class TestMyClass { * private static final MiniClusterRule miniClusterRule = MiniClusterRule.newBuilder().build(); * private static final ConnectionRule connectionRule = - * new ConnectionRule(miniClusterRule::createConnection); + * ConnectionRule.createAsyncConnectionRule(miniClusterRule::createConnection); * * @ClassRule * public static final TestRule rule = RuleChain @@ -44,32 +45,87 @@ import org.junit.rules.ExternalResource; * } * } */ -public class ConnectionRule extends ExternalResource { +public final class ConnectionRule extends ExternalResource { - private final Supplier> connectionSupplier; - private AsyncConnection connection; + private final Supplier connectionSupplier; + private final Supplier> asyncConnectionSupplier; - public ConnectionRule(final Supplier> connectionSupplier) { - this.connectionSupplier = connectionSupplier; + private Connection connection; + private AsyncConnection asyncConnection; + + public static ConnectionRule createConnectionRule( + final Supplier connectionSupplier + ) { + return new ConnectionRule(connectionSupplier, null); } - public AsyncConnection getConnection() { + public static ConnectionRule createAsyncConnectionRule( + final Supplier> asyncConnectionSupplier + ) { + return new ConnectionRule(null, asyncConnectionSupplier); + } + + public static ConnectionRule createConnectionRule( + final Supplier connectionSupplier, + final Supplier> asyncConnectionSupplier + ) { + return new ConnectionRule(connectionSupplier, asyncConnectionSupplier); + } + + private ConnectionRule( + final Supplier connectionSupplier, + final Supplier> asyncConnectionSupplier + ) { + this.connectionSupplier = connectionSupplier; + this.asyncConnectionSupplier = asyncConnectionSupplier; + } + + public Connection getConnection() { + if (connection == null) { + throw new IllegalStateException( + "ConnectionRule not initialized with a synchronous connection."); + } return connection; } + public AsyncConnection getAsyncConnection() { + if (asyncConnection == null) { + throw new IllegalStateException( + "ConnectionRule not initialized with an asynchronous connection."); + } + return asyncConnection; + } + @Override protected void before() throws Throwable { - this.connection = connectionSupplier.get().join(); + if (connectionSupplier != null) { + this.connection = connectionSupplier.get(); + } + if (asyncConnectionSupplier != null) { + this.asyncConnection = asyncConnectionSupplier.get().join(); + } } @Override protected void after() { - if (this.connection != null) { - try { - connection.close(); - } catch (IOException e) { - throw new RuntimeException(e); + CompletableFuture closeConnection = CompletableFuture.runAsync(() -> { + if (this.connection != null) { + try { + connection.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } } - } + }); + CompletableFuture closeAsyncConnection = CompletableFuture.runAsync(() -> { + if (this.asyncConnection != null) { + try { + asyncConnection.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + }); + CompletableFuture.allOf(closeConnection, closeAsyncConnection).join(); } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/MiniClusterRule.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/MiniClusterRule.java index 6c31f4f6f25..8789a9d19a9 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/MiniClusterRule.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/MiniClusterRule.java @@ -22,6 +22,7 @@ import java.util.concurrent.CompletableFuture; import java.util.function.Supplier; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.client.AsyncConnection; +import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.junit.ClassRule; import org.junit.Rule; @@ -42,7 +43,7 @@ import org.junit.rules.TestRule; * * @Rule * public final ConnectionRule connectionRule = - * new ConnectionRule(miniClusterRule::createConnection); + * ConnectionRule.createAsyncConnectionRule(miniClusterRule::createAsyncConnection); * } * } */ @@ -107,11 +108,26 @@ public final class MiniClusterRule extends ExternalResource { return testingUtility; } + /** + * Create a {@link Connection} to the managed {@link MiniHBaseCluster}. It's up to the caller + * to {@link Connection#close() close()} the connection when finished. + */ + public Connection createConnection() { + if (miniCluster == null) { + throw new IllegalStateException("test cluster not initialized"); + } + try { + return ConnectionFactory.createConnection(miniCluster.getConf()); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + /** * Create a {@link AsyncConnection} to the managed {@link MiniHBaseCluster}. It's up to the caller * to {@link AsyncConnection#close() close()} the connection when finished. */ - public CompletableFuture createConnection() { + public CompletableFuture createAsyncConnection() { if (miniCluster == null) { throw new IllegalStateException("test cluster not initialized"); } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestApiV1ClusterMetricsResource.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestApiV1ClusterMetricsResource.java index 07c76b57103..0b0c4cbf3a1 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestApiV1ClusterMetricsResource.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestApiV1ClusterMetricsResource.java @@ -80,8 +80,8 @@ public class TestApiV1ClusterMetricsResource { }) .build(); private static final ConnectionRule connectionRule = - new ConnectionRule(miniClusterRule::createConnection); - private static final ClassSetup classRule = new ClassSetup(connectionRule::getConnection); + ConnectionRule.createAsyncConnectionRule(miniClusterRule::createAsyncConnection); + private static final ClassSetup classRule = new ClassSetup(connectionRule::getAsyncConnection); private static final class ClassSetup extends ExternalResource { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestMetaBrowser.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestMetaBrowser.java index cd20fe9698e..cfceb47f001 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestMetaBrowser.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestMetaBrowser.java @@ -68,9 +68,9 @@ public class TestMetaBrowser { public static final MiniClusterRule miniClusterRule = MiniClusterRule.newBuilder().build(); private final ConnectionRule connectionRule = - new ConnectionRule(miniClusterRule::createConnection); + ConnectionRule.createAsyncConnectionRule(miniClusterRule::createAsyncConnection); private final ClearUserNamespacesAndTablesRule clearUserNamespacesAndTablesRule = - new ClearUserNamespacesAndTablesRule(connectionRule::getConnection); + new ClearUserNamespacesAndTablesRule(connectionRule::getAsyncConnection); @Rule public TestRule rule = RuleChain.outerRule(connectionRule) @@ -84,7 +84,7 @@ public class TestMetaBrowser { @Before public void before() { - connection = connectionRule.getConnection(); + connection = connectionRule.getAsyncConnection(); admin = connection.getAdmin(); }