diff --git a/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java b/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java index 036ce265..6f95aefe 100644 --- a/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java +++ b/src/main/java/com/alipay/oceanbase/hbase/util/ObTableClientManager.java @@ -21,8 +21,7 @@ import com.alipay.oceanbase.rpc.constant.Constants; import com.google.common.base.Objects; import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.ConnectionConfiguration; + import java.io.IOException; import java.util.Map; @@ -38,7 +37,7 @@ public class ObTableClientManager { public static final ConcurrentHashMap OB_TABLE_CLIENT_LOCK = new ConcurrentHashMap(); - public static final Map OB_TABLE_CLIENT_INSTANCE = new ConcurrentHashMap(); + public static final Map OB_TABLE_CLIENT_INSTANCE = new ConcurrentHashMap<>(); public static ObTableClient getOrCreateObTableClient(OHConnectionConfiguration connectionConfig) throws IllegalArgumentException, @@ -58,22 +57,27 @@ public static ObTableClient getOrCreateObTableClient(OHConnectionConfiguration c obTableClientKey.setDatabase(connectionConfig.getDatabase()); } else { checkArgument(isNotBlank(connectionConfig.getParamUrl()), HBASE_OCEANBASE_PARAM_URL - + " is blank"); - obTableClientKey = new ObTableClientKey(); - String paramUrl = connectionConfig.getParamUrl(); - if (!paramUrl.contains("database")) { - paramUrl += "&database=default"; - } - obTableClientKey.setParamUrl(paramUrl); - obTableClientKey.setSysUserName(connectionConfig.getSysUsername()); - if (connectionConfig.getSysPassword() == null) { - obTableClientKey.setSysPassword(Constants.EMPTY_STRING); - } else { - obTableClientKey.setSysPassword(connectionConfig.getSysPassword()); - } + + " is blank"); + obTableClientKey = generateObTableClientKey(connectionConfig); + } + return getOrCreateObTableClient(obTableClientKey, connectionConfig.getRpcConnectTimeout()); + } + + public static ObTableClientKey generateObTableClientKey(OHConnectionConfiguration connectionConfig) { + ObTableClientKey obTableClientKey = new ObTableClientKey(); + String paramUrl = connectionConfig.getParamUrl(); + if (!paramUrl.contains("database")) { + paramUrl += "&database=default"; + } + obTableClientKey.setParamUrl(paramUrl); + obTableClientKey.setSysUserName(connectionConfig.getSysUsername()); + if (connectionConfig.getSysPassword() == null) { + obTableClientKey.setSysPassword(Constants.EMPTY_STRING); + } else { + obTableClientKey.setSysPassword(connectionConfig.getSysPassword()); } checkArgument(isNotBlank(connectionConfig.getFullUsername()), - HBASE_OCEANBASE_FULL_USER_NAME + " is blank"); + HBASE_OCEANBASE_FULL_USER_NAME + " is blank"); obTableClientKey.setFullUserName(connectionConfig.getFullUsername()); if (connectionConfig.getPassword() == null) { @@ -85,8 +89,8 @@ public static ObTableClient getOrCreateObTableClient(OHConnectionConfiguration c for (Map.Entry property : connectionConfig.getProperties().entrySet()) { obTableClientKey.getProperties().put(property.getKey(), property.getValue()); } - - return getOrCreateObTableClient(obTableClientKey, connectionConfig.getRpcConnectTimeout()); + + return obTableClientKey; } public static ObTableClient getOrCreateObTableClient(ObTableClientKey obTableClientKey, diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartAppendTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartAppendTest.java index d296ba19..122b4880 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartAppendTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartAppendTest.java @@ -21,6 +21,7 @@ import com.alipay.oceanbase.hbase.OHTableClient; import com.alipay.oceanbase.hbase.util.ObHTableTestUtil; import com.alipay.oceanbase.hbase.util.TableTemplateManager; +import com.google.common.base.Strings; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.util.Bytes; import org.junit.*; @@ -34,6 +35,7 @@ import static com.alipay.oceanbase.hbase.util.ObHTableSecondaryPartUtil.*; import static com.alipay.oceanbase.hbase.util.ObHTableTestUtil.FOR_EACH; +import static com.alipay.oceanbase.hbase.util.ObHTableTestUtil.getConnection; import static com.alipay.oceanbase.hbase.util.TableTemplateManager.NORMAL_TABLES; import static org.junit.Assert.*; @@ -53,7 +55,7 @@ public static void before() throws Exception { @AfterClass public static void finish() throws Exception { closeDistributedExecute(); - dropTables(tableNames, group2tableNames); +// dropTables(tableNames, group2tableNames); } @Before @@ -68,133 +70,139 @@ private static void assertNullResult(Result result) throws Exception { private static void testAppend(String tableName) throws Exception { OHTableClient hTable = ObHTableTestUtil.newOHTableClient(getTableName(tableName)); hTable.init(); + try { + byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); + byte[] ROW = "appendKey".getBytes(); + byte[] v1 = Bytes.toBytes("42"); + byte[] v2 = Bytes.toBytes("23"); + byte[][] QUALIFIERS = new byte[][]{Bytes.toBytes("b"), Bytes.toBytes("a"), + Bytes.toBytes("c")}; + Append a = new Append(ROW); + a.add(FAMILY, QUALIFIERS[0], v1); + a.add(FAMILY, QUALIFIERS[1], v2); + a.setReturnResults(false); + assertNullResult(hTable.append(a)); - byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); - byte[] ROW = "appendKey".getBytes(); - byte[] v1 = Bytes.toBytes("42"); - byte[] v2 = Bytes.toBytes("23"); - byte[][] QUALIFIERS = new byte[][] { Bytes.toBytes("b"), Bytes.toBytes("a"), - Bytes.toBytes("c") }; - Append a = new Append(ROW); - a.add(FAMILY, QUALIFIERS[0], v1); - a.add(FAMILY, QUALIFIERS[1], v2); - a.setReturnResults(false); - assertNullResult(hTable.append(a)); - - a = new Append(ROW); - a.add(FAMILY, QUALIFIERS[0], v2); - a.add(FAMILY, QUALIFIERS[1], v1); - a.add(FAMILY, QUALIFIERS[2], v2); - Result r = hTable.append(a); - assertEquals(0, Bytes.compareTo(Bytes.add(v1, v2), r.getValue(FAMILY, QUALIFIERS[0]))); - assertEquals(0, Bytes.compareTo(Bytes.add(v2, v1), r.getValue(FAMILY, QUALIFIERS[1]))); - // QUALIFIERS[2] previously not exist, verify both value and timestamp are correct - assertEquals(0, Bytes.compareTo(v2, r.getValue(FAMILY, QUALIFIERS[2]))); - assertEquals(r.getColumnLatest(FAMILY, QUALIFIERS[0]).getTimestamp(), - r.getColumnLatest(FAMILY, QUALIFIERS[2]).getTimestamp()); - - Get get = new Get(ROW); - get.setMaxVersions(10); - get.addFamily(FAMILY); - Result result = hTable.get(get); - assertEquals(2, result.getColumnCells(FAMILY, QUALIFIERS[0]).size()); - assertEquals(2, result.getColumnCells(FAMILY, QUALIFIERS[1]).size()); - assertEquals(1, result.getColumnCells(FAMILY, QUALIFIERS[2]).size()); - assertEquals( - 0, - Bytes.compareTo(Bytes.add(v1, v2), result.getColumnCells(FAMILY, QUALIFIERS[0]).get(0) - .getValue())); - assertEquals(0, - Bytes.compareTo(v2, result.getColumnCells(FAMILY, QUALIFIERS[2]).get(0).getValue())); + a = new Append(ROW); + a.add(FAMILY, QUALIFIERS[0], v2); + a.add(FAMILY, QUALIFIERS[1], v1); + a.add(FAMILY, QUALIFIERS[2], v2); + Result r = hTable.append(a); + assertEquals(0, Bytes.compareTo(Bytes.add(v1, v2), r.getValue(FAMILY, QUALIFIERS[0]))); + assertEquals(0, Bytes.compareTo(Bytes.add(v2, v1), r.getValue(FAMILY, QUALIFIERS[1]))); + // QUALIFIERS[2] previously not exist, verify both value and timestamp are correct + assertEquals(0, Bytes.compareTo(v2, r.getValue(FAMILY, QUALIFIERS[2]))); + assertEquals(r.getColumnLatest(FAMILY, QUALIFIERS[0]).getTimestamp(), + r.getColumnLatest(FAMILY, QUALIFIERS[2]).getTimestamp()); - hTable.close(); + Get get = new Get(ROW); + get.setMaxVersions(10); + get.addFamily(FAMILY); + Result result = hTable.get(get); + assertEquals(2, result.getColumnCells(FAMILY, QUALIFIERS[0]).size()); + assertEquals(2, result.getColumnCells(FAMILY, QUALIFIERS[1]).size()); + assertEquals(1, result.getColumnCells(FAMILY, QUALIFIERS[2]).size()); + assertEquals( + 0, + Bytes.compareTo(Bytes.add(v1, v2), result.getColumnCells(FAMILY, QUALIFIERS[0]).get(0) + .getValue())); + assertEquals(0, + Bytes.compareTo(v2, result.getColumnCells(FAMILY, QUALIFIERS[2]).get(0).getValue())); + } finally { + hTable.close(); + } } private static void testAppendBorder(String tableName) throws Exception { OHTableClient hTable = ObHTableTestUtil.newOHTableClient(getTableName(tableName)); hTable.init(); - - byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); - byte[] ROW = "appendKey".getBytes(); - byte[] v1 = Bytes.toBytes("ab"); - byte[][] QUALIFIERS = new byte[][] { Bytes.toBytes("b"), Bytes.toBytes("a"), - Bytes.toBytes("c") }; - Put put = new Put(ROW); - put.addColumn(FAMILY, QUALIFIERS[1], v1); - hTable.put(put); - Append a = new Append(ROW); - a.add(FAMILY, QUALIFIERS[1], v1); - a.add(FAMILY, QUALIFIERS[2], "".getBytes()); - hTable.append(a); - Get get = new Get(ROW); - get.setMaxVersions(10); - get.addFamily(FAMILY); - Result result = hTable.get(get); - assertEquals(3, result.size()); - - a = new Append(ROW); - a.add(FAMILY, QUALIFIERS[2], v1); - a.add(FAMILY, QUALIFIERS[2], "".getBytes()); - hTable.append(a); - get = new Get(ROW); - get.setMaxVersions(10); - get.addFamily(FAMILY); - result = hTable.get(get); - assertEquals(4, result.size()); - - byte[] randomBytes = new byte[1025]; - Random random = new Random(); - random.nextBytes(randomBytes); - a = new Append(ROW); - a.add(FAMILY, QUALIFIERS[2], randomBytes); try { + byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); + byte[] ROW = "appendKey".getBytes(); + byte[] v1 = Bytes.toBytes("ab"); + byte[][] QUALIFIERS = new byte[][]{Bytes.toBytes("b"), Bytes.toBytes("a"), + Bytes.toBytes("c")}; + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIERS[1], v1); + hTable.put(put); + Append a = new Append(ROW); + a.add(FAMILY, QUALIFIERS[1], v1); + a.add(FAMILY, QUALIFIERS[2], "".getBytes()); hTable.append(a); - fail("unexpect error, too long data should fail"); - } catch (IOException e) { - assertTrue(e.getCause().getMessage().contains("Data too long for column 'V'")); - } + Get get = new Get(ROW); + get.setMaxVersions(10); + get.addFamily(FAMILY); + Result result = hTable.get(get); + assertEquals(3, result.size()); - hTable.close(); + a = new Append(ROW); + a.add(FAMILY, QUALIFIERS[2], v1); + a.add(FAMILY, QUALIFIERS[2], "".getBytes()); + hTable.append(a); + get = new Get(ROW); + get.setMaxVersions(10); + get.addFamily(FAMILY); + result = hTable.get(get); + assertEquals(4, result.size()); + + byte[] randomBytes = new byte[1025]; + Random random = new Random(); + random.nextBytes(randomBytes); + a = new Append(ROW); + a.add(FAMILY, QUALIFIERS[2], randomBytes); + try { + hTable.append(a); + fail("unexpect error, too long data should fail"); + } catch (IOException e) { + assertTrue(e.getCause().getMessage().contains("Data too long for column 'V'")); + } + } finally { + hTable.close(); + } } private static void testAppendCon(String tableName) throws Exception { OHTableClient hTable = ObHTableTestUtil.newOHTableClient(getTableName(tableName)); - hTable.init(); - byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); - String column = "appColumn"; - byte[] ROW = "appendKey".getBytes(); - byte[] v = "a".getBytes(); - byte[] expect = "a".getBytes(); - ThreadPoolExecutor threadPoolExecutor = OHTable.createDefaultThreadPoolExecutor(1, 100,100); - AtomicInteger atomicInteger = new AtomicInteger(0); - CountDownLatch countDownLatch = new CountDownLatch(100); - for (int i = 0; i < 100; i++) { - Append append = new Append(ROW); - append.add(FAMILY, column.getBytes(), v); - threadPoolExecutor.submit(() -> { - try { - hTable.append(append); - atomicInteger.incrementAndGet(); - } catch (Exception e) { - if (!e.getCause().getMessage().contains("OB_TRY_LOCK_ROW_CONFLICT") && !e.getCause().getMessage().contains("OB_TIMEOUT")) { - throw new RuntimeException(e); + try { + hTable.init(); + byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); + String column = "appColumn"; + byte[] ROW = "appendKey".getBytes(); + byte[] v = "a".getBytes(); + ThreadPoolExecutor threadPoolExecutor = OHTable.createDefaultThreadPoolExecutor(1, 100, 100); + AtomicInteger atomicInteger = new AtomicInteger(0); + CountDownLatch countDownLatch = new CountDownLatch(100); + + for (int i = 0; i < 100; i++) { + Append append = new Append(ROW); + append.add(FAMILY, column.getBytes(), v); + threadPoolExecutor.submit(() -> { + try { + hTable.append(append); + } catch (Exception e) { + if (!e.getCause().getMessage().contains("OB_TRY_LOCK_ROW_CONFLICT") + && !e.getCause().getMessage().contains("OB_TIMEOUT")) { + throw new RuntimeException(e); + } + } finally { + atomicInteger.incrementAndGet(); + countDownLatch.countDown(); } - } finally { - countDownLatch.countDown(); - } - }); - } - countDownLatch.await(100000, TimeUnit.MILLISECONDS); - for (int i = 0; i < atomicInteger.get() - 1; i++) { - expect = Bytes.add(expect, v); + }); + } + threadPoolExecutor.shutdown(); + countDownLatch.await(100000, TimeUnit.MILLISECONDS); + final byte[] expect = Strings.repeat("a", atomicInteger.get()).getBytes(); + System.out.println("atomicInteger: " + atomicInteger.get()); + Get get = new Get(ROW); + get.setMaxVersions(1); + get.addColumn(FAMILY, column.getBytes()); + Result result = hTable.get(get); + ObHTableTestUtil.Assert(tableName, ()-> assertTrue(0 <= Bytes.compareTo(expect, result.getColumnCells(FAMILY, column.getBytes()).get(0).getValue()))); + } finally { + hTable.close(); } - Get get = new Get(ROW); - get.setMaxVersions(1); - get.addColumn(FAMILY, column.getBytes()); - Result result = hTable.get(get); - assertEquals(0, Bytes.compareTo(expect, result.getColumnCells(FAMILY, column.getBytes()).get(0).getValue())); - hTable.close(); } private static void testAppendMultiCF(Map.Entry> entry) throws Exception { @@ -260,8 +268,8 @@ public void testAppendMultiCF() throws Throwable { @Test public void testAppendSeires() throws Throwable { - createTables(TableTemplateManager.TableType.SECONDARY_PARTITIONED_TIME_RANGE_KEY, series_tables, group2tableNames, true); + createTables(TableTemplateManager.TableType.SECONDARY_PARTITIONED_TIME_RANGE_KEY, series_tables, null, true); FOR_EACH(series_tables, OHTableSecondaryPartAppendTest::testAppendSeires); - dropTables(series_tables, group2tableNames); + dropTables(series_tables, null); } } diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchGetTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchGetTest.java index dc81b7ed..77d573ec 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchGetTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchGetTest.java @@ -42,7 +42,7 @@ public class OHTableSecondaryPartBatchGetTest { @BeforeClass public static void before() throws Exception { openDistributedExecute(); - for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_AND_SERIES_TABLES) { + for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_TABLES) { createTables(type, tableNames, group2tableNames, true); } } @@ -94,8 +94,9 @@ public static void testBatchGetImpl(String tableName) throws Exception { System.out.println("Column: " + Q + ", Value: " + V); } } + hTable.close(); } - + @Test public void testBatchGet() throws Throwable { FOR_EACH(tableNames, OHTableSecondaryPartBatchGetTest::testBatchGetImpl); diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchTest.java index d6ffc34e..1295a00a 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartBatchTest.java @@ -439,24 +439,20 @@ public static void testMultiCFBatchGetImpl(Map.Entry> entry @Test public void testBatchPut() throws Throwable { FOR_EACH(tableNames, OHTableSecondaryPartBatchTest::testBatchPutImpl); - truncateTables(ObHTableTestUtil.getConnection(), tableNames); } @Test public void testMultiCFPut() throws Throwable { FOR_EACH(group2tableNames, OHTableSecondaryPartBatchTest::testMultiCFBatchPutImpl); - truncateTables(ObHTableTestUtil.getConnection(), group2tableNames); } @Test public void testBatchGet() throws Throwable { FOR_EACH(tableNames, OHTableSecondaryPartBatchTest::testBatchGetImpl); - truncateTables(ObHTableTestUtil.getConnection(), tableNames); } @Test public void testMultiCFGet() throws Throwable { FOR_EACH(group2tableNames, OHTableSecondaryPartBatchTest::testMultiCFBatchGetImpl); - truncateTables(ObHTableTestUtil.getConnection(), group2tableNames); } } diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCellTTLTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCellTTLTest.java index 8837f527..3a48d867 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCellTTLTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCellTTLTest.java @@ -27,10 +27,7 @@ import org.apache.hadoop.hbase.util.Bytes; import org.junit.*; -import java.util.LinkedHashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; +import java.util.*; import static com.alipay.oceanbase.hbase.util.ObHTableSecondaryPartUtil.*; import static com.alipay.oceanbase.hbase.util.ObHTableTestUtil.FOR_EACH; @@ -40,7 +37,6 @@ public class OHTableSecondaryPartCellTTLTest { private static List tableNames = new LinkedList(); - private static List NonTTLTable = new LinkedList(); private static Map> group2tableNames = new LinkedHashMap>(); @BeforeClass @@ -54,7 +50,7 @@ public static void before() throws Exception { @AfterClass public static void finish() throws Exception { closeDistributedExecute(); - // dropTables(tableNames, group2tableNames); + dropTables(tableNames, group2tableNames); } @Before @@ -389,7 +385,9 @@ public static void testCellTTLSQL(String tableName) throws Exception { Thread.sleep(10000); enableTTL(); triggerTTL(); - checkUtilTimeout(()-> { + List tableNames = new ArrayList<>(); + tableNames.add(tableName); + checkUtilTimeout(tableNames, ()-> { try { return getRunningNormalTTLTaskCnt() == 0; } catch (Exception e) { @@ -479,8 +477,10 @@ public void testCellTTLWithRowkeyTTL() throws Throwable { @Test public void testCellTTLWithNonTTLTable() throws Throwable { - createTables(TableTemplateManager.TableType.NON_PARTITIONED_REGULAR, NonTTLTable, group2tableNames, true); + List NonTTLTable = new LinkedList(); + Map> NonTTLTableMultiCF = new LinkedHashMap<>(); + createTables(TableTemplateManager.TableType.NON_PARTITIONED_REGULAR, NonTTLTable, NonTTLTableMultiCF, true); FOR_EACH(NonTTLTable, OHTableSecondaryPartCellTTLTest::testCellTTLWithNonTTLTable); - dropTables(NonTTLTable, group2tableNames); + dropTables(NonTTLTable, NonTTLTableMultiCF); } } diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCheckAndMutateTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCheckAndMutateTest.java index 9dac671a..f36898c1 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCheckAndMutateTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartCheckAndMutateTest.java @@ -40,7 +40,6 @@ public class OHTableSecondaryPartCheckAndMutateTest { private static List tableNames = new LinkedList(); private static Map> group2tableNames = new LinkedHashMap>(); - private static List series_tables = new LinkedList(); private static byte[] ROW = Bytes.toBytes("testRow"); private static byte[] QUALIFIER = Bytes.toBytes("testQualifier"); private static byte[] VALUE_1 = Bytes.toBytes("testValue"); @@ -50,7 +49,7 @@ public class OHTableSecondaryPartCheckAndMutateTest { @BeforeClass public static void before() throws Exception { openDistributedExecute(); - for (TableTemplateManager.TableType type : NORMAL_TABLES) { + for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_TABLES) { createTables(type, tableNames, group2tableNames, true); } } @@ -244,6 +243,7 @@ public static void testCheckAndPutWithCompareOp(String tableName) throws Throwab ok = hTable .checkAndPut(ROW, FAMILY, QUALIFIER, CompareFilter.CompareOp.EQUAL, value2, put3); assertEquals(ok, true); + hTable.close(); } public static void testCheckAndDeleteWithCompareOp(String tableName) throws Throwable { @@ -339,6 +339,7 @@ public static void testCheckAndDeleteWithCompareOp(String tableName) throws Thro ok = hTable.checkAndDelete(ROW, FAMILY, QUALIFIER, CompareFilter.CompareOp.EQUAL, value2, delete); assertEquals(ok, true); + hTable.close(); } private static void testCheckAndMutateMultiCF(Map.Entry> entry) @@ -436,8 +437,10 @@ public void testCheckAndMutateMultiCF() throws Throwable { @Test public void testCheckAndMutateSeires() throws Throwable { - createTables(TableTemplateManager.TableType.SECONDARY_PARTITIONED_TIME_RANGE_KEY, series_tables, group2tableNames, true); - FOR_EACH(series_tables, OHTableSecondaryPartCheckAndMutateTest::testCheckAndMutateSeires); - dropTables(series_tables, group2tableNames); + List seriesTables = new LinkedList(); + Map> seriesTablesMultiCF = new LinkedHashMap>(); + createTables(TableTemplateManager.TableType.SECONDARY_PARTITIONED_TIME_RANGE_KEY, seriesTables, seriesTablesMultiCF, true); + FOR_EACH(seriesTables, OHTableSecondaryPartCheckAndMutateTest::testCheckAndMutateSeires); + dropTables(seriesTables, seriesTablesMultiCF); } } diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartDeleteTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartDeleteTest.java index 5a187817..73172f1f 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartDeleteTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartDeleteTest.java @@ -91,6 +91,7 @@ public static void testDeleteLastVersionImpl(String tableName) throws Exception Assert.assertTrue(cell.getTimestamp() != ts4); } } + hTable.close(); } public static void testDeleteSpecifiedImpl(String tableName) throws Exception { @@ -129,6 +130,7 @@ public static void testDeleteSpecifiedImpl(String tableName) throws Exception { Assert(tableName, ()->Assert.assertTrue(cell.getTimestamp() != ts1)); } } + hTable.close(); } public static void testDeleteColumnImpl(String tableName) throws Exception { @@ -167,6 +169,7 @@ public static void testDeleteColumnImpl(String tableName) throws Exception { Assert.assertTrue(cell.getTimestamp() != ts1); } } + hTable.close(); } public static void testDeleteFamilyImpl(String tableName) throws Exception { @@ -201,6 +204,7 @@ public static void testDeleteFamilyImpl(String tableName) throws Exception { Result result = hTable.get(get); Assert.assertEquals(0, result.size()); } + hTable.close(); } public static void testDeleteFamilyVersionImpl(String tableName) throws Exception { @@ -239,6 +243,7 @@ public static void testDeleteFamilyVersionImpl(String tableName) throws Exceptio Assert(tableName, ()->Assert.assertTrue(cell.getTimestamp() != ts2)); } } + hTable.close(); } public static void testMultiCFDeleteLastVersionImpl(Map.Entry> entry) throws Exception { @@ -284,6 +289,7 @@ public static void testMultiCFDeleteLastVersionImpl(Map.EntryAssert.assertTrue("should not found last version ts4", cell.getTimestamp() != ts4)); } } + hTable.close(); } public static void testMultiCFDeleteSpecifiedImpl(Map.Entry> entry) throws Exception { @@ -325,6 +331,7 @@ public static void testMultiCFDeleteSpecifiedImpl(Map.Entry Assert(tableName, ()->Assert.assertTrue("should not found last version ts2",cell.getTimestamp() != ts2)); } } + hTable.close(); } public static void testMultiCFDeleteColumnImpl(Map.Entry> entry) throws Exception { @@ -366,6 +373,7 @@ public static void testMultiCFDeleteColumnImpl(Map.Entry> e Assert(tableName, ()->Assert.assertTrue("should not found last version ts2",cell.getTimestamp() != ts2)); } } + hTable.close(); } public static void testMultiCFDeleteFamilyImpl(Map.Entry> entry) @@ -402,6 +410,7 @@ public static void testMultiCFDeleteFamilyImpl(Map.Entry> e } Result result = hTable.get(get); Assert.assertEquals(0, result.size()); + hTable.close(); } public static void testMultiCFDeleteFamilyVersionImpl(Map.Entry> entry) throws Exception { @@ -438,6 +447,7 @@ public static void testMultiCFDeleteFamilyVersionImpl(Map.EntryAssert.assertEquals(6, result.size())); + hTable.close(); } public static void testDeleteAllImpl(Map.Entry> entry) throws Exception { @@ -473,6 +483,7 @@ public static void testDeleteAllImpl(Map.Entry> entry) thro } Result result = hTable.get(get); Assert(entry.getValue(), ()->Assert.assertEquals(0, result.size())); + hTable.close(); } @Test diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartGetTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartGetTest.java index b2ae0b0c..b6eb6eb8 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartGetTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartGetTest.java @@ -49,10 +49,8 @@ public class OHTableSecondaryPartGetTest { @BeforeClass public static void before() throws Exception { openDistributedExecute(); - for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_AND_SERIES_TABLES) { - if (!type.name().contains("TIME")) { - createTables(type, tableNames, group2tableNames, true); - } + for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_TABLES) { + createTables(type, tableNames, group2tableNames, true); } } @@ -291,6 +289,7 @@ public static void testMultiCFGetImpl(Map.Entry> entry) thr } } } + hTable.close(); } @Test diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartIncrementTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartIncrementTest.java index 89e8041c..04b65f02 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartIncrementTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartIncrementTest.java @@ -41,7 +41,6 @@ public class OHTableSecondaryPartIncrementTest { private static List tableNames = new LinkedList(); - private static List series_tables = new LinkedList(); private static Map> group2tableNames = new LinkedHashMap>(); @BeforeClass @@ -103,89 +102,94 @@ public static void testIncrement(String tableName) throws Exception { private static void testIncBorder(String tableName) throws Exception { OHTableClient hTable = ObHTableTestUtil.newOHTableClient(getTableName(tableName)); hTable.init(); - - byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); - byte[] ROW = "incKey".getBytes(); - byte[] v1 = Bytes.toBytes("ab"); - byte[][] QUALIFIERS = new byte[][] { Bytes.toBytes("b"), Bytes.toBytes("a"), - Bytes.toBytes("c") }; - Put put = new Put(ROW); - put.addColumn(FAMILY, QUALIFIERS[1], v1); - hTable.put(put); - Increment inc = new Increment(ROW); - inc.addColumn(FAMILY, QUALIFIERS[1], 2L); try { + byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); + byte[] ROW = "incKey".getBytes(); + byte[] v1 = Bytes.toBytes("ab"); + byte[][] QUALIFIERS = new byte[][]{Bytes.toBytes("b"), Bytes.toBytes("a"), + Bytes.toBytes("c")}; + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIERS[1], v1); + hTable.put(put); + Increment inc = new Increment(ROW); + inc.addColumn(FAMILY, QUALIFIERS[1], 2L); + try { + hTable.increment(inc); + fail("unexpect error, increment only support long value type"); + } catch (Exception e) { + assertTrue(e.getCause().getMessage().contains("OB_KV_HBASE_INCR_FIELD_IS_NOT_LONG")); + } + Get get = new Get(ROW); + get.setMaxVersions(10); + get.addFamily(FAMILY); + Result result = hTable.get(get); + assertEquals(1, result.size()); + byte[] ROW1 = "incKey1".getBytes(); + inc = new Increment(ROW1); + inc.addColumn(FAMILY, QUALIFIERS[1], 2L); hTable.increment(inc); - fail("unexpect error, increment only support long value type"); - } catch (Exception e) { - assertTrue(e.getCause().getMessage().contains("OB_KV_HBASE_INCR_FIELD_IS_NOT_LONG")); + get = new Get(ROW1); + get.setMaxVersions(10); + get.addFamily(FAMILY); + result = hTable.get(get); + assertEquals(1, result.size()); + assertEquals(2L, Bytes.toLong(result.raw()[0].getValue())); + inc.addColumn(FAMILY, QUALIFIERS[0], 2L); + hTable.increment(inc); + get.setMaxVersions(10); + get.addFamily(FAMILY); + result = hTable.get(get); + assertEquals(3, result.size()); + assertEquals(4L, + Bytes.toLong(result.getColumnCells(FAMILY, QUALIFIERS[1]).get(0).getValue())); + assertEquals(2L, + Bytes.toLong(result.getColumnCells(FAMILY, QUALIFIERS[1]).get(1).getValue())); + assertEquals(2L, + Bytes.toLong(result.getColumnCells(FAMILY, QUALIFIERS[0]).get(0).getValue())); + } finally { + hTable.close(); } - Get get = new Get(ROW); - get.setMaxVersions(10); - get.addFamily(FAMILY); - Result result = hTable.get(get); - assertEquals(1, result.size()); - byte[] ROW1 = "incKey1".getBytes(); - inc = new Increment(ROW1); - inc.addColumn(FAMILY, QUALIFIERS[1], 2L); - hTable.increment(inc); - get = new Get(ROW1); - get.setMaxVersions(10); - get.addFamily(FAMILY); - result = hTable.get(get); - assertEquals(1, result.size()); - assertEquals(2L, Bytes.toLong(result.raw()[0].getValue())); - inc.addColumn(FAMILY, QUALIFIERS[0], 2L); - hTable.increment(inc); - get.setMaxVersions(10); - get.addFamily(FAMILY); - result = hTable.get(get); - assertEquals(3, result.size()); - assertEquals(4L, - Bytes.toLong(result.getColumnCells(FAMILY, QUALIFIERS[1]).get(0).getValue())); - assertEquals(2L, - Bytes.toLong(result.getColumnCells(FAMILY, QUALIFIERS[1]).get(1).getValue())); - assertEquals(2L, - Bytes.toLong(result.getColumnCells(FAMILY, QUALIFIERS[0]).get(0).getValue())); - hTable.close(); } private static void testIncCon(String tableName) throws Exception { OHTableClient hTable = ObHTableTestUtil.newOHTableClient(getTableName(tableName)); hTable.init(); - byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); - String column = "incColumn"; - byte[] ROW = "incKey".getBytes(); - long expect = 0; - ThreadPoolExecutor threadPoolExecutor = OHTable.createDefaultThreadPoolExecutor(1, 100,100); - AtomicInteger atomicInteger = new AtomicInteger(0); - CountDownLatch countDownLatch = new CountDownLatch(100); - for (int i = 0; i < 100; i++) { - Increment inc = new Increment(ROW); - inc.addColumn(FAMILY, column.getBytes(), 2L); - threadPoolExecutor.submit(() -> { - try { - hTable.increment(inc); - atomicInteger.incrementAndGet(); - } catch (Exception e) { - if (!e.getCause().getMessage().contains("OB_TRY_LOCK_ROW_CONFLICT") && !e.getCause().getMessage().contains("OB_TIMEOUT")) { - throw new RuntimeException(e); + try { + byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); + String column = "incColumn"; + byte[] ROW = "incKey".getBytes(); + long expect = 0; + ThreadPoolExecutor threadPoolExecutor = OHTable.createDefaultThreadPoolExecutor(1, 100, 100); + AtomicInteger atomicInteger = new AtomicInteger(0); + CountDownLatch countDownLatch = new CountDownLatch(100); + for (int i = 0; i < 100; i++) { + Increment inc = new Increment(ROW); + inc.addColumn(FAMILY, column.getBytes(), 2L); + threadPoolExecutor.submit(() -> { + try { + hTable.increment(inc); + atomicInteger.incrementAndGet(); + } catch (Exception e) { + if (!e.getCause().getMessage().contains("OB_TRY_LOCK_ROW_CONFLICT") && !e.getCause().getMessage().contains("OB_TIMEOUT")) { + throw new RuntimeException(e); + } + } finally { + countDownLatch.countDown(); } - } finally { - countDownLatch.countDown(); - } - }); - } - countDownLatch.await(100000, TimeUnit.MILLISECONDS); - for (int i = 0; i < atomicInteger.get(); i++) { - expect += 2; + }); + } + countDownLatch.await(100000, TimeUnit.MILLISECONDS); + for (int i = 0; i < atomicInteger.get(); i++) { + expect += 2; + } + Get get = new Get(ROW); + get.setMaxVersions(1); + get.addColumn(FAMILY, column.getBytes()); + Result result = hTable.get(get); + assertEquals(expect, Bytes.toLong(result.getColumnCells(FAMILY, column.getBytes()).get(0).getValue())); + } finally { + hTable.close(); } - Get get = new Get(ROW); - get.setMaxVersions(1); - get.addColumn(FAMILY, column.getBytes()); - Result result = hTable.get(get); - assertEquals(expect, Bytes.toLong(result.getColumnCells(FAMILY, column.getBytes()).get(0).getValue())); - hTable.close(); } private static void testIncrementMultiCF(Map.Entry> entry) @@ -193,22 +197,25 @@ private static void testIncrementMultiCF(Map.Entry> entry) String groupName = getTableName(entry.getKey()); OHTableClient hTable = ObHTableTestUtil.newOHTableClient(groupName); hTable.init(); - List tableNames = entry.getValue(); - String column = "appColumn"; - byte[] ROW = "appendKey".getBytes(); - Long v = 11L; - Increment increment = new Increment(ROW); - for (String tableName : tableNames) { - byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); - increment.addColumn(FAMILY, column.getBytes(), v); - } try { - hTable.increment(increment); - fail("unexpect error, increment should not support multi cf"); - } catch (Exception e) { - assertTrue(e.getMessage().contains("multi family is not supported")); + List tableNames = entry.getValue(); + String column = "appColumn"; + byte[] ROW = "appendKey".getBytes(); + Long v = 11L; + Increment increment = new Increment(ROW); + for (String tableName : tableNames) { + byte[] FAMILY = getColumnFamilyName(tableName).getBytes(); + increment.addColumn(FAMILY, column.getBytes(), v); + } + try { + hTable.increment(increment); + fail("unexpect error, increment should not support multi cf"); + } catch (Exception e) { + assertTrue(e.getMessage().contains("multi family is not supported")); + } + } finally { + hTable.close(); } - hTable.close(); } private static void testIncrementSeires(String tableName) throws Exception { @@ -252,8 +259,9 @@ public void testIncrementMultiCF() throws Throwable { @Test public void testIncrementSeires() throws Throwable { - createTables(TableTemplateManager.TableType.SECONDARY_PARTITIONED_TIME_RANGE_KEY, series_tables, group2tableNames, true); + List series_tables = new LinkedList(); + createTables(TableTemplateManager.TableType.SECONDARY_PARTITIONED_TIME_RANGE_KEY, series_tables, null, true); FOR_EACH(series_tables, OHTableSecondaryPartIncrementTest::testIncrementSeires); - dropTables(series_tables, group2tableNames); + dropTables(series_tables, null); } } diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartPutTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartPutTest.java index cb446782..69c0ac09 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartPutTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartPutTest.java @@ -40,7 +40,7 @@ public class OHTableSecondaryPartPutTest { @BeforeClass public static void before() throws Exception { openDistributedExecute(); - for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_AND_SERIES_TABLES) { + for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_TABLES) { createTables(type, tableNames, group2tableNames, true); } } @@ -67,9 +67,10 @@ public static void testPutImpl(String tableName) throws Exception { String value = "value"; { // put new key and get + long timestamp = System.currentTimeMillis(); Put put = new Put(toBytes(key)); - put.add(family.getBytes(), column1.getBytes(), toBytes(column1 + value)); - put.add(family.getBytes(), column2.getBytes(), toBytes(column2 + value)); + put.add(family.getBytes(), column1.getBytes(), timestamp, toBytes(column1 + value)); + put.add(family.getBytes(), column2.getBytes(), timestamp, toBytes(column2 + value)); hTable.put(put); Get get = new Get(toBytes(key)); @@ -196,8 +197,8 @@ public static void testMultiCFPutImpl(Map.Entry> entry) thr Get get = new Get(toBytes(key)); for (String tableName : entry.getValue()) { String family = getColumnFamilyName(tableName); - put.add(family.getBytes(), column1.getBytes(), toBytes(column1 + value)); - put.add(family.getBytes(), column2.getBytes(), toBytes(column2 + value)); + put.add(family.getBytes(), column1.getBytes(), currentTime, toBytes(column1 + value)); + put.add(family.getBytes(), column2.getBytes(), currentTime, toBytes(column2 + value)); get.addColumn(family.getBytes(), column1.getBytes()); get.addColumn(family.getBytes(), column2.getBytes()); } @@ -231,7 +232,6 @@ public static void testMultiCFPutImpl(Map.Entry> entry) thr } for (String tableName : entry.getValue()) { String family = getColumnFamilyName(tableName); - // TODO: Get/Scan返回的结果Q 带了cf, 这里预期跑不过 Assert(entry.getValue(), () -> Assert.assertTrue(secureCompare((column1 + value).getBytes(), r.getValue(family.getBytes(), column1.getBytes())))); Assert(entry.getValue(), () -> Assert.assertTrue(secureCompare((column2 + value).getBytes(), r.getValue(family.getBytes(), column2.getBytes())))); } diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartScanTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartScanTest.java index eaafd1e3..257793ec 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartScanTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartScanTest.java @@ -41,10 +41,8 @@ public class OHTableSecondaryPartScanTest { @BeforeClass public static void before() throws Exception { openDistributedExecute(); - for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_AND_SERIES_TABLES) { - if (!type.name().contains("TIME")) { - createTables(type, tableNames, group2tableNames, true); - } + for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_TABLES) { + createTables(type, tableNames, group2tableNames, true); } } @@ -226,21 +224,22 @@ public static void testScanImpl(String tableName) throws Exception { // 8. scan in reverse { - // Scan scan = new Scan(keys[2].getBytes(), keys[0].getBytes()); - // scan.addFamily(family.getBytes()); - // scan.setReversed(true); - // ResultScanner scanner = hTable.getScanner(scan); - // List cells = getCellsFromScanner(scanner); - // - // int cellIndex = 0; - // for (int i = 1; i >= 0; i--) { - // for (String column : columns) { - // AssertKeyValue(keys[i], column, lastTs, latestValue, cells.get(cellIndex)); - // cellIndex++; - // } - // } - // assertEquals(columns.length * 2, cells.size()); + Scan scan = new Scan(keys[1].getBytes(), "putKey".getBytes()); + scan.addFamily(family.getBytes()); + scan.setReversed(true); + ResultScanner scanner = hTable.getScanner(scan); + List cells = getCellsFromScanner(scanner); + + int cellIndex = 0; + for (int i = 1; i >= 0; i--) { + for (String column : columns) { + AssertKeyValue(keys[i], column, lastTs, latestValue, cells.get(cellIndex)); + cellIndex++; + } + } + assertEquals(columns.length * 2, cells.size()); } + hTable.close(); } public static void testMultiCFScanImpl(Map.Entry> entry) throws Exception { @@ -492,6 +491,7 @@ public static void testMultiCFScanImpl(Map.Entry> entry) th } } } + hTable.close(); } @Test diff --git a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartTTLTest.java b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartTTLTest.java index be3f526d..08d600df 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartTTLTest.java +++ b/src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartTTLTest.java @@ -22,10 +22,8 @@ public class OHTableSecondaryPartTTLTest { @BeforeClass public static void before() throws Exception { openDistributedExecute(); - for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_AND_SERIES_TABLES) { - if (!type.name().contains("TIME")) { - createTables(type, tableNames, group2tableNames, true); - } + for (TableTemplateManager.TableType type : TableTemplateManager.NORMAL_TABLES) { + createTables(type, tableNames, group2tableNames, true); } alterTableTimeToLive(tableNames, true, 10); for (List groupTableNames : group2tableNames.values()) { @@ -64,6 +62,7 @@ public static void testTTLImpl(List tableNames) throws Exception { } } } + hTable.close(); } // 1. sleep util data expired @@ -84,6 +83,7 @@ public static void testTTLImpl(List tableNames) throws Exception { get.addFamily(family.getBytes()); Result result = hTable.get(get); Assert.assertEquals(0, result.rawCells().length); + hTable.close(); } // 3. using sql to scan expired but not delete yet hbase data @@ -97,7 +97,7 @@ public static void testTTLImpl(List tableNames) throws Exception { triggerTTL(); // 5. check util expired hbase data is deleted by ttl tasks - checkUtilTimeout(()-> { + checkUtilTimeout(tableNames, ()-> { try { return getRunningNormalTTLTaskCnt() == 0; } catch (Exception e) { @@ -139,7 +139,7 @@ public static void testMultiCFTTLImpl(Map> group2tableNames } } } - + hTable.close(); } // 1. sleep util data expired @@ -158,6 +158,7 @@ public static void testMultiCFTTLImpl(Map> group2tableNames Get get = new Get(keys[0].getBytes()); Result result = hTable.get(get); Assert.assertEquals(0, result.rawCells().length); + hTable.close(); } // 3. using sql to scan expired but not delete yet hbase data @@ -171,7 +172,7 @@ public static void testMultiCFTTLImpl(Map> group2tableNames triggerTTL(); // 5. check util expired hbase data is deleted by ttl tasks - checkUtilTimeout(()-> { + checkUtilTimeout(allTableNames, ()-> { try { return getRunningNormalTTLTaskCnt() == 0; } catch (Exception e) { @@ -192,6 +193,7 @@ void testRowkeyTTL(List tableNames, Boolean useScan, Boolean isReversed) // 0. prepare data String keys[] = {"putKey1", "putKey2", "putKey3"}; String endKey = "putKey4"; + String reversedEndKey = "putKey"; String columns[] = {"putColumn1", "putColumn2"}; String values[] = {"putValue1", "putValue2", "putValue3", "putValue4"}; for (String tableName : tableNames) { @@ -202,11 +204,12 @@ void testRowkeyTTL(List tableNames, Boolean useScan, Boolean isReversed) for (String column : columns) { for (int i = 0; i < values.length; i++) { Put put = new Put(toBytes(key)); - put.add(family.getBytes(), column.getBytes(), values[i].getBytes()); + put.add(family.getBytes(), column.getBytes(), values[i].getBytes()); hTable.put(put); } } } + hTable.close(); } // 1. sleep util data expired @@ -217,8 +220,14 @@ void testRowkeyTTL(List tableNames, Boolean useScan, Boolean isReversed) // 3. SQL can scan expired but not delete yet hbase data for (String tableName : tableNames) { - Assert.assertEquals(keys.length * columns.length * values.length, - getSQLTableRowCnt(tableName)); + ObHTableTestUtil.Assert(tableName, ()-> { + try { + Assert.assertEquals(keys.length * columns.length * values.length, + getSQLTableRowCnt(tableName)); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); } // 4. use Hbase scan/get expired data to trigger ttl @@ -227,9 +236,16 @@ void testRowkeyTTL(List tableNames, Boolean useScan, Boolean isReversed) hTable.init(); String family = getColumnFamilyName(tableName); if (useScan) { - Scan scan = new Scan(keys[0].getBytes(), endKey.getBytes()); + Scan scan = new Scan(); + if (isReversed) { + scan.setReversed(true); + scan.setStartRow(keys[2].getBytes()); + scan.setStopRow(reversedEndKey.getBytes()); + } else { + scan.setStartRow(keys[0].getBytes()); + scan.setStopRow(endKey.getBytes()); + } scan.addFamily(family.getBytes()); - scan.setReversed(isReversed); ResultScanner scanner = hTable.getScanner(scan); List cells = getCellsFromScanner(scanner); assertEquals(0, cells.size()); @@ -241,10 +257,11 @@ void testRowkeyTTL(List tableNames, Boolean useScan, Boolean isReversed) assertEquals(0, result.rawCells().length); } } + hTable.close(); } // 5. wait to disable - checkUtilTimeout(()-> { + checkUtilTimeout(tableNames, ()-> { try { Boolean passed = true; for (int i = 0; passed && i < tableNames.size(); i++) { @@ -286,6 +303,7 @@ void testMultiCFRowkeyTTL(Map> group2tableNames, Boolean us } } } + hTable.close(); } // 1. sleep util data expired @@ -318,10 +336,11 @@ void testMultiCFRowkeyTTL(Map> group2tableNames, Boolean us assertEquals(0, result.rawCells().length); } } + hTable.close(); } // 5. wait to disable - checkUtilTimeout(()-> { + checkUtilTimeout(allTableNames, ()-> { try { Boolean passed = true; for (int i = 0; passed && i < allTableNames.size(); i++) { @@ -354,7 +373,7 @@ public void testRowkeyTTL() throws Exception { testRowkeyTTL(tableNames, true, false); testRowkeyTTL(tableNames, false, false); // TODO: open the test after reverse scan is ok - // testRowkeyTTL(tableNames, true); +// testRowkeyTTL(tableNames, true, true); } @Test diff --git a/src/test/java/com/alipay/oceanbase/hbase/util/ObHTableSecondaryPartUtil.java b/src/test/java/com/alipay/oceanbase/hbase/util/ObHTableSecondaryPartUtil.java index 26370b94..fc24e3f6 100644 --- a/src/test/java/com/alipay/oceanbase/hbase/util/ObHTableSecondaryPartUtil.java +++ b/src/test/java/com/alipay/oceanbase/hbase/util/ObHTableSecondaryPartUtil.java @@ -92,7 +92,16 @@ public static void createTables(Connection conn, TableTemplateManager.TableType TimeGenerator.TimeRange timeRange = TimeGenerator.generateTestTimeRange(); String tableGroup = TableTemplateManager.getTableGroupName(type, true); String tableGroupSql = TableTemplateManager.generateTableGroupSQL(tableGroup); - conn.createStatement().execute(tableGroupSql); + try { + conn.createStatement().execute(tableGroupSql); + System.out.println("============= create table_group: " + getTableName(tableGroup) + " =============\n" + (printSql ? tableGroupSql : "") + " \n============= done =============\n"); + } catch (SQLSyntaxErrorException e) { + if (!e.getMessage().contains("already exists")) { + throw e; + } else { + System.out.println("============= table_group: " + getTableName(tableGroup) + " already exist ============="); + } + } group2tableNames.put(tableGroup, new LinkedList<>()); for (int i = 1; i <= 3; ++i) { String tableName = TableTemplateManager.generateTableName(tableGroup, true, i); @@ -267,6 +276,7 @@ public static int getSQLTableRowCnt(String tableName) throws Exception { if (resultSet.next()) { rowCnt = resultSet.getInt(1); } + System.out.println("============= rowCnt: " + rowCnt + " ============="); return rowCnt; } @@ -333,8 +343,8 @@ public static List getCellsFromScanner(ResultScanner scanner) { return cells; } - public static void checkUtilTimeout(Supplier function, long timeout, long interval) - throws Exception { + + public static void checkUtilTimeout(List tableNames, Supplier function, long timeout, long interval) throws Exception { long startTime = System.currentTimeMillis(); while (System.currentTimeMillis() - startTime < timeout) { if (function.get()) { @@ -342,7 +352,6 @@ public static void checkUtilTimeout(Supplier function, long timeout, lo } Thread.sleep(interval); } - Assert - .assertTrue("Timeout while waiting for the function to return expected result", false); + ObHTableTestUtil.Assert(tableNames, ()-> Assert.fail("Timeout while waiting for the function to return expected result")); } }