diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScannersFromClientSide.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScannersFromClientSide.java
index f53071c21ed3..d070c7c7aa85 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScannersFromClientSide.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScannersFromClientSide.java
@@ -20,18 +20,16 @@
 import static org.apache.hadoop.hbase.HConstants.RPC_CODEC_CONF_KEY;
 import static org.apache.hadoop.hbase.client.FromClientSideTest3.generateHugeValue;
 import static org.apache.hadoop.hbase.ipc.RpcClient.DEFAULT_CODEC_CLASS;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -39,8 +37,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CompareOperator;
-import org.apache.hadoop.hbase.HBaseClassTestRule;
-import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionInfo;
@@ -48,9 +45,7 @@
 import org.apache.hadoop.hbase.HTestConst;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.MiniHBaseCluster;
-import org.apache.hadoop.hbase.StartMiniClusterOption;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.TableNameTestRule;
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.filter.BinaryComparator;
@@ -63,303 +58,226 @@
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.TestTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
-
 /**
  * A client-side test, mostly testing scanners with various parameters. Parameterized on different
  * registry implementations.
  */
-@Category({ ClientTests.class, LargeTests.class })
-@RunWith(Parameterized.class)
-public class TestScannersFromClientSide {
-
-  @ClassRule
-  public static final HBaseClassTestRule CLASS_RULE =
-    HBaseClassTestRule.forClass(TestScannersFromClientSide.class);
-
-  private static final Logger LOG = LoggerFactory.getLogger(TestScannersFromClientSide.class);
-
-  private static HBaseTestingUtility TEST_UTIL;
-  private static byte[] ROW = Bytes.toBytes("testRow");
-  private static byte[] FAMILY = Bytes.toBytes("testFamily");
-  private static byte[] QUALIFIER = Bytes.toBytes("testQualifier");
-  private static byte[] VALUE = Bytes.toBytes("testValue");
-
-  @Rule
-  public TableNameTestRule name = new TableNameTestRule();
-
-  /**
-   * @throws java.lang.Exception
-   */
-  @AfterClass
-  public static void tearDownAfterClass() throws Exception {
-    if (TEST_UTIL != null) {
-      TEST_UTIL.shutdownMiniCluster();
-    }
+@Tag(LargeTests.TAG)
+@Tag(ClientTests.TAG)
+@HBaseParameterizedTestTemplate(name = "{index}: registryImpl={0}, numHedgedReqs={1}")
+public class TestScannersFromClientSide extends FromClientSideTestBase {
+
+  public TestScannersFromClientSide(Class<? extends ConnectionRegistry> registryImpl,
+    int numHedgedReqs) {
+    super(registryImpl, numHedgedReqs);
   }
 
-  /**
-   * @throws java.lang.Exception
-   */
-  @Before
-  public void setUp() throws Exception {
-    // Nothing to do.
-  }
-
-  @Parameterized.Parameters
-  public static Collection<Object[]> parameters() {
-    return Arrays.asList(new Object[][] { { MasterRegistry.class, 1 }, { MasterRegistry.class, 2 },
-      { ZKConnectionRegistry.class, 1 } });
-  }
+  private static final Logger LOG = LoggerFactory.getLogger(TestScannersFromClientSide.class);
 
-  /**
-   * JUnit does not provide an easy way to run a hook after each parameterized run. Without that
-   * there is no easy way to restart the test cluster after each parameterized run. Annotation
-   * BeforeParam does not work either because it runs before parameterization and hence does not
-   * have access to the test parameters (which is weird). This *hack* checks if the current instance
-   * of test cluster configuration has the passed parameterized configs. In such a case, we can just
-   * reuse the cluster for test and do not need to initialize from scratch. While this is a hack, it
-   * saves a ton of time for the full test and de-flakes it.
-   */
-  private static boolean isSameParameterizedCluster(Class<?> registryImpl, int numHedgedReqs) {
-    // initialize() is called for every unit test, however we only want to reset the cluster state
-    // at the end of every parameterized run.
-    if (TEST_UTIL == null) {
-      return false;
-    }
-    Configuration conf = TEST_UTIL.getConfiguration();
-    Class<?> confClass = conf.getClass(HConstants.CLIENT_CONNECTION_REGISTRY_IMPL_CONF_KEY,
-      ZKConnectionRegistry.class);
-    int hedgedReqConfig = conf.getInt(MasterRegistry.MASTER_REGISTRY_HEDGED_REQS_FANOUT_KEY,
-      AbstractRpcBasedConnectionRegistry.HEDGED_REQS_FANOUT_DEFAULT);
-    return confClass.getName().equals(registryImpl.getName()) && numHedgedReqs == hedgedReqConfig;
-  }
-
-  public TestScannersFromClientSide(Class<?> registryImpl, int numHedgedReqs) throws Exception {
-    if (isSameParameterizedCluster(registryImpl, numHedgedReqs)) {
-      return;
-    }
-    if (TEST_UTIL != null) {
-      // We reached the end of a parameterized run, clean up the cluster.
-      TEST_UTIL.shutdownMiniCluster();
-    }
-    TEST_UTIL = new HBaseTestingUtility();
+  @BeforeAll
+  public static void setUpBeforeClass() throws Exception {
     Configuration conf = TEST_UTIL.getConfiguration();
     conf.setLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, 10 * 1024 * 1024);
-    conf.setClass(HConstants.CLIENT_CONNECTION_REGISTRY_IMPL_CONF_KEY, registryImpl,
-      ConnectionRegistry.class);
-    Preconditions.checkArgument(numHedgedReqs > 0);
-    conf.setInt(MasterRegistry.MASTER_REGISTRY_HEDGED_REQS_FANOUT_KEY, numHedgedReqs);
-    StartMiniClusterOption.Builder builder = StartMiniClusterOption.builder();
-    // Multiple masters needed only when hedged reads for master registry are enabled.
-    builder.numMasters(numHedgedReqs > 1 ? 3 : 1).numRegionServers(3);
-    TEST_UTIL.startMiniCluster(builder.build());
+    SLAVES = 3;
+    initialize();
   }
 
-  @Test
+  @TestTemplate
   public void testScanImmutable() throws IOException {
-    TableName tableName = name.getTableName();
-    Table table = TEST_UTIL.createTable(tableName, FAMILY);
-    TEST_UTIL.loadRandomRows(table, FAMILY, 100, 100);
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName)) {
+      TEST_UTIL.loadRandomRows(table, FAMILY, 100, 100);
 
-    Scan scan = new Scan().setCaching(-1).setMvccReadPoint(-1).setScanMetricsEnabled(true);
+      Scan scan = new Scan().setCaching(-1).setMvccReadPoint(-1).setScanMetricsEnabled(true);
 
-    try (ResultScanner scanner = table.getScanner(scan)) {
-      scanner.next(1000);
+      try (ResultScanner scanner = table.getScanner(scan)) {
+        scanner.next(1000);
+      }
+      // these 2 should be unchanged
+      assertEquals(-1, scan.getCaching());
+      assertEquals(-1, scan.getMvccReadPoint());
+      // scan metrics should be populated
+      assertNotNull(scan.getScanMetrics());
+      assertEquals(scan.getScanMetrics().countOfRegions.get(), 1);
     }
-    // these 2 should be unchanged
-    assertEquals(-1, scan.getCaching());
-    assertEquals(-1, scan.getMvccReadPoint());
-    // scan metrics should be populated
-    assertNotNull(scan.getScanMetrics());
-    assertEquals(scan.getScanMetrics().countOfRegions.get(), 1);
   }
 
   /**
    * Test from client side for batch of scan
    */
-  @Test
+  @TestTemplate
   public void testScanBatch() throws Exception {
-    final TableName tableName = name.getTableName();
     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 8);
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      Put put;
+      Scan scan;
+      Delete delete;
+      Result result;
+      ResultScanner scanner;
+      boolean toLog = true;
+      List<Cell> kvListExp;
 
-    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
-
-    Put put;
-    Scan scan;
-    Delete delete;
-    Result result;
-    ResultScanner scanner;
-    boolean toLog = true;
-    List<Cell> kvListExp;
+      // table: row, family, c0:0, c1:1, ... , c7:7
+      put = new Put(ROW);
+      for (int i = 0; i < QUALIFIERS.length; i++) {
+        KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
+        put.add(kv);
+      }
+      ht.put(put);
 
-    // table: row, family, c0:0, c1:1, ... , c7:7
-    put = new Put(ROW);
-    for (int i = 0; i < QUALIFIERS.length; i++) {
-      KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
+      // table: row, family, c0:0, c1:1, ..., c6:2, c6:6 , c7:7
+      put = new Put(ROW);
+      KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[6], 2, VALUE);
       put.add(kv);
+      ht.put(put);
+
+      // delete upto ts: 3
+      delete = new Delete(ROW);
+      delete.addFamily(FAMILY, 3);
+      ht.delete(delete);
+
+      // without batch
+      scan = new Scan().withStartRow(ROW);
+      scan.setMaxVersions();
+      scanner = ht.getScanner(scan);
+
+      // c4:4, c5:5, c6:6, c7:7
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
+      result = scanner.next();
+      verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
+
+      // with batch
+      scan = new Scan().withStartRow(ROW);
+      scan.setMaxVersions();
+      scan.setBatch(2);
+      scanner = ht.getScanner(scan);
+
+      // First batch: c4:4, c5:5
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
+      result = scanner.next();
+      verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
+
+      // Second batch: c6:6, c7:7
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
+      result = scanner.next();
+      verifyResult(result, kvListExp, toLog, "Testing second batch of scan");
     }
-    ht.put(put);
-
-    // table: row, family, c0:0, c1:1, ..., c6:2, c6:6 , c7:7
-    put = new Put(ROW);
-    KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[6], 2, VALUE);
-    put.add(kv);
-    ht.put(put);
-
-    // delete upto ts: 3
-    delete = new Delete(ROW);
-    delete.addFamily(FAMILY, 3);
-    ht.delete(delete);
-
-    // without batch
-    scan = new Scan().withStartRow(ROW);
-    scan.setMaxVersions();
-    scanner = ht.getScanner(scan);
-
-    // c4:4, c5:5, c6:6, c7:7
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
-    result = scanner.next();
-    verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
-
-    // with batch
-    scan = new Scan().withStartRow(ROW);
-    scan.setMaxVersions();
-    scan.setBatch(2);
-    scanner = ht.getScanner(scan);
-
-    // First batch: c4:4, c5:5
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[4], 4, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[5], 5, VALUE));
-    result = scanner.next();
-    verifyResult(result, kvListExp, toLog, "Testing first batch of scan");
-
-    // Second batch: c6:6, c7:7
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[6], 6, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[7], 7, VALUE));
-    result = scanner.next();
-    verifyResult(result, kvListExp, toLog, "Testing second batch of scan");
 
   }
 
-  @Test
+  @TestTemplate
   public void testMaxResultSizeIsSetToDefault() throws Exception {
-    final TableName tableName = name.getTableName();
-    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      // The max result size we expect the scan to use by default.
+      long expectedMaxResultSize =
+        TEST_UTIL.getConfiguration().getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
+          HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
 
-    // The max result size we expect the scan to use by default.
-    long expectedMaxResultSize =
-      TEST_UTIL.getConfiguration().getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
-        HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
+      int numRows = 5;
+      byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
 
-    int numRows = 5;
-    byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
+      int numQualifiers = 10;
+      byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
 
-    int numQualifiers = 10;
-    byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
+      // Specify the cell size such that a single row will be larger than the default
+      // value of maxResultSize. This means that Scan RPCs should return at most a single
+      // result back to the client.
+      int cellSize = (int) (expectedMaxResultSize / (numQualifiers - 1));
+      byte[] cellValue = Bytes.createMaxByteArray(cellSize);
 
-    // Specify the cell size such that a single row will be larger than the default
-    // value of maxResultSize. This means that Scan RPCs should return at most a single
-    // result back to the client.
-    int cellSize = (int) (expectedMaxResultSize / (numQualifiers - 1));
-    byte[] cellValue = Bytes.createMaxByteArray(cellSize);
-
-    Put put;
-    List<Put> puts = new ArrayList<>();
-    for (int row = 0; row < ROWS.length; row++) {
-      put = new Put(ROWS[row]);
-      for (int qual = 0; qual < QUALIFIERS.length; qual++) {
-        KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], cellValue);
-        put.add(kv);
+      Put put;
+      List<Put> puts = new ArrayList<>();
+      for (int row = 0; row < ROWS.length; row++) {
+        put = new Put(ROWS[row]);
+        for (int qual = 0; qual < QUALIFIERS.length; qual++) {
+          KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], cellValue);
+          put.add(kv);
+        }
+        puts.add(put);
       }
-      puts.add(put);
-    }
-    ht.put(puts);
+      ht.put(puts);
 
-    // Create a scan with the default configuration.
-    Scan scan = new Scan();
+      // Create a scan with the default configuration.
+      Scan scan = new Scan();
 
-    ResultScanner scanner = ht.getScanner(scan);
-    assertTrue(scanner instanceof ClientScanner);
-    ClientScanner clientScanner = (ClientScanner) scanner;
+      try (ResultScanner scanner = ht.getScanner(scan)) {
+        assertTrue(scanner instanceof ClientScanner);
+        ClientScanner clientScanner = (ClientScanner) scanner;
 
-    // Call next to issue a single RPC to the server
-    scanner.next();
+        // Call next to issue a single RPC to the server
+        scanner.next();
 
-    // The scanner should have, at most, a single result in its cache. If there more results exists
-    // in the cache it means that more than the expected max result size was fetched.
-    assertTrue("The cache contains: " + clientScanner.getCacheSize() + " results",
-      clientScanner.getCacheSize() <= 1);
+        // The scanner should have, at most, a single result in its cache. If there more results
+        // exists in the cache it means that more than the expected max result size was fetched.
+        assertTrue(clientScanner.getCacheSize() <= 1,
+          "The cache contains: " + clientScanner.getCacheSize() + " results");
+      }
+    }
   }
 
   /**
    * Scan on not existing table should throw the exception with correct message
    */
-  @Test
-  public void testScannerForNotExistingTable() {
+  @TestTemplate
+  public void testScannerForNotExistingTable() throws Exception {
     String[] tableNames = { "A", "Z", "A:A", "Z:Z" };
     for (String tableName : tableNames) {
-      try {
-        Table table = TEST_UTIL.getConnection().getTable(TableName.valueOf(tableName));
-        testSmallScan(table, true, 1, 5);
-        fail("TableNotFoundException was not thrown");
-      } catch (TableNotFoundException e) {
+      try (Connection conn = getConnection();
+        Table table = conn.getTable(TableName.valueOf(tableName))) {
+        TableNotFoundException e = assertThrows(TableNotFoundException.class,
+          () -> testSmallScan(table, true, 1, 5), "TableNotFoundException was not thrown");
         // We expect that the message for TableNotFoundException would have only the table name only
         // Otherwise that would mean that localeRegionInMeta doesn't work properly
         assertEquals(e.getMessage(), tableName);
-      } catch (Exception e) {
-        fail("Unexpected exception " + e.getMessage());
       }
     }
   }
 
-  @Test
+  @TestTemplate
   public void testSmallScan() throws Exception {
-    final TableName tableName = name.getTableName();
-
     int numRows = 10;
     byte[][] ROWS = HTestConst.makeNAscii(ROW, numRows);
 
     int numQualifiers = 10;
     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, numQualifiers);
 
-    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
-
-    Put put;
-    List<Put> puts = new ArrayList<>();
-    for (int row = 0; row < ROWS.length; row++) {
-      put = new Put(ROWS[row]);
-      for (int qual = 0; qual < QUALIFIERS.length; qual++) {
-        KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], VALUE);
-        put.add(kv);
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      Put put;
+      List<Put> puts = new ArrayList<>();
+      for (int row = 0; row < ROWS.length; row++) {
+        put = new Put(ROWS[row]);
+        for (int qual = 0; qual < QUALIFIERS.length; qual++) {
+          KeyValue kv = new KeyValue(ROWS[row], FAMILY, QUALIFIERS[qual], VALUE);
+          put.add(kv);
+        }
+        puts.add(put);
       }
-      puts.add(put);
-    }
-    ht.put(puts);
+      ht.put(puts);
 
-    int expectedRows = numRows;
-    int expectedCols = numRows * numQualifiers;
+      int expectedRows = numRows;
+      int expectedCols = numRows * numQualifiers;
 
-    // Test normal and reversed
-    testSmallScan(ht, true, expectedRows, expectedCols);
-    testSmallScan(ht, false, expectedRows, expectedCols);
+      // Test normal and reversed
+      testSmallScan(ht, true, expectedRows, expectedCols);
+      testSmallScan(ht, false, expectedRows, expectedCols);
+    }
   }
 
   /**
@@ -386,284 +304,281 @@ private void testSmallScan(Table table, boolean reversed, int rows, int columns)
 
   private void verifyExpectedCounts(Table table, Scan scan, int expectedRowCount,
     int expectedCellCount) throws Exception {
-    ResultScanner scanner = table.getScanner(scan);
-
-    int rowCount = 0;
-    int cellCount = 0;
-    Result r = null;
-    while ((r = scanner.next()) != null) {
-      rowCount++;
-      cellCount += r.rawCells().length;
-    }
+    try (ResultScanner scanner = table.getScanner(scan)) {
+      int rowCount = 0;
+      int cellCount = 0;
+      Result r = null;
+      while ((r = scanner.next()) != null) {
+        rowCount++;
+        cellCount += r.rawCells().length;
+      }
 
-    assertTrue("Expected row count: " + expectedRowCount + " Actual row count: " + rowCount,
-      expectedRowCount == rowCount);
-    assertTrue("Expected cell count: " + expectedCellCount + " Actual cell count: " + cellCount,
-      expectedCellCount == cellCount);
-    scanner.close();
+      assertEquals(expectedRowCount, rowCount,
+        "Expected row count: " + expectedRowCount + " Actual row count: " + rowCount);
+      assertEquals(expectedCellCount, cellCount,
+        "Expected cell count: " + expectedCellCount + " Actual cell count: " + cellCount);
+    }
   }
 
   /**
    * Test from client side for get with maxResultPerCF set
    */
-  @Test
+  @TestTemplate
   public void testGetMaxResults() throws Exception {
-    final TableName tableName = name.getTableName();
     byte[][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
 
-    Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
+    TEST_UTIL.createTable(tableName, FAMILIES);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      Get get;
+      Put put;
+      Result result;
+      boolean toLog = true;
+      List<Cell> kvListExp;
 
-    Get get;
-    Put put;
-    Result result;
-    boolean toLog = true;
-    List<Cell> kvListExp;
+      kvListExp = new ArrayList<>();
+      // Insert one CF for row[0]
+      put = new Put(ROW);
+      for (int i = 0; i < 10; i++) {
+        KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
+        put.add(kv);
+        kvListExp.add(kv);
+      }
+      ht.put(put);
 
-    kvListExp = new ArrayList<>();
-    // Insert one CF for row[0]
-    put = new Put(ROW);
-    for (int i = 0; i < 10; i++) {
-      KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
-      put.add(kv);
-      kvListExp.add(kv);
-    }
-    ht.put(put);
-
-    get = new Get(ROW);
-    result = ht.get(get);
-    verifyResult(result, kvListExp, toLog, "Testing without setting maxResults");
-
-    get = new Get(ROW);
-    get.setMaxResultsPerColumnFamily(2);
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[0], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
-    verifyResult(result, kvListExp, toLog, "Testing basic setMaxResults");
-
-    // Filters: ColumnRangeFilter
-    get = new Get(ROW);
-    get.setMaxResultsPerColumnFamily(5);
-    get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5], true));
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[2], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
-    verifyResult(result, kvListExp, toLog, "Testing single CF with CRF");
-
-    // Insert two more CF for row[0]
-    // 20 columns for CF2, 10 columns for CF1
-    put = new Put(ROW);
-    for (int i = 0; i < QUALIFIERS.length; i++) {
-      KeyValue kv = new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE);
-      put.add(kv);
-    }
-    ht.put(put);
+      get = new Get(ROW);
+      result = ht.get(get);
+      verifyResult(result, kvListExp, toLog, "Testing without setting maxResults");
+
+      get = new Get(ROW);
+      get.setMaxResultsPerColumnFamily(2);
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[0], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
+      verifyResult(result, kvListExp, toLog, "Testing basic setMaxResults");
+
+      // Filters: ColumnRangeFilter
+      get = new Get(ROW);
+      get.setMaxResultsPerColumnFamily(5);
+      get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5], true));
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[2], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
+      verifyResult(result, kvListExp, toLog, "Testing single CF with CRF");
+
+      // Insert two more CF for row[0]
+      // 20 columns for CF2, 10 columns for CF1
+      put = new Put(ROW);
+      for (int i = 0; i < QUALIFIERS.length; i++) {
+        KeyValue kv = new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE);
+        put.add(kv);
+      }
+      ht.put(put);
 
-    put = new Put(ROW);
-    for (int i = 0; i < 10; i++) {
-      KeyValue kv = new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE);
-      put.add(kv);
-    }
-    ht.put(put);
-
-    get = new Get(ROW);
-    get.setMaxResultsPerColumnFamily(12);
-    get.addFamily(FAMILIES[1]);
-    get.addFamily(FAMILIES[2]);
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    // Exp: CF1:q0, ..., q9, CF2: q0, q1, q10, q11, ..., q19
-    for (int i = 0; i < 10; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
-    }
-    for (int i = 0; i < 2; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
-    }
-    for (int i = 10; i < 20; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
-    }
-    verifyResult(result, kvListExp, toLog, "Testing multiple CFs");
-
-    // Filters: ColumnRangeFilter and ColumnPrefixFilter
-    get = new Get(ROW);
-    get.setMaxResultsPerColumnFamily(3);
-    get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, null, true));
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    for (int i = 2; i < 5; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
-    }
-    for (int i = 2; i < 5; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
-    }
-    for (int i = 2; i < 5; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
-    }
-    verifyResult(result, kvListExp, toLog, "Testing multiple CFs + CRF");
-
-    get = new Get(ROW);
-    get.setMaxResultsPerColumnFamily(7);
-    get.setFilter(new ColumnPrefixFilter(QUALIFIERS[1]));
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[1], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[1], 1, VALUE));
-    for (int i = 10; i < 16; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
-    }
-    verifyResult(result, kvListExp, toLog, "Testing multiple CFs + PFF");
+      put = new Put(ROW);
+      for (int i = 0; i < 10; i++) {
+        KeyValue kv = new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE);
+        put.add(kv);
+      }
+      ht.put(put);
+
+      get = new Get(ROW);
+      get.setMaxResultsPerColumnFamily(12);
+      get.addFamily(FAMILIES[1]);
+      get.addFamily(FAMILIES[2]);
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      // Exp: CF1:q0, ..., q9, CF2: q0, q1, q10, q11, ..., q19
+      for (int i = 0; i < 10; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
+      }
+      for (int i = 0; i < 2; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
+      }
+      for (int i = 10; i < 20; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
+      }
+      verifyResult(result, kvListExp, toLog, "Testing multiple CFs");
+
+      // Filters: ColumnRangeFilter and ColumnPrefixFilter
+      get = new Get(ROW);
+      get.setMaxResultsPerColumnFamily(3);
+      get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, null, true));
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      for (int i = 2; i < 5; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
+      }
+      for (int i = 2; i < 5; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[i], 1, VALUE));
+      }
+      for (int i = 2; i < 5; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
+      }
+      verifyResult(result, kvListExp, toLog, "Testing multiple CFs + CRF");
 
+      get = new Get(ROW);
+      get.setMaxResultsPerColumnFamily(7);
+      get.setFilter(new ColumnPrefixFilter(QUALIFIERS[1]));
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[1], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[1], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[1], 1, VALUE));
+      for (int i = 10; i < 16; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[i], 1, VALUE));
+      }
+      verifyResult(result, kvListExp, toLog, "Testing multiple CFs + PFF");
+    }
   }
 
   /**
    * Test from client side for scan with maxResultPerCF set
    */
-  @Test
+  @TestTemplate
   public void testScanMaxResults() throws Exception {
-    final TableName tableName = name.getTableName();
     byte[][] ROWS = HTestConst.makeNAscii(ROW, 2);
     byte[][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 10);
 
-    Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
-
-    Put put;
-    Scan scan;
-    Result result;
-    boolean toLog = true;
-    List<Cell> kvListExp, kvListScan;
-
-    kvListExp = new ArrayList<>();
-
-    for (int r = 0; r < ROWS.length; r++) {
-      put = new Put(ROWS[r]);
-      for (int c = 0; c < FAMILIES.length; c++) {
-        for (int q = 0; q < QUALIFIERS.length; q++) {
-          KeyValue kv = new KeyValue(ROWS[r], FAMILIES[c], QUALIFIERS[q], 1, VALUE);
-          put.add(kv);
-          if (q < 4) {
-            kvListExp.add(kv);
+    TEST_UTIL.createTable(tableName, FAMILIES);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      Put put;
+      Scan scan;
+      Result result;
+      boolean toLog = true;
+      List<Cell> kvListExp, kvListScan;
+
+      kvListExp = new ArrayList<>();
+
+      for (int r = 0; r < ROWS.length; r++) {
+        put = new Put(ROWS[r]);
+        for (int c = 0; c < FAMILIES.length; c++) {
+          for (int q = 0; q < QUALIFIERS.length; q++) {
+            KeyValue kv = new KeyValue(ROWS[r], FAMILIES[c], QUALIFIERS[q], 1, VALUE);
+            put.add(kv);
+            if (q < 4) {
+              kvListExp.add(kv);
+            }
           }
         }
+        ht.put(put);
       }
-      ht.put(put);
-    }
 
-    scan = new Scan();
-    scan.setMaxResultsPerColumnFamily(4);
-    ResultScanner scanner = ht.getScanner(scan);
-    kvListScan = new ArrayList<>();
-    while ((result = scanner.next()) != null) {
-      for (Cell kv : result.listCells()) {
-        kvListScan.add(kv);
+      scan = new Scan();
+      scan.setMaxResultsPerColumnFamily(4);
+      ResultScanner scanner = ht.getScanner(scan);
+      kvListScan = new ArrayList<>();
+      while ((result = scanner.next()) != null) {
+        for (Cell kv : result.listCells()) {
+          kvListScan.add(kv);
+        }
       }
+      result = Result.create(kvListScan);
+      verifyResult(result, kvListExp, toLog, "Testing scan with maxResults");
     }
-    result = Result.create(kvListScan);
-    verifyResult(result, kvListExp, toLog, "Testing scan with maxResults");
-
   }
 
   /**
    * Test from client side for get with rowOffset
    */
-  @Test
+  @TestTemplate
   public void testGetRowOffset() throws Exception {
-    final TableName tableName = name.getTableName();
     byte[][] FAMILIES = HTestConst.makeNAscii(FAMILY, 3);
     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 20);
 
-    Table ht = TEST_UTIL.createTable(tableName, FAMILIES);
-
-    Get get;
-    Put put;
-    Result result;
-    boolean toLog = true;
-    List<Cell> kvListExp;
+    TEST_UTIL.createTable(tableName, FAMILIES);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      Get get;
+      Put put;
+      Result result;
+      boolean toLog = true;
+      List<Cell> kvListExp;
 
-    // Insert one CF for row
-    kvListExp = new ArrayList<>();
-    put = new Put(ROW);
-    for (int i = 0; i < 10; i++) {
-      KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
-      put.add(kv);
-      // skipping first two kvs
-      if (i < 2) {
-        continue;
-      }
-      kvListExp.add(kv);
-    }
-    ht.put(put);
-
-    // setting offset to 2
-    get = new Get(ROW);
-    get.setRowOffsetPerColumnFamily(2);
-    result = ht.get(get);
-    verifyResult(result, kvListExp, toLog, "Testing basic setRowOffset");
-
-    // setting offset to 20
-    get = new Get(ROW);
-    get.setRowOffsetPerColumnFamily(20);
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    verifyResult(result, kvListExp, toLog, "Testing offset > #kvs");
-
-    // offset + maxResultPerCF
-    get = new Get(ROW);
-    get.setRowOffsetPerColumnFamily(4);
-    get.setMaxResultsPerColumnFamily(5);
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    for (int i = 4; i < 9; i++) {
-      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
-    }
-    verifyResult(result, kvListExp, toLog, "Testing offset + setMaxResultsPerCF");
-
-    // Filters: ColumnRangeFilter
-    get = new Get(ROW);
-    get.setRowOffsetPerColumnFamily(1);
-    get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5], true));
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
-    verifyResult(result, kvListExp, toLog, "Testing offset with CRF");
-
-    // Insert into two more CFs for row
-    // 10 columns for CF2, 10 columns for CF1
-    for (int j = 2; j > 0; j--) {
+      // Insert one CF for row
+      kvListExp = new ArrayList<>();
       put = new Put(ROW);
       for (int i = 0; i < 10; i++) {
-        KeyValue kv = new KeyValue(ROW, FAMILIES[j], QUALIFIERS[i], 1, VALUE);
+        KeyValue kv = new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE);
         put.add(kv);
+        // skipping first two kvs
+        if (i < 2) {
+          continue;
+        }
+        kvListExp.add(kv);
       }
       ht.put(put);
-    }
 
-    get = new Get(ROW);
-    get.setRowOffsetPerColumnFamily(4);
-    get.setMaxResultsPerColumnFamily(2);
-    get.addFamily(FAMILIES[1]);
-    get.addFamily(FAMILIES[2]);
-    result = ht.get(get);
-    kvListExp = new ArrayList<>();
-    // Exp: CF1:q4, q5, CF2: q4, q5
-    kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[4], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[5], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[4], 1, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[5], 1, VALUE));
-    verifyResult(result, kvListExp, toLog, "Testing offset + multiple CFs + maxResults");
+      // setting offset to 2
+      get = new Get(ROW);
+      get.setRowOffsetPerColumnFamily(2);
+      result = ht.get(get);
+      verifyResult(result, kvListExp, toLog, "Testing basic setRowOffset");
+
+      // setting offset to 20
+      get = new Get(ROW);
+      get.setRowOffsetPerColumnFamily(20);
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      verifyResult(result, kvListExp, toLog, "Testing offset > #kvs");
+
+      // offset + maxResultPerCF
+      get = new Get(ROW);
+      get.setRowOffsetPerColumnFamily(4);
+      get.setMaxResultsPerColumnFamily(5);
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      for (int i = 4; i < 9; i++) {
+        kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[i], 1, VALUE));
+      }
+      verifyResult(result, kvListExp, toLog, "Testing offset + setMaxResultsPerCF");
+
+      // Filters: ColumnRangeFilter
+      get = new Get(ROW);
+      get.setRowOffsetPerColumnFamily(1);
+      get.setFilter(new ColumnRangeFilter(QUALIFIERS[2], true, QUALIFIERS[5], true));
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[3], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[4], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[0], QUALIFIERS[5], 1, VALUE));
+      verifyResult(result, kvListExp, toLog, "Testing offset with CRF");
+
+      // Insert into two more CFs for row
+      // 10 columns for CF2, 10 columns for CF1
+      for (int j = 2; j > 0; j--) {
+        put = new Put(ROW);
+        for (int i = 0; i < 10; i++) {
+          KeyValue kv = new KeyValue(ROW, FAMILIES[j], QUALIFIERS[i], 1, VALUE);
+          put.add(kv);
+        }
+        ht.put(put);
+      }
+
+      get = new Get(ROW);
+      get.setRowOffsetPerColumnFamily(4);
+      get.setMaxResultsPerColumnFamily(2);
+      get.addFamily(FAMILIES[1]);
+      get.addFamily(FAMILIES[2]);
+      result = ht.get(get);
+      kvListExp = new ArrayList<>();
+      // Exp: CF1:q4, q5, CF2: q4, q5
+      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[4], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[1], QUALIFIERS[5], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[4], 1, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILIES[2], QUALIFIERS[5], 1, VALUE));
+      verifyResult(result, kvListExp, toLog, "Testing offset + multiple CFs + maxResults");
+    }
   }
 
-  @Test
+  @TestTemplate
   public void testRawScanExpiredCell() throws Exception {
-    final TableName tableName = name.getTableName();
-    try (final Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName)) {
       final Put put = new Put(ROW);
       put.addColumn(FAMILY, QUALIFIER, VALUE);
       put.setTTL(0);
@@ -674,16 +589,13 @@ public void testRawScanExpiredCell() throws Exception {
         assertArrayEquals(VALUE, result.getValue(FAMILY, QUALIFIER));
         assertNull(scanner.next());
       }
-    } finally {
-      TEST_UTIL.deleteTable(tableName);
     }
   }
 
-  @Test
+  @TestTemplate
   public void testScanRawDeleteFamilyVersion() throws Exception {
-    TableName tableName = name.getTableName();
     TEST_UTIL.createTable(tableName, FAMILY);
-    Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
+    Configuration conf = getClientConf();
     conf.set(RPC_CODEC_CONF_KEY, "");
     conf.set(DEFAULT_CODEC_CLASS, "");
     try (Connection connection = ConnectionFactory.createConnection(conf);
@@ -691,105 +603,103 @@ public void testScanRawDeleteFamilyVersion() throws Exception {
       Delete delete = new Delete(ROW);
       delete.addFamilyVersion(FAMILY, 0L);
       table.delete(delete);
-      Scan scan = new Scan(ROW).setRaw(true);
+      Scan scan = new Scan().withStartRow(ROW).setRaw(true);
       ResultScanner scanner = table.getScanner(scan);
       int count = 0;
       while (scanner.next() != null) {
         count++;
       }
       assertEquals(1, count);
-    } finally {
-      TEST_UTIL.deleteTable(tableName);
     }
   }
 
   /**
    * Test from client side for scan while the region is reopened on the same region server.
    */
-  @Test
+  @TestTemplate
   public void testScanOnReopenedRegion() throws Exception {
-    final TableName tableName = name.getTableName();
     byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, 2);
 
-    Table ht = TEST_UTIL.createTable(tableName, FAMILY);
-
-    Put put;
-    Scan scan;
-    Result result;
-    ResultScanner scanner;
-    boolean toLog = false;
-    List<Cell> kvListExp;
-
-    // table: row, family, c0:0, c1:1
-    put = new Put(ROW);
-    for (int i = 0; i < QUALIFIERS.length; i++) {
-      KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
-      put.add(kv);
-    }
-    ht.put(put);
-
-    scan = new Scan().withStartRow(ROW);
-    scanner = ht.getScanner(scan);
-
-    HRegionLocation loc;
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(tableName)) {
+      Put put;
+      Scan scan;
+      Result result;
+      ResultScanner scanner;
+      boolean toLog = false;
+      List<Cell> kvListExp;
 
-    try (RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName)) {
-      loc = locator.getRegionLocation(ROW);
-    }
-    HRegionInfo hri = loc.getRegionInfo();
-    MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
-    byte[] regionName = hri.getRegionName();
-    int i = cluster.getServerWith(regionName);
-    HRegionServer rs = cluster.getRegionServer(i);
-    LOG.info("Unassigning " + hri);
-    TEST_UTIL.getAdmin().unassign(hri.getRegionName(), true);
-    long startTime = EnvironmentEdgeManager.currentTime();
-    long timeOut = 10000;
-    boolean offline = false;
-    while (true) {
-      if (rs.getOnlineRegion(regionName) == null) {
-        offline = true;
-        break;
+      // table: row, family, c0:0, c1:1
+      put = new Put(ROW);
+      for (int i = 0; i < QUALIFIERS.length; i++) {
+        KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIERS[i], i, VALUE);
+        put.add(kv);
       }
-      assertTrue("Timed out in closing the testing region",
-        EnvironmentEdgeManager.currentTime() < startTime + timeOut);
-    }
-    assertTrue(offline);
-    LOG.info("Assigning " + hri);
-    TEST_UTIL.getAdmin().assign(hri.getRegionName());
-    startTime = EnvironmentEdgeManager.currentTime();
-    while (true) {
-      rs = cluster.getRegionServer(cluster.getServerWith(regionName));
-      if (rs != null && rs.getOnlineRegion(regionName) != null) {
-        offline = false;
-        break;
+      ht.put(put);
+
+      scan = new Scan().withStartRow(ROW);
+      scanner = ht.getScanner(scan);
+
+      HRegionLocation loc;
+
+      try (RegionLocator locator = TEST_UTIL.getConnection().getRegionLocator(tableName)) {
+        loc = locator.getRegionLocation(ROW);
+      }
+      HRegionInfo hri = loc.getRegionInfo();
+      MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+      byte[] regionName = hri.getRegionName();
+      int i = cluster.getServerWith(regionName);
+      HRegionServer rs = cluster.getRegionServer(i);
+      LOG.info("Unassigning " + hri);
+      TEST_UTIL.getAdmin().unassign(hri.getRegionName(), true);
+      long startTime = EnvironmentEdgeManager.currentTime();
+      long timeOut = 10000;
+      boolean offline = false;
+      while (true) {
+        if (rs.getOnlineRegion(regionName) == null) {
+          offline = true;
+          break;
+        }
+        assertTrue(EnvironmentEdgeManager.currentTime() < startTime + timeOut,
+          "Timed out in closing the testing region");
+      }
+      assertTrue(offline);
+      LOG.info("Assigning " + hri);
+      TEST_UTIL.getAdmin().assign(hri.getRegionName());
+      startTime = EnvironmentEdgeManager.currentTime();
+      while (true) {
+        rs = cluster.getRegionServer(cluster.getServerWith(regionName));
+        if (rs != null && rs.getOnlineRegion(regionName) != null) {
+          offline = false;
+          break;
+        }
+        assertTrue(EnvironmentEdgeManager.currentTime() < startTime + timeOut,
+          "Timed out in open the testing region");
       }
-      assertTrue("Timed out in open the testing region",
-        EnvironmentEdgeManager.currentTime() < startTime + timeOut);
+      assertFalse(offline);
+
+      // c0:0, c1:1
+      kvListExp = new ArrayList<>();
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[0], 0, VALUE));
+      kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[1], 1, VALUE));
+      result = scanner.next();
+      verifyResult(result, kvListExp, toLog, "Testing scan on re-opened region");
     }
-    assertFalse(offline);
-
-    // c0:0, c1:1
-    kvListExp = new ArrayList<>();
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[0], 0, VALUE));
-    kvListExp.add(new KeyValue(ROW, FAMILY, QUALIFIERS[1], 1, VALUE));
-    result = scanner.next();
-    verifyResult(result, kvListExp, toLog, "Testing scan on re-opened region");
   }
 
-  @Test
+  @TestTemplate
   public void testAsyncScannerWithSmallData() throws Exception {
-    testAsyncScanner(name.getTableName(), 2, 3, 10, -1, null);
+    testAsyncScanner(tableName, 2, 3, 10, -1, null);
   }
 
-  @Test
+  @TestTemplate
   public void testAsyncScannerWithManyRows() throws Exception {
-    testAsyncScanner(name.getTableName(), 30000, 1, 1, -1, null);
+    testAsyncScanner(tableName, 30000, 1, 1, -1, null);
   }
 
-  @Test
+  @TestTemplate
   public void testAsyncScannerWithoutCaching() throws Exception {
-    testAsyncScanner(name.getTableName(), 5, 1, 1, 1, (b) -> {
+    testAsyncScanner(tableName, 5, 1, 1, 1, (b) -> {
       try {
         TimeUnit.MILLISECONDS.sleep(500);
       } catch (InterruptedException ex) {
@@ -809,63 +719,62 @@ private void testAsyncScanner(TableName table, int rowNumber, int familyNumber,
     byte[][] families = makeNAsciiWithZeroPrefix(family, familyNumber);
     byte[][] qualifiers = makeNAsciiWithZeroPrefix(qualifier, qualifierNumber);
 
-    Table ht = TEST_UTIL.createTable(table, families);
-
-    boolean toLog = true;
-    List<Cell> kvListExp = new ArrayList<>();
+    TEST_UTIL.createTable(table, families);
+    try (Connection conn = getConnection(); Table ht = conn.getTable(table)) {
+      boolean toLog = true;
+      List<Cell> kvListExp = new ArrayList<>();
 
-    List<Put> puts = new ArrayList<>();
-    for (byte[] r : rows) {
-      Put put = new Put(r);
-      for (byte[] f : families) {
-        for (byte[] q : qualifiers) {
-          KeyValue kv = new KeyValue(r, f, q, 1, VALUE);
-          put.add(kv);
-          kvListExp.add(kv);
+      List<Put> puts = new ArrayList<>();
+      for (byte[] r : rows) {
+        Put put = new Put(r);
+        for (byte[] f : families) {
+          for (byte[] q : qualifiers) {
+            KeyValue kv = new KeyValue(r, f, q, 1, VALUE);
+            put.add(kv);
+            kvListExp.add(kv);
+          }
+        }
+        puts.add(put);
+        if (puts.size() > 1000) {
+          ht.put(puts);
+          puts.clear();
         }
       }
-      puts.add(put);
-      if (puts.size() > 1000) {
+      if (!puts.isEmpty()) {
         ht.put(puts);
         puts.clear();
       }
-    }
-    if (!puts.isEmpty()) {
-      ht.put(puts);
-      puts.clear();
-    }
 
-    Scan scan = new Scan();
-    scan.setAsyncPrefetch(true);
-    if (caching > 0) {
-      scan.setCaching(caching);
-    }
-    try (ResultScanner scanner = ht.getScanner(scan)) {
-      assertTrue("Not instance of async scanner", scanner instanceof ClientAsyncPrefetchScanner);
-      ((ClientAsyncPrefetchScanner) scanner).setPrefetchListener(listener);
-      List<Cell> kvListScan = new ArrayList<>();
-      Result result;
-      boolean first = true;
-      int actualRows = 0;
-      while ((result = scanner.next()) != null) {
-        ++actualRows;
-        // waiting for cache. see HBASE-17376
-        if (first) {
-          TimeUnit.SECONDS.sleep(1);
-          first = false;
-        }
-        for (Cell kv : result.listCells()) {
-          kvListScan.add(kv);
+      Scan scan = new Scan();
+      scan.setAsyncPrefetch(true);
+      if (caching > 0) {
+        scan.setCaching(caching);
+      }
+      try (ResultScanner scanner = ht.getScanner(scan)) {
+        assertTrue(scanner instanceof ClientAsyncPrefetchScanner, "Not instance of async scanner");
+        ((ClientAsyncPrefetchScanner) scanner).setPrefetchListener(listener);
+        List<Cell> kvListScan = new ArrayList<>();
+        Result result;
+        boolean first = true;
+        int actualRows = 0;
+        while ((result = scanner.next()) != null) {
+          ++actualRows;
+          // waiting for cache. see HBASE-17376
+          if (first) {
+            TimeUnit.SECONDS.sleep(1);
+            first = false;
+          }
+          for (Cell kv : result.listCells()) {
+            kvListScan.add(kv);
+          }
         }
+        assertEquals(rowNumber, actualRows);
+        // These cells may have different rows but it is ok. The Result#getRow
+        // isn't used in the verifyResult()
+        result = Result.create(kvListScan);
+        verifyResult(result, kvListExp, toLog, "Testing async scan");
       }
-      assertEquals(rowNumber, actualRows);
-      // These cells may have different rows but it is ok. The Result#getRow
-      // isn't used in the verifyResult()
-      result = Result.create(kvListScan);
-      verifyResult(result, kvListExp, toLog, "Testing async scan");
     }
-
-    TEST_UTIL.deleteTable(table);
   }
 
   private static byte[][] makeNAsciiWithZeroPrefix(byte[] base, int n) {
@@ -882,7 +791,6 @@ private static byte[][] makeNAsciiWithZeroPrefix(byte[] base, int n) {
   }
 
   static void verifyResult(Result result, List<Cell> expKvList, boolean toLog, String msg) {
-
     LOG.info(msg);
     LOG.info("Expected count: " + expKvList.size());
     LOG.info("Actual count: " + result.size());
@@ -901,18 +809,18 @@ static void verifyResult(Result result, List<Cell> expKvList, boolean toLog, Str
         LOG.info("get kv is: " + kv.toString());
         LOG.info("exp kv is: " + kvExp.toString());
       }
-      assertTrue("Not equal", kvExp.equals(kv));
+      assertTrue(kvExp.equals(kv), "Not equal");
     }
 
     assertEquals(expKvList.size(), result.size());
   }
 
-  @Test
+  @TestTemplate
   public void testReadExpiredDataForRawScan() throws IOException {
-    TableName tableName = name.getTableName();
     long ts = EnvironmentEdgeManager.currentTime() - 10000;
     byte[] value = Bytes.toBytes("expired");
-    try (Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName)) {
       table.put(new Put(ROW).addColumn(FAMILY, QUALIFIER, ts, value));
       assertArrayEquals(value, table.get(new Get(ROW)).getValue(FAMILY, QUALIFIER));
       TEST_UTIL.getAdmin().modifyColumnFamily(tableName,
@@ -927,11 +835,11 @@ public void testReadExpiredDataForRawScan() throws IOException {
     }
   }
 
-  @Test
+  @TestTemplate
   public void testScanWithColumnsAndFilterAndVersion() throws IOException {
-    TableName tableName = name.getTableName();
     long now = EnvironmentEdgeManager.currentTime();
-    try (Table table = TEST_UTIL.createTable(tableName, FAMILY, 4)) {
+    TEST_UTIL.createTable(tableName, FAMILY, 4);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName)) {
       for (int i = 0; i < 4; i++) {
         Put put = new Put(ROW);
         put.addColumn(FAMILY, QUALIFIER, now + i, VALUE);
@@ -950,10 +858,10 @@ public void testScanWithColumnsAndFilterAndVersion() throws IOException {
     }
   }
 
-  @Test
+  @TestTemplate
   public void testScanWithSameStartRowStopRow() throws IOException {
-    TableName tableName = name.getTableName();
-    try (Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName)) {
       table.put(new Put(ROW).addColumn(FAMILY, QUALIFIER, VALUE));
 
       Scan scan = new Scan().withStartRow(ROW).withStopRow(ROW);
@@ -987,14 +895,13 @@ public void testScanWithSameStartRowStopRow() throws IOException {
     }
   }
 
-  @Test
+  @TestTemplate
   public void testReverseScanWithFlush() throws Exception {
-    TableName tableName = name.getTableName();
     final int BATCH_SIZE = 10;
     final int ROWS_TO_INSERT = 100;
     final byte[] LARGE_VALUE = generateHugeValue(128 * 1024);
-
-    try (Table table = TEST_UTIL.createTable(tableName, FAMILY);
+    TEST_UTIL.createTable(tableName, FAMILY);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName);
       Admin admin = TEST_UTIL.getAdmin()) {
       List<Put> putList = new ArrayList<>();
       for (long i = 0; i < ROWS_TO_INSERT; i++) {
@@ -1020,19 +927,19 @@ public void testReverseScanWithFlush() throws Exception {
       int count = 0;
 
       try (ResultScanner results = table.getScanner(scan)) {
-        for (Result result : results) {
+        while (results.next() != null) {
           count++;
         }
       }
-      assertEquals("Expected " + ROWS_TO_INSERT + " rows in the table but it is " + count,
-        ROWS_TO_INSERT, count);
+      assertEquals(ROWS_TO_INSERT, count,
+        "Expected " + ROWS_TO_INSERT + " rows in the table but it is " + count);
     }
   }
 
-  @Test
+  @TestTemplate
   public void testScannerWithPartialResults() throws Exception {
-    TableName tableName = TableName.valueOf("testScannerWithPartialResults");
-    try (Table table = TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("c"), 4)) {
+    TEST_UTIL.createMultiRegionTable(tableName, Bytes.toBytes("c"), 4);
+    try (Connection conn = getConnection(); Table table = conn.getTable(tableName)) {
       List<Put> puts = new ArrayList<>();
       byte[] largeArray = new byte[10000];
       Put put = new Put(Bytes.toBytes("aaaa0"));