1
2
3
4
5
6
7
8
9
10
11 package org.apache.hadoop.hbase.master.cleaner;
12
13 import static org.junit.Assert.assertEquals;
14 import static org.junit.Assert.assertFalse;
15 import static org.junit.Assert.assertTrue;
16 import static org.junit.Assert.fail;
17 import static org.mockito.Mockito.doThrow;
18 import static org.mockito.Mockito.spy;
19
20 import com.google.common.collect.Lists;
21
22 import java.io.IOException;
23 import java.lang.reflect.Field;
24 import java.util.ArrayList;
25 import java.util.Iterator;
26 import java.util.List;
27
28 import org.apache.commons.logging.Log;
29 import org.apache.commons.logging.LogFactory;
30 import org.apache.hadoop.conf.Configuration;
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.Abortable;
35 import org.apache.hadoop.hbase.ChoreService;
36 import org.apache.hadoop.hbase.CoordinatedStateManager;
37 import org.apache.hadoop.hbase.HBaseTestingUtility;
38 import org.apache.hadoop.hbase.HConstants;
39 import org.apache.hadoop.hbase.Server;
40 import org.apache.hadoop.hbase.ServerName;
41 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
42 import org.apache.hadoop.hbase.client.ClusterConnection;
43 import org.apache.hadoop.hbase.replication.ReplicationException;
44 import org.apache.hadoop.hbase.replication.ReplicationFactory;
45 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
46 import org.apache.hadoop.hbase.replication.ReplicationPeers;
47 import org.apache.hadoop.hbase.replication.ReplicationQueues;
48 import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
49 import org.apache.hadoop.hbase.replication.ReplicationQueuesZKImpl;
50 import org.apache.hadoop.hbase.replication.master.ReplicationHFileCleaner;
51 import org.apache.hadoop.hbase.replication.regionserver.Replication;
52 import org.apache.hadoop.hbase.testclassification.MasterTests;
53 import org.apache.hadoop.hbase.testclassification.SmallTests;
54 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
55 import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
56 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
57 import org.apache.zookeeper.KeeperException;
58 import org.apache.zookeeper.data.Stat;
59 import org.junit.After;
60 import org.junit.AfterClass;
61 import org.junit.Before;
62 import org.junit.BeforeClass;
63 import org.junit.Test;
64 import org.junit.experimental.categories.Category;
65 import org.mockito.Mockito;
66
67 @Category({ MasterTests.class, SmallTests.class })
68 public class TestReplicationHFileCleaner {
69 private static final Log LOG = LogFactory.getLog(ReplicationQueuesZKImpl.class);
70 private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
71 private static Server server;
72 private static ReplicationQueues rq;
73 private static ReplicationPeers rp;
74 private static final String peerId = "TestReplicationHFileCleaner";
75 private static Configuration conf = TEST_UTIL.getConfiguration();
76 static FileSystem fs = null;
77 Path root;
78
79
80
81
82 @BeforeClass
83 public static void setUpBeforeClass() throws Exception {
84 TEST_UTIL.startMiniZKCluster();
85 server = new DummyServer();
86 conf.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true);
87 Replication.decorateMasterConfiguration(conf);
88 rp = ReplicationFactory.getReplicationPeers(server.getZooKeeper(), conf, server);
89 rp.init();
90
91 rq = ReplicationFactory.getReplicationQueues(server.getZooKeeper(), conf, server);
92 rq.init(server.getServerName().toString());
93 try {
94 fs = FileSystem.get(conf);
95 } finally {
96 if (fs != null) {
97 fs.close();
98 }
99 }
100 }
101
102
103
104
105 @AfterClass
106 public static void tearDownAfterClass() throws Exception {
107 TEST_UTIL.shutdownMiniZKCluster();
108 }
109
110 @Before
111 public void setup() throws ReplicationException, IOException {
112 root = TEST_UTIL.getDataTestDirOnTestFS();
113 rp.addPeer(peerId, new ReplicationPeerConfig().setClusterKey(TEST_UTIL.getClusterKey()), null);
114 rq.addPeerToHFileRefs(peerId);
115 }
116
117 @After
118 public void cleanup() throws ReplicationException {
119 try {
120 fs.delete(root, true);
121 } catch (IOException e) {
122 LOG.warn("Failed to delete files recursively from path " + root);
123 }
124 rp.removePeer(peerId);
125 }
126
127 @Test
128 public void testIsFileDeletable() throws IOException, ReplicationException {
129
130 Path file = new Path(root, "testIsFileDeletableWithNoHFileRefs");
131 fs.createNewFile(file);
132
133 assertTrue("Test file not created!", fs.exists(file));
134 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
135 cleaner.setConf(conf);
136
137 assertTrue("Cleaner should allow to delete this file as there is no hfile reference node "
138 + "for it in the queue.",
139 cleaner.isFileDeletable(fs.getFileStatus(file)));
140
141 List<String> files = new ArrayList<String>(1);
142 files.add(file.getName());
143
144 rq.addHFileRefs(peerId, files);
145
146 assertFalse("Cleaner should not allow to delete this file as there is a hfile reference node "
147 + "for it in the queue.",
148 cleaner.isFileDeletable(fs.getFileStatus(file)));
149 }
150
151 @Test
152 public void testGetDeletableFiles() throws Exception {
153
154 Path notDeletablefile = new Path(root, "testGetDeletableFiles_1");
155 fs.createNewFile(notDeletablefile);
156 assertTrue("Test file not created!", fs.exists(notDeletablefile));
157 Path deletablefile = new Path(root, "testGetDeletableFiles_2");
158 fs.createNewFile(deletablefile);
159 assertTrue("Test file not created!", fs.exists(deletablefile));
160
161 List<FileStatus> files = new ArrayList<FileStatus>(2);
162 FileStatus f = new FileStatus();
163 f.setPath(deletablefile);
164 files.add(f);
165 f = new FileStatus();
166 f.setPath(notDeletablefile);
167 files.add(f);
168
169 List<String> hfiles = new ArrayList<>(1);
170 hfiles.add(notDeletablefile.getName());
171
172 rq.addHFileRefs(peerId, hfiles);
173
174 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
175 cleaner.setConf(conf);
176 Iterator<FileStatus> deletableFilesIterator = cleaner.getDeletableFiles(files).iterator();
177 int i = 0;
178 while (deletableFilesIterator.hasNext() && i < 2) {
179 i++;
180 }
181
182 if (i > 2) {
183 fail("File " + notDeletablefile
184 + " should not be deletable as its hfile reference node is not added.");
185 }
186 assertTrue(deletableFilesIterator.next().getPath().equals(deletablefile));
187 }
188
189
190
191
192
193 @Test(timeout = 15000)
194 public void testForDifferntHFileRefsZnodeVersion() throws Exception {
195
196 Path file = new Path(root, "testForDifferntHFileRefsZnodeVersion");
197 fs.createNewFile(file);
198
199 assertTrue("Test file not created!", fs.exists(file));
200 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
201 cleaner.setConf(conf);
202
203 ReplicationQueuesClient replicationQueuesClient = Mockito.mock(ReplicationQueuesClient.class);
204
205 Mockito.when(replicationQueuesClient.getHFileRefsNodeChangeVersion()).thenReturn(1, 2);
206
207 Class<? extends ReplicationHFileCleaner> cleanerClass = cleaner.getClass();
208 Field rqc = cleanerClass.getDeclaredField("rqc");
209 rqc.setAccessible(true);
210 rqc.set(cleaner, replicationQueuesClient);
211
212 cleaner.isFileDeletable(fs.getFileStatus(file));
213 }
214
215
216
217
218 @Test
219 public void testZooKeeperAbort() throws Exception {
220 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
221
222 List<FileStatus> dummyFiles =
223 Lists.newArrayList(new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path(
224 "hfile1")), new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path(
225 "hfile2")));
226
227 FaultyZooKeeperWatcher faultyZK =
228 new FaultyZooKeeperWatcher(conf, "testZooKeeperAbort-faulty", null);
229 try {
230 faultyZK.init();
231 cleaner.setConf(conf, faultyZK);
232
233 Iterable<FileStatus> toDelete = cleaner.getDeletableFiles(dummyFiles);
234 assertFalse(toDelete.iterator().hasNext());
235 assertFalse(cleaner.isStopped());
236 } finally {
237 faultyZK.close();
238 }
239
240
241 cleaner = new ReplicationHFileCleaner();
242 ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "testZooKeeperAbort-normal", null);
243 try {
244 cleaner.setConf(conf, zkw);
245 Iterable<FileStatus> filesToDelete = cleaner.getDeletableFiles(dummyFiles);
246 Iterator<FileStatus> iter = filesToDelete.iterator();
247 assertTrue(iter.hasNext());
248 assertEquals(new Path("hfile1"), iter.next().getPath());
249 assertTrue(iter.hasNext());
250 assertEquals(new Path("hfile2"), iter.next().getPath());
251 assertFalse(iter.hasNext());
252 } finally {
253 zkw.close();
254 }
255 }
256
257 static class DummyServer implements Server {
258
259 @Override
260 public Configuration getConfiguration() {
261 return TEST_UTIL.getConfiguration();
262 }
263
264 @Override
265 public ZooKeeperWatcher getZooKeeper() {
266 try {
267 return new ZooKeeperWatcher(getConfiguration(), "dummy server", this);
268 } catch (IOException e) {
269 e.printStackTrace();
270 }
271 return null;
272 }
273
274 @Override
275 public CoordinatedStateManager getCoordinatedStateManager() {
276 return null;
277 }
278
279 @Override
280 public ClusterConnection getConnection() {
281 return null;
282 }
283
284 @Override
285 public MetaTableLocator getMetaTableLocator() {
286 return null;
287 }
288
289 @Override
290 public ServerName getServerName() {
291 return ServerName.valueOf("regionserver,60020,000000");
292 }
293
294 @Override
295 public void abort(String why, Throwable e) {
296 }
297
298 @Override
299 public boolean isAborted() {
300 return false;
301 }
302
303 @Override
304 public void stop(String why) {
305 }
306
307 @Override
308 public boolean isStopped() {
309 return false;
310 }
311
312 @Override
313 public ChoreService getChoreService() {
314 return null;
315 }
316 }
317
318 static class FaultyZooKeeperWatcher extends ZooKeeperWatcher {
319 private RecoverableZooKeeper zk;
320 public FaultyZooKeeperWatcher(Configuration conf, String identifier, Abortable abortable)
321 throws ZooKeeperConnectionException, IOException {
322 super(conf, identifier, abortable);
323 }
324
325 public void init() throws Exception {
326 this.zk = spy(super.getRecoverableZooKeeper());
327 doThrow(new KeeperException.ConnectionLossException())
328 .when(zk).getData("/hbase/replication/hfile-refs", null, new Stat());
329 }
330
331 public RecoverableZooKeeper getRecoverableZooKeeper() {
332 return zk;
333 }
334 }
335 }