1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.master.cleaner;
19
20 import static org.junit.Assert.assertEquals;
21 import static org.junit.Assert.assertFalse;
22 import static org.junit.Assert.assertTrue;
23 import static org.mockito.Mockito.doThrow;
24 import static org.mockito.Mockito.spy;
25 import static org.mockito.Mockito.doAnswer;
26
27 import java.io.IOException;
28 import java.lang.reflect.Field;
29 import java.net.URLEncoder;
30 import java.util.Iterator;
31 import java.util.LinkedList;
32 import java.util.List;
33 import java.util.Random;
34 import java.util.concurrent.atomic.AtomicBoolean;
35
36 import com.google.common.collect.Lists;
37 import org.apache.hadoop.conf.Configuration;
38 import org.apache.hadoop.fs.FSDataOutputStream;
39 import org.apache.hadoop.fs.FileStatus;
40 import org.apache.hadoop.fs.FileSystem;
41 import org.apache.hadoop.fs.Path;
42 import org.apache.hadoop.hbase.Abortable;
43 import org.apache.hadoop.hbase.ChoreService;
44 import org.apache.hadoop.hbase.CoordinatedStateManager;
45 import org.apache.hadoop.hbase.HBaseTestingUtility;
46 import org.apache.hadoop.hbase.HConstants;
47 import org.apache.hadoop.hbase.testclassification.MediumTests;
48 import org.apache.hadoop.hbase.Server;
49 import org.apache.hadoop.hbase.ServerName;
50 import org.apache.hadoop.hbase.Waiter;
51 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
52 import org.apache.hadoop.hbase.client.ClusterConnection;
53 import org.apache.hadoop.hbase.replication.ReplicationFactory;
54 import org.apache.hadoop.hbase.replication.ReplicationQueues;
55 import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
56 import org.apache.hadoop.hbase.replication.master.ReplicationLogCleaner;
57 import org.apache.hadoop.hbase.replication.regionserver.Replication;
58 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
59 import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
60 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
61 import org.apache.zookeeper.KeeperException;
62 import org.junit.AfterClass;
63 import org.junit.BeforeClass;
64 import org.junit.Test;
65 import org.junit.experimental.categories.Category;
66 import org.mockito.Mockito;
67 import org.mockito.invocation.InvocationOnMock;
68 import org.mockito.stubbing.Answer;
69
70 @Category(MediumTests.class)
71 public class TestLogsCleaner {
72
73 private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
74
75
76
77
78 @BeforeClass
79 public static void setUpBeforeClass() throws Exception {
80 TEST_UTIL.startMiniZKCluster();
81 TEST_UTIL.startMiniDFSCluster(1);
82 CleanerChore.initChorePool(TEST_UTIL.getConfiguration());
83 }
84
85
86
87
88 @AfterClass
89 public static void tearDownAfterClass() throws Exception {
90 TEST_UTIL.shutdownMiniZKCluster();
91 TEST_UTIL.shutdownMiniDFSCluster();
92 }
93
94 @Test
95 public void testLogCleaning() throws Exception{
96 Configuration conf = TEST_UTIL.getConfiguration();
97
98 long ttl = 10000;
99 conf.setLong("hbase.master.logcleaner.ttl", ttl);
100 conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY, HConstants.REPLICATION_ENABLE_DEFAULT);
101 Replication.decorateMasterConfiguration(conf);
102 Server server = new DummyServer();
103 ReplicationQueues repQueues =
104 ReplicationFactory.getReplicationQueues(server.getZooKeeper(), conf, server);
105 repQueues.init(server.getServerName().toString());
106 final Path oldLogDir = new Path(TEST_UTIL.getDataTestDir(),
107 HConstants.HREGION_OLDLOGDIR_NAME);
108 String fakeMachineName =
109 URLEncoder.encode(server.getServerName().toString(), "UTF8");
110
111 final FileSystem fs = FileSystem.get(conf);
112
113
114 long now = System.currentTimeMillis();
115 fs.delete(oldLogDir, true);
116 fs.mkdirs(oldLogDir);
117
118 fs.createNewFile(new Path(oldLogDir, "a"));
119 fs.createNewFile(new Path(oldLogDir, fakeMachineName + "." + "a"));
120
121
122 System.out.println("Now is: " + now);
123 for (int i = 1; i < 31; i++) {
124
125
126 Path fileName = new Path(oldLogDir, fakeMachineName + "." + (now - i) );
127 fs.createNewFile(fileName);
128
129
130
131
132 if (i % (30/3) == 1) {
133 repQueues.addLog(fakeMachineName, fileName.getName());
134 System.out.println("Replication log file: " + fileName);
135 }
136 }
137
138
139 Thread.sleep(ttl);
140 fs.createNewFile(new Path(oldLogDir, fakeMachineName + "." + now));
141
142
143
144 fs.createNewFile(new Path(oldLogDir, fakeMachineName + "." + (now + 10000) ));
145
146 for (FileStatus stat : fs.listStatus(oldLogDir)) {
147 System.out.println(stat.getPath().toString());
148 }
149
150 assertEquals(34, fs.listStatus(oldLogDir).length);
151
152 LogCleaner cleaner = new LogCleaner(1000, server, conf, fs, oldLogDir);
153
154 cleaner.chore();
155
156
157
158 TEST_UTIL.waitFor(1000, new Waiter.Predicate<Exception>() {
159 @Override
160 public boolean evaluate() throws Exception {
161 return 5 == fs.listStatus(oldLogDir).length;
162 }
163 });
164
165 for (FileStatus file : fs.listStatus(oldLogDir)) {
166 System.out.println("Kept log files: " + file.getPath().getName());
167 }
168 }
169
170 @Test(timeout=5000)
171 public void testZnodeCversionChange() throws Exception {
172 Configuration conf = TEST_UTIL.getConfiguration();
173 ReplicationLogCleaner cleaner = new ReplicationLogCleaner();
174 cleaner.setConf(conf);
175
176 ReplicationQueuesClient rqcMock = Mockito.mock(ReplicationQueuesClient.class);
177 Mockito.when(rqcMock.getQueuesZNodeCversion()).thenReturn(1, 2, 3, 4);
178
179 Field rqc = ReplicationLogCleaner.class.getDeclaredField("replicationQueues");
180 rqc.setAccessible(true);
181
182 rqc.set(cleaner, rqcMock);
183
184
185 cleaner.getDeletableFiles(new LinkedList<FileStatus>());
186 }
187
188 @Test(timeout=10000)
189 public void testZooKeeperAbortDuringGetListOfReplicators() throws Exception {
190 Configuration conf = TEST_UTIL.getConfiguration();
191
192 ReplicationLogCleaner cleaner = new ReplicationLogCleaner();
193
194 List<FileStatus> dummyFiles = Lists.newArrayList(
195 new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path("log1")),
196 new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path("log2"))
197 );
198
199 FaultyZooKeeperWatcher faultyZK =
200 new FaultyZooKeeperWatcher(conf, "testZooKeeperAbort-faulty", null);
201 final AtomicBoolean getListOfReplicatorsFailed = new AtomicBoolean(false);
202
203 try {
204 faultyZK.init();
205 ReplicationQueuesClient replicationQueuesClient = spy(ReplicationFactory.getReplicationQueuesClient(
206 faultyZK, conf, new ReplicationLogCleaner.WarnOnlyAbortable()));
207 doAnswer(new Answer<Object>() {
208 @Override
209 public Object answer(InvocationOnMock invocation) throws Throwable {
210 try {
211 return invocation.callRealMethod();
212 } catch (KeeperException.ConnectionLossException e) {
213 getListOfReplicatorsFailed.set(true);
214 throw e;
215 }
216 }
217 }).when(replicationQueuesClient).getListOfReplicators();
218 replicationQueuesClient.init();
219
220 cleaner.setConf(conf, faultyZK, replicationQueuesClient);
221
222 Iterable<FileStatus> toDelete = cleaner.getDeletableFiles(dummyFiles);
223
224 assertTrue(getListOfReplicatorsFailed.get());
225 assertFalse(toDelete.iterator().hasNext());
226 assertFalse(cleaner.isStopped());
227 } finally {
228 faultyZK.close();
229 }
230 }
231
232
233
234
235
236 @Test(timeout=10000)
237 public void testZooKeeperNormal() throws Exception {
238 Configuration conf = TEST_UTIL.getConfiguration();
239 ReplicationLogCleaner cleaner = new ReplicationLogCleaner();
240
241 List<FileStatus> dummyFiles = Lists.newArrayList(
242 new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path("log1")),
243 new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path("log2"))
244 );
245
246 ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "testZooKeeperAbort-normal", null);
247 try {
248 cleaner.setConf(conf, zkw);
249 Iterable<FileStatus> filesToDelete = cleaner.getDeletableFiles(dummyFiles);
250 Iterator<FileStatus> iter = filesToDelete.iterator();
251 assertTrue(iter.hasNext());
252 assertEquals(new Path("log1"), iter.next().getPath());
253 assertTrue(iter.hasNext());
254 assertEquals(new Path("log2"), iter.next().getPath());
255 assertFalse(iter.hasNext());
256 } finally {
257 zkw.close();
258 }
259 }
260
261 @Test
262 public void testOnConfigurationChange() throws Exception {
263 Configuration conf = TEST_UTIL.getConfiguration();
264 conf.setInt(LogCleaner.OLD_WALS_CLEANER_THREAD_SIZE,
265 LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE);
266 conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
267 LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC);
268 conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
269 LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC);
270
271 Server server = new DummyServer();
272 Path oldWALsDir = new Path(TEST_UTIL.getDefaultRootDirPath(),
273 HConstants.HREGION_OLDLOGDIR_NAME);
274 FileSystem fs = TEST_UTIL.getDFSCluster().getFileSystem();
275 final LogCleaner cleaner = new LogCleaner(3000, server, conf, fs, oldWALsDir);
276 assertEquals(LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE, cleaner.getSizeOfCleaners());
277 assertEquals(LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
278 cleaner.getCleanerThreadTimeoutMsec());
279 assertEquals(LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
280 cleaner.getCleanerThreadCheckIntervalMsec());
281
282 fs.delete(oldWALsDir, true);
283 fs.mkdirs(oldWALsDir);
284 int numOfFiles = 10;
285 createFiles(fs, oldWALsDir, numOfFiles);
286 FileStatus[] status = fs.listStatus(oldWALsDir);
287 assertEquals(numOfFiles, status.length);
288
289 Thread thread = new Thread(new Runnable() {
290 @Override
291 public void run() {
292 cleaner.chore();
293 }
294 });
295 thread.setDaemon(true);
296 thread.start();
297
298 int sizeToChange = 4;
299 long threadTimeoutToChange = 30 * 1000L;
300 long threadCheckIntervalToChange = 250L;
301 conf.setInt(LogCleaner.OLD_WALS_CLEANER_THREAD_SIZE, sizeToChange);
302 conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC, threadTimeoutToChange);
303 conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
304 threadCheckIntervalToChange);
305 cleaner.onConfigurationChange(conf);
306 assertEquals(sizeToChange, cleaner.getSizeOfCleaners());
307 assertEquals(threadTimeoutToChange, cleaner.getCleanerThreadTimeoutMsec());
308 assertEquals(threadCheckIntervalToChange, cleaner.getCleanerThreadCheckIntervalMsec());
309
310 thread.join();
311 status = fs.listStatus(oldWALsDir);
312 assertEquals(0, status.length);
313 }
314
315 private void createFiles(FileSystem fs, Path parentDir, int numOfFiles) throws IOException {
316 Random random = new Random();
317 for (int i = 0; i < numOfFiles; i++) {
318 int xMega = 1 + random.nextInt(3);
319 try (FSDataOutputStream fsdos = fs.create(new Path(parentDir, "file-" + i))) {
320 for (int m = 0; m < xMega; m++) {
321 byte[] M = new byte[1024 * 1024];
322 random.nextBytes(M);
323 fsdos.write(M);
324 }
325 }
326 }
327 }
328
329 static class DummyServer implements Server {
330
331 @Override
332 public Configuration getConfiguration() {
333 return TEST_UTIL.getConfiguration();
334 }
335
336 @Override
337 public ZooKeeperWatcher getZooKeeper() {
338 try {
339 return new ZooKeeperWatcher(getConfiguration(), "dummy server", this);
340 } catch (IOException e) {
341 e.printStackTrace();
342 }
343 return null;
344 }
345
346 @Override
347 public CoordinatedStateManager getCoordinatedStateManager() {
348 return null;
349 }
350
351 @Override
352 public ClusterConnection getConnection() {
353 return null;
354 }
355
356 @Override
357 public MetaTableLocator getMetaTableLocator() {
358 return null;
359 }
360
361 @Override
362 public ServerName getServerName() {
363 return ServerName.valueOf("regionserver,60020,000000");
364 }
365
366 @Override
367 public void abort(String why, Throwable e) {}
368
369 @Override
370 public boolean isAborted() {
371 return false;
372 }
373
374 @Override
375 public void stop(String why) {}
376
377 @Override
378 public boolean isStopped() {
379 return false;
380 }
381
382 @Override
383 public ChoreService getChoreService() {
384 return null;
385 }
386 }
387
388 static class FaultyZooKeeperWatcher extends ZooKeeperWatcher {
389 private RecoverableZooKeeper zk;
390
391 public FaultyZooKeeperWatcher(Configuration conf, String identifier, Abortable abortable)
392 throws ZooKeeperConnectionException, IOException {
393 super(conf, identifier, abortable);
394 }
395
396 public void init() throws Exception {
397 this.zk = spy(super.getRecoverableZooKeeper());
398 doThrow(new KeeperException.ConnectionLossException())
399 .when(zk).getChildren("/hbase/replication/rs", null);
400 }
401
402 public RecoverableZooKeeper getRecoverableZooKeeper() {
403 return zk;
404 }
405 }
406 }