1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.regionserver.compactions;
19
20 import static org.junit.Assert.assertEquals;
21
22 import java.io.IOException;
23 import java.util.List;
24 import java.util.concurrent.ThreadLocalRandom;
25
26 import org.apache.hadoop.conf.Configuration;
27 import org.apache.hadoop.hbase.DoNotRetryIOException;
28 import org.apache.hadoop.hbase.HBaseTestingUtility;
29 import org.apache.hadoop.hbase.HColumnDescriptor;
30 import org.apache.hadoop.hbase.HConstants;
31 import org.apache.hadoop.hbase.HTableDescriptor;
32 import org.apache.hadoop.hbase.MiniHBaseCluster;
33 import org.apache.hadoop.hbase.TableName;
34 import org.apache.hadoop.hbase.Waiter.ExplainingPredicate;
35 import org.apache.hadoop.hbase.client.HBaseAdmin;
36 import org.apache.hadoop.hbase.client.Put;
37 import org.apache.hadoop.hbase.client.Table;
38 import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine;
39 import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
40 import org.apache.hadoop.hbase.regionserver.HRegionServer;
41 import org.apache.hadoop.hbase.regionserver.HStore;
42 import org.apache.hadoop.hbase.regionserver.Region;
43 import org.apache.hadoop.hbase.regionserver.Store;
44 import org.apache.hadoop.hbase.regionserver.StoreFile;
45 import org.apache.hadoop.hbase.testclassification.MediumTests;
46 import org.apache.hadoop.hbase.util.Bytes;
47 import org.apache.hadoop.hbase.util.EnvironmentEdge;
48 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
49 import org.apache.hadoop.hbase.util.JVMClusterUtil;
50 import org.apache.hadoop.hbase.util.TimeOffsetEnvironmentEdge;
51 import org.junit.AfterClass;
52 import org.junit.Assert;
53 import org.junit.BeforeClass;
54 import org.junit.Rule;
55 import org.junit.Test;
56 import org.junit.experimental.categories.Category;
57 import org.junit.rules.ExpectedException;
58
59 @Category({ MediumTests.class })
60 public class TestFIFOCompactionPolicy {
61
62 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
63
64 private final byte[] family = Bytes.toBytes("f");
65
66 private final byte[] qualifier = Bytes.toBytes("q");
67
68 @Rule
69 public ExpectedException error = ExpectedException.none();
70
71 private Store getStoreWithName(TableName tableName) {
72 MiniHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
73 List<JVMClusterUtil.RegionServerThread> rsts = cluster.getRegionServerThreads();
74 for (int i = 0; i < cluster.getRegionServerThreads().size(); i++) {
75 HRegionServer hrs = rsts.get(i).getRegionServer();
76 for (Region region : hrs.getOnlineRegions(tableName)) {
77 return region.getStores().iterator().next();
78 }
79 }
80 return null;
81 }
82
83 private Store prepareData(TableName tableName) throws IOException {
84 HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
85 HTableDescriptor desc = new HTableDescriptor(tableName);
86 desc.setConfiguration(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
87 FIFOCompactionPolicy.class.getName());
88 desc.setConfiguration(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
89 DisabledRegionSplitPolicy.class.getName());
90 HColumnDescriptor colDesc = new HColumnDescriptor(family);
91 colDesc.setTimeToLive(1);
92 desc.addFamily(colDesc);
93
94 admin.createTable(desc);
95 Table table = TEST_UTIL.getConnection().getTable(tableName);
96 TimeOffsetEnvironmentEdge edge =
97 (TimeOffsetEnvironmentEdge) EnvironmentEdgeManager.getDelegate();
98 for (int i = 0; i < 10; i++) {
99 for (int j = 0; j < 10; j++) {
100 byte[] value = new byte[128 * 1024];
101 ThreadLocalRandom.current().nextBytes(value);
102 table.put(new Put(Bytes.toBytes(i * 10 + j)).addColumn(family, qualifier, value));
103 }
104 admin.flush(tableName);
105 edge.increment(1001);
106 }
107 return getStoreWithName(tableName);
108 }
109
110 @BeforeClass
111 public static void setEnvironmentEdge() throws Exception {
112 EnvironmentEdge ee = new TimeOffsetEnvironmentEdge();
113 EnvironmentEdgeManager.injectEdge(ee);
114 Configuration conf = TEST_UTIL.getConfiguration();
115 conf.setInt(HStore.BLOCKING_STOREFILES_KEY, 10000);
116 TEST_UTIL.startMiniCluster(1);
117 }
118
119 @AfterClass
120 public static void resetEnvironmentEdge() throws Exception {
121 TEST_UTIL.shutdownMiniCluster();
122 EnvironmentEdgeManager.reset();
123 }
124
125 @Test
126 public void testPurgeExpiredFiles() throws Exception {
127 TableName tableName = TableName.valueOf(getClass().getSimpleName());
128 final Store store = prepareData(tableName);
129 assertEquals(10, store.getStorefilesCount());
130 TEST_UTIL.getHBaseAdmin().majorCompact(tableName);
131 TEST_UTIL.waitFor(30000, new ExplainingPredicate<Exception>() {
132
133 @Override
134 public boolean evaluate() throws Exception {
135 return store.getStorefilesCount() == 1;
136 }
137
138 @Override
139 public String explainFailure() throws Exception {
140 return "The store file count " + store.getStorefilesCount() + " is still greater than 1";
141 }
142 });
143 }
144
145 @Test
146 public void testSanityCheckTTL() throws IOException {
147 error.expect(DoNotRetryIOException.class);
148 error.expectMessage("Default TTL is not supported");
149 TableName tableName = TableName.valueOf(getClass().getSimpleName() + "-TTL");
150 HTableDescriptor desc = new HTableDescriptor(tableName);
151 desc.setConfiguration(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
152 FIFOCompactionPolicy.class.getName());
153 desc.setConfiguration(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
154 DisabledRegionSplitPolicy.class.getName());
155 HColumnDescriptor colDesc = new HColumnDescriptor(family);
156 desc.addFamily(colDesc);
157 TEST_UTIL.getHBaseAdmin().createTable(desc);
158 }
159
160 @Test
161 public void testSanityCheckMinVersion() throws IOException {
162 error.expect(DoNotRetryIOException.class);
163 error.expectMessage("MIN_VERSION > 0 is not supported for FIFO compaction");
164 TableName tableName = TableName.valueOf(getClass().getSimpleName() + "-MinVersion");
165 HTableDescriptor desc = new HTableDescriptor(tableName);
166 desc.setConfiguration(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
167 FIFOCompactionPolicy.class.getName());
168 desc.setConfiguration(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
169 DisabledRegionSplitPolicy.class.getName());
170 HColumnDescriptor colDesc = new HColumnDescriptor(family);
171 colDesc.setTimeToLive(1);
172 colDesc.setMinVersions(1);
173 desc.addFamily(colDesc);
174 TEST_UTIL.getHBaseAdmin().createTable(desc);
175 }
176
177 @Test
178 public void testSanityCheckBlockingStoreFiles() throws IOException {
179 error.expect(DoNotRetryIOException.class);
180 error.expectMessage("blocking file count 'hbase.hstore.blockingStoreFiles'");
181 error.expectMessage("is below recommended minimum of 1000");
182 TableName tableName = TableName.valueOf(getClass().getSimpleName() + "-BlockingStoreFiles");
183 HTableDescriptor desc = new HTableDescriptor(tableName);
184 desc.setConfiguration(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
185 FIFOCompactionPolicy.class.getName());
186 desc.setConfiguration(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
187 DisabledRegionSplitPolicy.class.getName());
188 desc.setConfiguration(HStore.BLOCKING_STOREFILES_KEY, "10");
189 HColumnDescriptor colDesc = new HColumnDescriptor(family);
190 colDesc.setTimeToLive(1);
191 desc.addFamily(colDesc);
192 TEST_UTIL.getHBaseAdmin().createTable(desc);
193 }
194
195
196
197
198 @Test
199 public void testFIFOCompactionPolicyExpiredEmptyHFiles() throws Exception {
200 TableName tableName = TableName.valueOf("testFIFOCompactionPolicyExpiredEmptyHFiles");
201 HColumnDescriptor colDesc = new HColumnDescriptor(family).setTimeToLive(1);
202 HTableDescriptor desc = new HTableDescriptor(tableName)
203 .setValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
204 FIFOCompactionPolicy.class.getName())
205 .setValue(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
206 DisabledRegionSplitPolicy.class.getName())
207 .addFamily(colDesc);
208 Table table = TEST_UTIL.createTable(desc, null);
209 long ts = System.currentTimeMillis() - 10 * 1000;
210 Put put =
211 new Put(Bytes.toBytes("row1")).addColumn(family, qualifier, ts, Bytes.toBytes("value0"));
212 table.put(put);
213 TEST_UTIL.getHBaseAdmin().flush(tableName);
214 put = new Put(Bytes.toBytes("row2")).addColumn(family, qualifier, ts, Bytes.toBytes("value1"));
215 table.put(put);
216 TEST_UTIL.getHBaseAdmin().flush(tableName);
217 Store store = getStoreWithName(tableName);
218 Assert.assertNotNull(store);
219 Assert.assertEquals(2, store.getStorefilesCount());
220 TEST_UTIL.getHBaseAdmin().majorCompact(tableName);
221 for (int i = 0; i < 100; i++) {
222 if (store.getStorefilesCount() > 1) {
223 Thread.sleep(100);
224 } else {
225 break;
226 }
227 }
228 Assert.assertEquals(1, store.getStorefilesCount());
229 StoreFile sf = store.getStorefiles().iterator().next();
230 Assert.assertNotNull(sf);
231 Assert.assertEquals(0, sf.getReader().getEntries());
232 put = new Put(Bytes.toBytes("row3")).addColumn(family, qualifier, ts, Bytes.toBytes("value1"));
233 table.put(put);
234 TEST_UTIL.getHBaseAdmin().flush(tableName);
235 Assert.assertEquals(2, store.getStorefilesCount());
236 TEST_UTIL.getHBaseAdmin().majorCompact(tableName);
237 for (int i = 0; i < 100; i++) {
238 if (store.getStorefilesCount() > 1) {
239 Thread.sleep(100);
240 } else {
241 break;
242 }
243 }
244 Assert.assertEquals(1, store.getStorefilesCount());
245 sf = store.getStorefiles().iterator().next();
246 Assert.assertNotNull(sf);
247 Assert.assertEquals(0, sf.getReader().getEntries());
248 }
249 }