1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.mob.mapreduce;
19
20 import static org.junit.Assert.assertEquals;
21 import static org.mockito.Mockito.mock;
22 import static org.mockito.Mockito.when;
23
24 import java.util.ArrayList;
25 import java.util.List;
26 import java.util.Set;
27 import java.util.TreeSet;
28
29 import org.apache.hadoop.conf.Configuration;
30 import org.apache.hadoop.fs.CommonConfigurationKeys;
31 import org.apache.hadoop.fs.FileStatus;
32 import org.apache.hadoop.fs.FileSystem;
33 import org.apache.hadoop.fs.Path;
34 import org.apache.hadoop.hbase.HBaseTestingUtility;
35 import org.apache.hadoop.hbase.HColumnDescriptor;
36 import org.apache.hadoop.hbase.HTableDescriptor;
37 import org.apache.hadoop.hbase.KeyValue;
38 import org.apache.hadoop.hbase.testclassification.MediumTests;
39 import org.apache.hadoop.hbase.ServerName;
40 import org.apache.hadoop.hbase.TableName;
41 import org.apache.hadoop.hbase.client.Admin;
42 import org.apache.hadoop.hbase.client.HTable;
43 import org.apache.hadoop.hbase.client.Put;
44 import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
45 import org.apache.hadoop.hbase.master.TableLockManager;
46 import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
47 import org.apache.hadoop.hbase.mob.MobConstants;
48 import org.apache.hadoop.hbase.mob.MobUtils;
49 import org.apache.hadoop.hbase.mob.mapreduce.SweepJob.DummyMobAbortable;
50 import org.apache.hadoop.hbase.mob.mapreduce.SweepJob.SweepCounter;
51 import org.apache.hadoop.hbase.util.Bytes;
52 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
53 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
54 import org.apache.hadoop.io.SequenceFile;
55 import org.apache.hadoop.io.Text;
56 import org.apache.hadoop.io.Writable;
57 import org.apache.hadoop.io.serializer.JavaSerialization;
58 import org.apache.hadoop.mapreduce.Counter;
59 import org.apache.hadoop.mapreduce.Reducer;
60 import org.apache.hadoop.mapreduce.counters.GenericCounter;
61 import org.junit.After;
62 import org.junit.AfterClass;
63 import org.junit.Before;
64 import org.junit.BeforeClass;
65 import org.junit.Test;
66 import org.junit.experimental.categories.Category;
67 import org.mockito.Matchers;
68
69 @Category(MediumTests.class)
70 public class TestMobSweepReducer {
71
72 private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
73 private final static String tableName = "testSweepReducer";
74 private final static String row = "row";
75 private final static String family = "family";
76 private final static String qf = "qf";
77 private static HTable table;
78 private static Admin admin;
79
80 @BeforeClass
81 public static void setUpBeforeClass() throws Exception {
82 TEST_UTIL.getConfiguration().setInt("hbase.master.info.port", 0);
83 TEST_UTIL.getConfiguration().setBoolean("hbase.regionserver.info.port.auto", true);
84
85 TEST_UTIL.getConfiguration().setInt("hfile.format.version", 3);
86
87 TEST_UTIL.startMiniCluster(1);
88 }
89
90 @AfterClass
91 public static void tearDownAfterClass() throws Exception {
92 TEST_UTIL.shutdownMiniCluster();
93 }
94
95 @SuppressWarnings("deprecation")
96 @Before
97 public void setUp() throws Exception {
98 HTableDescriptor desc = new HTableDescriptor(tableName);
99 HColumnDescriptor hcd = new HColumnDescriptor(family);
100 hcd.setMobEnabled(true);
101 hcd.setMobThreshold(3L);
102 hcd.setMaxVersions(4);
103 desc.addFamily(hcd);
104
105 admin = TEST_UTIL.getHBaseAdmin();
106 admin.createTable(desc);
107 table = new HTable(TEST_UTIL.getConfiguration(), tableName);
108 }
109
110 @After
111 public void tearDown() throws Exception {
112 admin.disableTable(TableName.valueOf(tableName));
113 admin.deleteTable(TableName.valueOf(tableName));
114 admin.close();
115 }
116
117 private List<String> getKeyFromSequenceFile(FileSystem fs, Path path,
118 Configuration conf) throws Exception {
119 List<String> list = new ArrayList<String>();
120 SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(path));
121
122 String next = (String) reader.next((String) null);
123 while (next != null) {
124 list.add(next);
125 next = (String) reader.next((String) null);
126 }
127 reader.close();
128 return list;
129 }
130
131 @Test
132 public void testRun() throws Exception {
133
134 TableName tn = TableName.valueOf(tableName);
135 byte[] mobValueBytes = new byte[100];
136
137
138 Path mobFamilyPath = MobUtils.getMobFamilyPath(TEST_UTIL.getConfiguration(), tn, family);
139
140 Put put = new Put(Bytes.toBytes(row));
141 put.add(Bytes.toBytes(family), Bytes.toBytes(qf), 1, mobValueBytes);
142 Put put2 = new Put(Bytes.toBytes(row + "ignore"));
143 put2.add(Bytes.toBytes(family), Bytes.toBytes(qf), 1, mobValueBytes);
144 table.put(put);
145 table.put(put2);
146 table.flushCommits();
147 admin.flush(tn);
148
149 FileStatus[] fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
150
151 assertEquals(1, fileStatuses.length);
152
153 String mobFile1 = fileStatuses[0].getPath().getName();
154
155 Configuration configuration = new Configuration(TEST_UTIL.getConfiguration());
156 configuration.setFloat(MobConstants.MOB_SWEEP_TOOL_COMPACTION_RATIO, 0.6f);
157 configuration.setStrings(TableInputFormat.INPUT_TABLE, tableName);
158 configuration.setStrings(TableInputFormat.SCAN_COLUMN_FAMILY, family);
159 configuration.setStrings(SweepJob.WORKING_VISITED_DIR_KEY, "jobWorkingNamesDir");
160 configuration.setStrings(SweepJob.WORKING_FILES_DIR_KEY, "compactionFileDir");
161 configuration.setStrings(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY,
162 JavaSerialization.class.getName());
163 configuration.set(SweepJob.WORKING_VISITED_DIR_KEY, "compactionVisitedDir");
164 configuration.setLong(MobConstants.MOB_SWEEP_TOOL_COMPACTION_START_DATE,
165 System.currentTimeMillis() + 24 * 3600 * 1000);
166
167 ZooKeeperWatcher zkw = new ZooKeeperWatcher(configuration, "1", new DummyMobAbortable());
168 TableName lockName = MobUtils.getTableLockName(tn);
169 String znode = ZKUtil.joinZNode(zkw.tableLockZNode, lockName.getNameAsString());
170 configuration.set(SweepJob.SWEEP_JOB_ID, "1");
171 configuration.set(SweepJob.SWEEP_JOB_TABLE_NODE, znode);
172 ServerName serverName = SweepJob.getCurrentServerName(configuration);
173 configuration.set(SweepJob.SWEEP_JOB_SERVERNAME, serverName.toString());
174
175 TableLockManager tableLockManager = TableLockManager.createTableLockManager(configuration, zkw,
176 serverName);
177 TableLock lock = tableLockManager.writeLock(lockName, "Run sweep tool");
178 lock.acquire();
179 try {
180
181 Counter counter = new GenericCounter();
182 Reducer<Text, KeyValue, Writable, Writable>.Context ctx = mock(Reducer.Context.class);
183 when(ctx.getConfiguration()).thenReturn(configuration);
184 when(ctx.getCounter(Matchers.any(SweepCounter.class))).thenReturn(counter);
185 when(ctx.nextKey()).thenReturn(true).thenReturn(false);
186 when(ctx.getCurrentKey()).thenReturn(new Text(mobFile1));
187
188 byte[] refBytes = Bytes.toBytes(mobFile1);
189 long valueLength = refBytes.length;
190 byte[] newValue = Bytes.add(Bytes.toBytes(valueLength), refBytes);
191 KeyValue kv2 = new KeyValue(Bytes.toBytes(row), Bytes.toBytes(family), Bytes.toBytes(qf), 1,
192 KeyValue.Type.Put, newValue);
193 List<KeyValue> list = new ArrayList<KeyValue>();
194 list.add(kv2);
195
196 when(ctx.getValues()).thenReturn(list);
197
198 SweepReducer reducer = new SweepReducer();
199 reducer.run(ctx);
200 } finally {
201 lock.release();
202 }
203 FileStatus[] filsStatuses2 = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
204 String mobFile2 = filsStatuses2[0].getPath().getName();
205
206 assertEquals(1, filsStatuses2.length);
207 assertEquals(false, mobFile2.equalsIgnoreCase(mobFile1));
208
209
210 String workingPath = configuration.get(SweepJob.WORKING_VISITED_DIR_KEY);
211 FileStatus[] statuses = TEST_UTIL.getTestFileSystem().listStatus(new Path(workingPath));
212 Set<String> files = new TreeSet<String>();
213 for (FileStatus st : statuses) {
214 files.addAll(getKeyFromSequenceFile(TEST_UTIL.getTestFileSystem(),
215 st.getPath(), configuration));
216 }
217 assertEquals(1, files.size());
218 assertEquals(true, files.contains(mobFile1));
219 }
220 }