1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.fs;
20
21 import java.io.FileNotFoundException;
22 import java.io.IOException;
23 import java.lang.reflect.Field;
24 import java.lang.reflect.InvocationTargetException;
25 import java.lang.reflect.Method;
26 import java.net.BindException;
27 import java.net.ServerSocket;
28 import java.util.List;
29 import java.util.concurrent.CountDownLatch;
30
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.commons.logging.impl.Log4JLogger;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.fs.BlockLocation;
36 import org.apache.hadoop.fs.FSDataInputStream;
37 import org.apache.hadoop.fs.FSDataOutputStream;
38 import org.apache.hadoop.fs.FileStatus;
39 import org.apache.hadoop.fs.FileSystem;
40 import org.apache.hadoop.fs.Path;
41 import org.apache.hadoop.ipc.RemoteException;
42 import org.apache.hadoop.hbase.HBaseTestingUtility;
43 import org.apache.hadoop.hbase.HConstants;
44 import org.apache.hadoop.hbase.testclassification.LargeTests;
45 import org.apache.hadoop.hbase.MiniHBaseCluster;
46 import org.apache.hadoop.hbase.TableName;
47 import org.apache.hadoop.hbase.client.Put;
48 import org.apache.hadoop.hbase.client.Table;
49 import org.apache.hadoop.hbase.regionserver.HRegion;
50 import org.apache.hadoop.hbase.regionserver.HRegionServer;
51 import org.apache.hadoop.hbase.regionserver.Region;
52 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
53 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
54 import org.apache.hadoop.hbase.util.FSUtils;
55 import org.apache.hadoop.hdfs.DFSClient;
56 import org.apache.hadoop.hdfs.DistributedFileSystem;
57 import org.apache.hadoop.hdfs.MiniDFSCluster;
58 import org.apache.hadoop.hdfs.protocol.ClientProtocol;
59 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
60 import org.apache.hadoop.hdfs.protocol.DirectoryListing;
61 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
62 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
63 import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
64 import org.apache.hadoop.hdfs.server.datanode.DataNode;
65 import org.apache.log4j.Level;
66 import org.junit.After;
67 import org.junit.Assert;
68 import org.junit.Before;
69 import org.junit.Test;
70 import org.junit.experimental.categories.Category;
71
72
73
74
75 @Category(LargeTests.class)
76 public class TestBlockReorder {
77 private static final Log LOG = LogFactory.getLog(TestBlockReorder.class);
78
79 static {
80 ((Log4JLogger) DFSClient.LOG).getLogger().setLevel(Level.ALL);
81 ((Log4JLogger) HFileSystem.LOG).getLogger().setLevel(Level.ALL);
82 }
83
84 private Configuration conf;
85 private MiniDFSCluster cluster;
86 private HBaseTestingUtility htu;
87 private DistributedFileSystem dfs;
88 private static final String host1 = "host1";
89 private static final String host2 = "host2";
90 private static final String host3 = "host3";
91 private static Path rootDir;
92 private static Path walRootDir;
93
94 @Before
95 public void setUp() throws Exception {
96 htu = new HBaseTestingUtility();
97 htu.getConfiguration().setInt("dfs.blocksize", 1024);
98 htu.getConfiguration().setBoolean("dfs.support.append", true);
99 htu.getConfiguration().setInt("dfs.replication", 3);
100 htu.startMiniDFSCluster(3,
101 new String[]{"/r1", "/r2", "/r3"}, new String[]{host1, host2, host3});
102
103 conf = htu.getConfiguration();
104 cluster = htu.getDFSCluster();
105 dfs = (DistributedFileSystem) FileSystem.get(conf);
106 rootDir = htu.createRootDir();
107 walRootDir = htu.createWALRootDir();
108 }
109
110 @After
111 public void tearDownAfterClass() throws Exception {
112 dfs.delete(rootDir, true);
113 dfs.delete(walRootDir, true);
114 htu.shutdownMiniCluster();
115 }
116
117
118
119
120 @Test
121 public void testBlockLocationReorder() throws Exception {
122 Path p = new Path("hello");
123
124 Assert.assertTrue((short) cluster.getDataNodes().size() > 1);
125 final int repCount = 2;
126
127
128 FSDataOutputStream fop = dfs.create(p, (short) repCount);
129 final double toWrite = 875.5613;
130 fop.writeDouble(toWrite);
131 fop.close();
132
133
134 long start = System.currentTimeMillis();
135 FSDataInputStream fin = dfs.open(p);
136 Assert.assertTrue(toWrite == fin.readDouble());
137 long end = System.currentTimeMillis();
138 LOG.info("readtime= " + (end - start));
139 fin.close();
140 Assert.assertTrue((end - start) < 30 * 1000);
141
142
143
144 FileStatus f = dfs.getFileStatus(p);
145 BlockLocation[] lbs;
146 do {
147 lbs = dfs.getFileBlockLocations(f, 0, 1);
148 } while (lbs.length != 1 && lbs[0].getLength() != repCount);
149 final String name = lbs[0].getNames()[0];
150 Assert.assertTrue(name.indexOf(':') > 0);
151 String portS = name.substring(name.indexOf(':') + 1);
152 final int port = Integer.parseInt(portS);
153 LOG.info("port= " + port);
154 int ipcPort = -1;
155
156
157
158 boolean ok = false;
159 final String lookup = lbs[0].getHosts()[0];
160 StringBuilder sb = new StringBuilder();
161 for (DataNode dn : cluster.getDataNodes()) {
162 final String dnName = getHostName(dn);
163 sb.append(dnName).append(' ');
164 if (lookup.equals(dnName)) {
165 ok = true;
166 LOG.info("killing datanode " + name + " / " + lookup);
167 ipcPort = dn.ipcServer.getListenerAddress().getPort();
168 dn.shutdown();
169 LOG.info("killed datanode " + name + " / " + lookup);
170 break;
171 }
172 }
173 Assert.assertTrue(
174 "didn't find the server to kill, was looking for " + lookup + " found " + sb, ok);
175 LOG.info("ipc port= " + ipcPort);
176
177
178 Assert.assertTrue(HFileSystem.addLocationsOrderInterceptor(conf,
179 new HFileSystem.ReorderBlocks() {
180 @Override
181 public void reorderBlocks(Configuration c, LocatedBlocks lbs, String src) {
182 for (LocatedBlock lb : lbs.getLocatedBlocks()) {
183 if (lb.getLocations().length > 1) {
184 DatanodeInfo[] infos = lb.getLocations();
185 if (infos[0].getHostName().equals(lookup)) {
186 LOG.info("HFileSystem bad host, inverting");
187 DatanodeInfo tmp = infos[0];
188 infos[0] = infos[1];
189 infos[1] = tmp;
190 }
191 }
192 }
193 }
194 }));
195
196
197 final int retries = 10;
198 ServerSocket ss = null;
199 ServerSocket ssI;
200 try {
201 ss = new ServerSocket(port);
202 ssI = new ServerSocket(ipcPort);
203 } catch (BindException be) {
204 LOG.warn("Got bind exception trying to set up socket on " + port + " or " + ipcPort +
205 ", this means that the datanode has not closed the socket or" +
206 " someone else took it. It may happen, skipping this test for this time.", be);
207 if (ss != null) {
208 ss.close();
209 }
210 return;
211 }
212
213
214
215 for (int i = 0; i < retries; i++) {
216 start = System.currentTimeMillis();
217
218 fin = dfs.open(p);
219 Assert.assertTrue(toWrite == fin.readDouble());
220 fin.close();
221 end = System.currentTimeMillis();
222 LOG.info("HFileSystem readtime= " + (end - start));
223 Assert.assertFalse("We took too much time to read", (end - start) > 60000);
224 }
225
226 ss.close();
227 ssI.close();
228 }
229
230
231
232
233 private String getHostName(DataNode dn) throws InvocationTargetException, IllegalAccessException {
234 Method m;
235 try {
236 m = DataNode.class.getMethod("getDisplayName");
237 } catch (NoSuchMethodException e) {
238 try {
239 m = DataNode.class.getMethod("getHostName");
240 } catch (NoSuchMethodException e1) {
241 throw new RuntimeException(e1);
242 }
243 }
244
245 String res = (String) m.invoke(dn);
246 if (res.contains(":")) {
247 return res.split(":")[0];
248 } else {
249 return res;
250 }
251 }
252
253
254
255
256 @Test()
257 public void testHBaseCluster() throws Exception {
258 byte[] sb = "sb".getBytes();
259 htu.startMiniZKCluster();
260
261 MiniHBaseCluster hbm = htu.startMiniHBaseCluster(1, 1);
262 hbm.waitForActiveAndReadyMaster();
263 hbm.getRegionServer(0).waitForServerOnline();
264 HRegionServer targetRs = hbm.getRegionServer(0);
265
266
267
268 String host4 = targetRs.getServerName().getHostname();
269 LOG.info("Starting a new datanode with the name=" + host4);
270 cluster.startDataNodes(conf, 1, true, null, new String[]{"/r4"}, new String[]{host4}, null);
271 cluster.waitClusterUp();
272
273 final int repCount = 3;
274
275
276 conf = targetRs.getConfiguration();
277 HFileSystem rfs = (HFileSystem) targetRs.getFileSystem();
278 Table h = htu.createTable(TableName.valueOf("table"), sb);
279
280
281
282
283
284
285
286 String walDir = new Path(FSUtils.getWALRootDir(conf) + "/" + HConstants.HREGION_LOGDIR_NAME +
287 "/" + targetRs.getServerName().toString()).toUri().getPath();
288
289 DistributedFileSystem mdfs = (DistributedFileSystem)
290 hbm.getMaster().getMasterFileSystem().getFileSystem();
291
292
293 int nbTest = 0;
294 while (nbTest < 10) {
295 final List<Region> regions = targetRs.getOnlineRegions(h.getName());
296 final CountDownLatch latch = new CountDownLatch(regions.size());
297
298 final WALActionsListener listener = new WALActionsListener.Base() {
299 @Override
300 public void postLogRoll(final Path oldPath, final Path newPath) throws IOException {
301 latch.countDown();
302 }
303 };
304 for (Region region : regions) {
305 ((HRegion)region).getWAL().registerWALActionsListener(listener);
306 }
307
308 htu.getHBaseAdmin().rollWALWriter(targetRs.getServerName());
309
310
311 try {
312 latch.await();
313 } catch (InterruptedException exception) {
314 LOG.warn("Interrupted while waiting for the wal of '" + targetRs + "' to roll. If later " +
315 "tests fail, it's probably because we should still be waiting.");
316 Thread.currentThread().interrupt();
317 }
318 for (Region region : regions) {
319 ((HRegion)region).getWAL().unregisterWALActionsListener(listener);
320 }
321
322
323 Thread.sleep(100);
324
325
326 Put p = new Put(sb);
327 p.add(sb, sb, sb);
328 h.put(p);
329
330 DirectoryListing dl = dfs.getClient().listPaths(walDir, HdfsFileStatus.EMPTY_NAME);
331 HdfsFileStatus[] hfs = dl.getPartialListing();
332
333
334 Assert.assertTrue(hfs.length >= 1);
335 for (HdfsFileStatus hf : hfs) {
336
337 try {
338 LOG.info("Log file found: " + hf.getLocalName() + " in " + walDir);
339 String logFile = walDir + "/" + hf.getLocalName();
340 FileStatus fsLog = rfs.getFileStatus(new Path(logFile));
341
342 LOG.info("Checking log file: " + logFile);
343
344
345
346
347 BlockLocation[] bls = rfs.getFileBlockLocations(fsLog, 0, 1);
348 if (bls.length > 0) {
349 BlockLocation bl = bls[0];
350
351 LOG.info(bl.getHosts().length + " replicas for block 0 in " + logFile + " ");
352 for (int i = 0; i < bl.getHosts().length - 1; i++) {
353 LOG.info(bl.getHosts()[i] + " " + logFile);
354 Assert.assertNotSame(bl.getHosts()[i], host4);
355 }
356 String last = bl.getHosts()[bl.getHosts().length - 1];
357 LOG.info(last + " " + logFile);
358 if (host4.equals(last)) {
359 nbTest++;
360 LOG.info(logFile + " is on the new datanode and is ok");
361 if (bl.getHosts().length == 3) {
362
363
364 testFromDFS(dfs, logFile, repCount, host4);
365
366
367 testFromDFS(mdfs, logFile, repCount, host4);
368 }
369 }
370 }
371 } catch (FileNotFoundException exception) {
372 LOG.debug("Failed to find log file '" + hf.getLocalName() + "'; it probably was " +
373 "archived out from under us so we'll ignore and retry. If this test hangs " +
374 "indefinitely you should treat this failure as a symptom.", exception);
375 } catch (RemoteException exception) {
376 if (exception.unwrapRemoteException() instanceof FileNotFoundException) {
377 LOG.debug("Failed to find log file '" + hf.getLocalName() + "'; it probably was " +
378 "archived out from under us so we'll ignore and retry. If this test hangs " +
379 "indefinitely you should treat this failure as a symptom.", exception);
380 } else {
381 throw exception;
382 }
383 }
384 }
385 }
386 }
387
388 private void testFromDFS(DistributedFileSystem dfs, String src, int repCount, String localhost)
389 throws Exception {
390
391 for (int i = 0; i < 10; i++) {
392 LocatedBlocks l;
393
394 final long max = System.currentTimeMillis() + 10000;
395 boolean done;
396 do {
397 Assert.assertTrue("Can't get enouth replica.", System.currentTimeMillis() < max);
398 l = getNamenode(dfs.getClient()).getBlockLocations(src, 0, 1);
399 Assert.assertNotNull("Can't get block locations for " + src, l);
400 Assert.assertNotNull(l.getLocatedBlocks());
401 Assert.assertTrue(l.getLocatedBlocks().size() > 0);
402
403 done = true;
404 for (int y = 0; y < l.getLocatedBlocks().size() && done; y++) {
405 done = (l.get(y).getLocations().length == repCount);
406 }
407 } while (!done);
408
409 for (int y = 0; y < l.getLocatedBlocks().size() && done; y++) {
410 Assert.assertEquals(localhost, l.get(y).getLocations()[repCount - 1].getHostName());
411 }
412 }
413 }
414
415 private static ClientProtocol getNamenode(DFSClient dfsc) throws Exception {
416 Field nf = DFSClient.class.getDeclaredField("namenode");
417 nf.setAccessible(true);
418 return (ClientProtocol) nf.get(dfsc);
419 }
420
421
422
423
424 @Test
425 public void testBlockLocation() throws Exception {
426
427 htu.startMiniZKCluster();
428 MiniHBaseCluster hbm = htu.startMiniHBaseCluster(1, 1);
429 conf = hbm.getConfiguration();
430
431
432
433 final String fileName = "/helloWorld";
434 Path p = new Path(fileName);
435
436 final int repCount = 3;
437 Assert.assertTrue((short) cluster.getDataNodes().size() >= repCount);
438
439
440 FSDataOutputStream fop = dfs.create(p, (short) repCount);
441 final double toWrite = 875.5613;
442 fop.writeDouble(toWrite);
443 fop.close();
444
445 for (int i=0; i<10; i++){
446
447 LocatedBlocks l;
448 final long max = System.currentTimeMillis() + 10000;
449 do {
450 l = getNamenode(dfs.getClient()).getBlockLocations(fileName, 0, 1);
451 Assert.assertNotNull(l.getLocatedBlocks());
452 Assert.assertEquals(l.getLocatedBlocks().size(), 1);
453 Assert.assertTrue("Expecting " + repCount + " , got " + l.get(0).getLocations().length,
454 System.currentTimeMillis() < max);
455 } while (l.get(0).getLocations().length != repCount);
456
457
458 Object originalList[] = l.getLocatedBlocks().toArray();
459 HFileSystem.ReorderWALBlocks lrb = new HFileSystem.ReorderWALBlocks();
460 lrb.reorderBlocks(conf, l, fileName);
461 Assert.assertArrayEquals(originalList, l.getLocatedBlocks().toArray());
462
463
464 Assert.assertNotNull(conf.get(HConstants.HBASE_DIR));
465 Assert.assertFalse(conf.get(HConstants.HBASE_DIR).isEmpty());
466 String pseudoLogFile = conf.get(HFileSystem.HBASE_WAL_DIR) + "/" +
467 HConstants.HREGION_LOGDIR_NAME + "/" + host1 + ",6977,6576" + "/mylogfile";
468
469
470 Assert.assertNotNull("log= " + pseudoLogFile,
471 DefaultWALProvider.getServerNameFromWALDirectoryName(dfs.getConf(), pseudoLogFile));
472
473
474 lrb.reorderBlocks(conf, l, pseudoLogFile);
475 Assert.assertEquals(host1, l.get(0).getLocations()[2].getHostName());
476
477
478 lrb.reorderBlocks(conf, l, pseudoLogFile);
479 Assert.assertEquals(host1, l.get(0).getLocations()[2].getHostName());
480 }
481 }
482
483 }