hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFlushFromClient.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import static org.junit.Assert.assertEquals;
  21 import static org.junit.Assert.assertFalse;
  22 import static org.junit.Assert.assertTrue;
  23
  24 import java.util.Arrays;
  25 import java.util.List;
  26 import java.util.concurrent.TimeUnit;
  27 import java.util.stream.Collectors;
  28 import org.apache.hadoop.hbase.HBaseClassTestRule;
  29 import org.apache.hadoop.hbase.HBaseTestingUtility;
  30 import org.apache.hadoop.hbase.TableName;
  31 import org.apache.hadoop.hbase.regionserver.HRegion;
  32 import org.apache.hadoop.hbase.regionserver.HRegionServer;
  33 import org.apache.hadoop.hbase.testclassification.ClientTests;
  34 import org.apache.hadoop.hbase.testclassification.MediumTests;
  35 import org.apache.hadoop.hbase.util.Bytes;
  36 import org.apache.hadoop.hbase.util.JVMClusterUtil;
  37 import org.apache.hadoop.io.IOUtils;
  38 import org.junit.After;
  39 import org.junit.AfterClass;
  40 import org.junit.Before;
  41 import org.junit.BeforeClass;
  42 import org.junit.ClassRule;
  43 import org.junit.Rule;
  44 import org.junit.Test;
  45 import org.junit.experimental.categories.Category;
  46 import org.junit.rules.TestName;
  47
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 @Category({MediumTests.class, ClientTests.class})
  52 public class TestFlushFromClient {
  53
  54   @ClassRule
  55   public static final HBaseClassTestRule CLASS_RULE =
  56       HBaseClassTestRule.forClass(TestFlushFromClient.class);
  57
  58   private static final Logger LOG = LoggerFactory.getLogger(TestFlushFromClient.class);
  59   private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
  60   private static AsyncConnection asyncConn;
  61   private static final byte[][] SPLITS = new byte[][]{Bytes.toBytes("3"), Bytes.toBytes("7")};
  62   private static final List<byte[]> ROWS = Arrays.asList(
  63     Bytes.toBytes("1"),
  64     Bytes.toBytes("4"),
  65     Bytes.toBytes("8"));
  66   private static final byte[] FAMILY = Bytes.toBytes("f1");
  67
  68   @Rule
  69   public TestName name = new TestName();
  70
  71   public TableName tableName;
  72
  73   @BeforeClass
  74   public static void setUpBeforeClass() throws Exception {
  75     TEST_UTIL.startMiniCluster(ROWS.size());
  76     asyncConn = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
  77   }
  78
  79   @AfterClass
  80   public static void tearDownAfterClass() throws Exception {
  81     IOUtils.cleanup(null, asyncConn);
  82     TEST_UTIL.shutdownMiniCluster();
  83   }
  84
  85   @Before
  86   public void setUp() throws Exception {
  87     tableName = TableName.valueOf(name.getMethodName());
  88     try (Table t = TEST_UTIL.createTable(tableName, FAMILY, SPLITS)) {
  89       List<Put> puts = ROWS.stream().map(r -> new Put(r)).collect(Collectors.toList());
  90       for (int i = 0; i != 20; ++i) {
  91         byte[] value = Bytes.toBytes(i);
  92         puts.forEach(p -> p.addColumn(FAMILY, value, value));
  93       }
  94       t.put(puts);
  95     }
  96     assertFalse(getRegionInfo().isEmpty());
  97     assertTrue(getRegionInfo().stream().allMatch(r -> r.getMemStoreDataSize() != 0));
  98   }
  99
 100   @After
 101   public void tearDown() throws Exception {
 102     for (TableDescriptor htd : TEST_UTIL.getAdmin().listTableDescriptors()) {
 103       LOG.info("Tear down, remove table=" + htd.getTableName());
 104       TEST_UTIL.deleteTable(htd.getTableName());
 105     }
 106   }
 107
 108   @Test
 109   public void testFlushTable() throws Exception {
 110     try (Admin admin = TEST_UTIL.getAdmin()) {
 111       admin.flush(tableName);
 112       assertFalse(getRegionInfo().stream().anyMatch(r -> r.getMemStoreDataSize() != 0));
 113     }
 114   }
 115
 116   @Test
 117   public void testAsyncFlushTable() throws Exception {
 118     AsyncAdmin admin = asyncConn.getAdmin();
 119     admin.flush(tableName).get();
 120     assertFalse(getRegionInfo().stream().anyMatch(r -> r.getMemStoreDataSize() != 0));
 121   }
 122
 123   @Test
 124   public void testFlushRegion() throws Exception {
 125     try (Admin admin = TEST_UTIL.getAdmin()) {
 126       for (HRegion r : getRegionInfo()) {
 127         admin.flushRegion(r.getRegionInfo().getRegionName());
 128         TimeUnit.SECONDS.sleep(1);
 129         assertEquals(0, r.getMemStoreDataSize());
 130       }
 131     }
 132   }
 133
 134   @Test
 135   public void testAsyncFlushRegion() throws Exception {
 136     AsyncAdmin admin = asyncConn.getAdmin();
 137     for (HRegion r : getRegionInfo()) {
 138       admin.flushRegion(r.getRegionInfo().getRegionName()).get();
 139       TimeUnit.SECONDS.sleep(1);
 140       assertEquals(0, r.getMemStoreDataSize());
 141     }
 142   }
 143
 144   @Test
 145   public void testFlushRegionServer() throws Exception {
 146     try (Admin admin = TEST_UTIL.getAdmin()) {
 147       for (HRegionServer rs : TEST_UTIL.getHBaseCluster()
 148             .getLiveRegionServerThreads()
 149             .stream().map(JVMClusterUtil.RegionServerThread::getRegionServer)
 150             .collect(Collectors.toList())) {
 151         admin.flushRegionServer(rs.getServerName());
 152         assertFalse(getRegionInfo(rs).stream().anyMatch(r -> r.getMemStoreDataSize() != 0));
 153       }
 154     }
 155   }
 156
 157   @Test
 158   public void testAsyncFlushRegionServer() throws Exception {
 159     AsyncAdmin admin = asyncConn.getAdmin();
 160     for (HRegionServer rs : TEST_UTIL.getHBaseCluster()
 161       .getLiveRegionServerThreads()
 162       .stream().map(JVMClusterUtil.RegionServerThread::getRegionServer)
 163       .collect(Collectors.toList())) {
 164       admin.flushRegionServer(rs.getServerName()).get();
 165       assertFalse(getRegionInfo(rs).stream().anyMatch(r -> r.getMemStoreDataSize() != 0));
 166     }
 167   }
 168
 169   private List<HRegion> getRegionInfo() {
 170     return TEST_UTIL.getHBaseCluster().getLiveRegionServerThreads().stream()
 171       .map(JVMClusterUtil.RegionServerThread::getRegionServer)
 172       .flatMap(r -> r.getRegions().stream())
 173       .filter(r -> r.getTableDescriptor().getTableName().equals(tableName))
 174       .collect(Collectors.toList());
 175   }
 176
 177   private List<HRegion> getRegionInfo(HRegionServer rs) {
 178     return rs.getRegions().stream()
 179       .filter(v -> v.getTableDescriptor().getTableName().equals(tableName))
 180       .collect(Collectors.toList());
 181   }
 182 }