1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertNotNull;
24 import static org.junit.Assert.assertNull;
25 import static org.junit.Assert.assertTrue;
26 import static org.junit.Assert.fail;
27
28 import java.io.IOException;
29 import java.lang.reflect.InvocationTargetException;
30 import java.lang.reflect.Method;
31 import java.util.List;
32 import java.util.Map;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.conf.Configuration;
37 import org.apache.hadoop.hbase.client.Admin;
38 import org.apache.hadoop.hbase.client.Get;
39 import org.apache.hadoop.hbase.client.HConnection;
40 import org.apache.hadoop.hbase.client.HConnectionManager;
41 import org.apache.hadoop.hbase.client.HTable;
42 import org.apache.hadoop.hbase.client.Put;
43 import org.apache.hadoop.hbase.client.Result;
44 import org.apache.hadoop.hbase.client.ResultScanner;
45 import org.apache.hadoop.hbase.client.Scan;
46 import org.apache.hadoop.hbase.client.Table;
47 import org.apache.hadoop.hbase.coordination.ZkSplitLogWorkerCoordination;
48 import org.apache.hadoop.hbase.master.HMaster;
49 import org.apache.hadoop.hbase.master.LoadBalancer;
50 import org.apache.hadoop.hbase.master.balancer.SimpleLoadBalancer;
51 import org.apache.hadoop.hbase.testclassification.LargeTests;
52 import org.apache.hadoop.hbase.util.Bytes;
53 import org.apache.hadoop.hbase.util.FSUtils;
54 import org.apache.hadoop.hbase.util.Threads;
55 import org.apache.hadoop.hbase.zookeeper.EmptyWatcher;
56 import org.apache.hadoop.hbase.zookeeper.ZKAssign;
57 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
58 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
59 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
60 import org.apache.zookeeper.CreateMode;
61 import org.apache.zookeeper.KeeperException;
62 import org.apache.zookeeper.ZooDefs;
63 import org.apache.zookeeper.ZooKeeper;
64 import org.apache.zookeeper.ZooKeeper.States;
65 import org.apache.zookeeper.data.ACL;
66 import org.apache.zookeeper.data.Stat;
67 import org.junit.After;
68 import org.junit.AfterClass;
69 import org.junit.Assert;
70 import org.junit.Before;
71 import org.junit.BeforeClass;
72 import org.junit.Test;
73 import org.junit.experimental.categories.Category;
74
75
76
77 @Category(LargeTests.class)
78 public class TestZooKeeper {
79 private final Log LOG = LogFactory.getLog(this.getClass());
80
81 private final static HBaseTestingUtility
82 TEST_UTIL = new HBaseTestingUtility();
83
84
85
86
87 @BeforeClass
88 public static void setUpBeforeClass() throws Exception {
89
90 Configuration conf = TEST_UTIL.getConfiguration();
91 TEST_UTIL.startMiniDFSCluster(2);
92 TEST_UTIL.startMiniZKCluster();
93 conf.setBoolean("dfs.support.append", true);
94 conf.setInt(HConstants.ZK_SESSION_TIMEOUT, 1000);
95 conf.setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, MockLoadBalancer.class,
96 LoadBalancer.class);
97 }
98
99
100
101
102 @AfterClass
103 public static void tearDownAfterClass() throws Exception {
104 TEST_UTIL.shutdownMiniCluster();
105 }
106
107
108
109
110 @Before
111 public void setUp() throws Exception {
112 TEST_UTIL.startMiniHBaseCluster(2, 2);
113 }
114
115 @After
116 public void after() throws Exception {
117 try {
118
119
120 TEST_UTIL.getHBaseCluster().killAll();
121
122
123 TEST_UTIL.shutdownMiniHBaseCluster();
124 } finally {
125 TEST_UTIL.getTestFileSystem().delete(FSUtils.getRootDir(TEST_UTIL.getConfiguration()), true);
126 ZKUtil.deleteNodeRecursively(TEST_UTIL.getZooKeeperWatcher(), "/hbase");
127 }
128 }
129
130 private ZooKeeperWatcher getZooKeeperWatcher(HConnection c)
131 throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
132 Method getterZK = c.getClass().getDeclaredMethod("getKeepAliveZooKeeperWatcher");
133 getterZK.setAccessible(true);
134 return (ZooKeeperWatcher) getterZK.invoke(c);
135 }
136
137
138
139
140
141
142
143
144
145 public void testClientSessionExpired() throws Exception {
146 Configuration c = new Configuration(TEST_UTIL.getConfiguration());
147
148
149 c.set(HConstants.HBASE_CLIENT_INSTANCE_ID, "1111");
150
151 HConnection connection = HConnectionManager.getConnection(c);
152
153 ZooKeeperWatcher connectionZK = getZooKeeperWatcher(connection);
154 LOG.info("ZooKeeperWatcher= 0x"+ Integer.toHexString(
155 connectionZK.hashCode()));
156 LOG.info("getRecoverableZooKeeper= 0x"+ Integer.toHexString(
157 connectionZK.getRecoverableZooKeeper().hashCode()));
158 LOG.info("session="+Long.toHexString(
159 connectionZK.getRecoverableZooKeeper().getSessionId()));
160
161 TEST_UTIL.expireSession(connectionZK);
162
163 LOG.info("Before using zkw state=" +
164 connectionZK.getRecoverableZooKeeper().getState());
165
166 try {
167 connectionZK.getRecoverableZooKeeper().getZooKeeper().exists(
168 "/1/1", false);
169 } catch (KeeperException ignored) {
170 }
171
172
173 States state = connectionZK.getRecoverableZooKeeper().getState();
174 LOG.info("After using zkw state=" + state);
175 LOG.info("session="+Long.toHexString(
176 connectionZK.getRecoverableZooKeeper().getSessionId()));
177
178
179 final long limit1 = System.currentTimeMillis() + 3000;
180 while (System.currentTimeMillis() < limit1 && state != States.CLOSED){
181 state = connectionZK.getRecoverableZooKeeper().getState();
182 }
183 LOG.info("After using zkw loop=" + state);
184 LOG.info("ZooKeeper should have timed out");
185 LOG.info("session="+Long.toHexString(
186 connectionZK.getRecoverableZooKeeper().getSessionId()));
187
188
189
190
191
192
193
194 ZooKeeperWatcher newConnectionZK = getZooKeeperWatcher(connection);
195
196 States state2 = newConnectionZK.getRecoverableZooKeeper().getState();
197 LOG.info("After new get state=" +state2);
198
199
200
201 final long limit2 = System.currentTimeMillis() + 3000;
202 while (System.currentTimeMillis() < limit2 &&
203 state2 != States.CONNECTED && state2 != States.CONNECTING) {
204
205 newConnectionZK = getZooKeeperWatcher(connection);
206 state2 = newConnectionZK.getRecoverableZooKeeper().getState();
207 }
208 LOG.info("After new get state loop=" + state2);
209
210 Assert.assertTrue(
211 state2 == States.CONNECTED || state2 == States.CONNECTING);
212
213 connection.close();
214 }
215
216 @Test (timeout = 120000)
217 public void testRegionServerSessionExpired() throws Exception {
218 LOG.info("Starting testRegionServerSessionExpired");
219 TEST_UTIL.expireRegionServerSession(0);
220 testSanity("testRegionServerSessionExpired");
221 }
222
223 @Test(timeout = 300000)
224 public void testMasterSessionExpired() throws Exception {
225 LOG.info("Starting testMasterSessionExpired");
226 TEST_UTIL.expireMasterSession();
227 testSanity("testMasterSessionExpired");
228 }
229
230
231
232
233
234
235 @Test(timeout = 300000)
236 public void testMasterZKSessionRecoveryFailure() throws Exception {
237 LOG.info("Starting testMasterZKSessionRecoveryFailure");
238 MiniHBaseCluster cluster = TEST_UTIL.getHBaseCluster();
239 HMaster m = cluster.getMaster();
240 m.abort("Test recovery from zk session expired",
241 new KeeperException.SessionExpiredException());
242 assertTrue(m.isStopped());
243 testSanity("testMasterZKSessionRecoveryFailure");
244 }
245
246
247
248
249
250 private void testSanity(final String testName) throws Exception{
251 String tableName = testName + "_" + System.currentTimeMillis();
252 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
253 HColumnDescriptor family = new HColumnDescriptor("fam");
254 desc.addFamily(family);
255 LOG.info("Creating table " + tableName);
256 Admin admin = TEST_UTIL.getHBaseAdmin();
257 try {
258 admin.createTable(desc);
259 } finally {
260 admin.close();
261 }
262
263 Table table =
264 new HTable(new Configuration(TEST_UTIL.getConfiguration()), desc.getTableName());
265 Put put = new Put(Bytes.toBytes("testrow"));
266 put.add(Bytes.toBytes("fam"),
267 Bytes.toBytes("col"), Bytes.toBytes("testdata"));
268 LOG.info("Putting table " + tableName);
269 table.put(put);
270 table.close();
271 }
272
273 @Test
274 public void testMultipleZK()
275 throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
276 Table localMeta =
277 new HTable(new Configuration(TEST_UTIL.getConfiguration()), TableName.META_TABLE_NAME);
278 Configuration otherConf = new Configuration(TEST_UTIL.getConfiguration());
279 otherConf.set(HConstants.ZOOKEEPER_QUORUM, "127.0.0.1");
280 Table ipMeta = new HTable(otherConf, TableName.META_TABLE_NAME);
281
282
283 final byte [] row = new byte [] {'r'};
284 localMeta.exists(new Get(row));
285 ipMeta.exists(new Get(row));
286
287
288 ZooKeeperWatcher z1 =
289 getZooKeeperWatcher(HConnectionManager.getConnection(localMeta.getConfiguration()));
290 ZooKeeperWatcher z2 =
291 getZooKeeperWatcher(HConnectionManager.getConnection(otherConf));
292 assertFalse(z1 == z2);
293 assertFalse(z1.getQuorum().equals(z2.getQuorum()));
294
295 localMeta.close();
296 ipMeta.close();
297 }
298
299
300
301
302
303 @Test
304 public void testCreateWithParents() throws Exception {
305 ZooKeeperWatcher zkw =
306 new ZooKeeperWatcher(new Configuration(TEST_UTIL.getConfiguration()),
307 TestZooKeeper.class.getName(), null);
308 byte[] expectedData = new byte[] { 1, 2, 3 };
309 ZKUtil.createWithParents(zkw, "/l1/l2/l3/l4/testCreateWithParents", expectedData);
310 byte[] data = ZKUtil.getData(zkw, "/l1/l2/l3/l4/testCreateWithParents");
311 assertTrue(Bytes.equals(expectedData, data));
312 ZKUtil.deleteNodeRecursively(zkw, "/l1");
313
314 ZKUtil.createWithParents(zkw, "/testCreateWithParents", expectedData);
315 data = ZKUtil.getData(zkw, "/testCreateWithParents");
316 assertTrue(Bytes.equals(expectedData, data));
317 ZKUtil.deleteNodeRecursively(zkw, "/testCreateWithParents");
318 }
319
320
321
322
323
324
325 @Test
326 public void testZNodeDeletes() throws Exception {
327 ZooKeeperWatcher zkw = new ZooKeeperWatcher(
328 new Configuration(TEST_UTIL.getConfiguration()),
329 TestZooKeeper.class.getName(), null);
330 ZKUtil.createWithParents(zkw, "/l1/l2/l3/l4");
331 try {
332 ZKUtil.deleteNode(zkw, "/l1/l2");
333 fail("We should not be able to delete if znode has childs");
334 } catch (KeeperException ex) {
335 assertNotNull(ZKUtil.getDataNoWatch(zkw, "/l1/l2/l3/l4", null));
336 }
337 ZKUtil.deleteNodeRecursively(zkw, "/l1/l2");
338
339 assertNull(ZKUtil.getDataNoWatch(zkw, "/l1/l2/l3/l4", null));
340
341
342 ZKUtil.deleteNodeRecursively(zkw, "/l1/l2");
343
344 ZKUtil.deleteNode(zkw, "/l1");
345 assertNull(ZKUtil.getDataNoWatch(zkw, "/l1/l2", null));
346 }
347
348 @Test
349 public void testClusterKey() throws Exception {
350 testKey("server", 2181, "hbase");
351 testKey("server1,server2,server3", 2181, "hbase");
352 try {
353 ZKUtil.transformClusterKey("2181:hbase");
354 } catch (IOException ex) {
355
356 }
357 }
358
359 @Test
360 public void testClusterKeyWithMultiplePorts() throws Exception {
361
362 testKey("server1:2182", 2181, "hbase", true);
363
364 testKey("server1:2182,server2:2183,server3:2184", 2181, "hbase", true);
365
366 testKey("server1:2182,server2,server3:2184", 2181, "hbase", true);
367
368 testKey("server1:2182,server2:2183,server3", 2181, "hbase", true);
369
370 testKey("server1:2182,server2,server3:2184,server4", 2181, "hbase", true);
371
372 testKey("server1:2182,server1:2183,server1", 2181, "hbase", true);
373
374 testKey("server1:2182,server2:2183,server1", 2181, "hbase", true);
375 }
376
377 private void testKey(String ensemble, int port, String znode)
378 throws IOException {
379 testKey(ensemble, port, znode, false);
380 }
381
382 private void testKey(String ensemble, int port, String znode, Boolean multiplePortSupport)
383 throws IOException {
384 Configuration conf = new Configuration();
385 String key = ensemble+":"+port+":"+znode;
386 String ensemble2 = null;
387 ZKUtil.ZKClusterKey zkClusterKey = ZKUtil.transformClusterKey(key);
388 if (multiplePortSupport) {
389 ensemble2 = ZKUtil.standardizeQuorumServerString(ensemble, Integer.toString(port));
390 assertEquals(ensemble2, zkClusterKey.quorumString);
391 }
392 else {
393 assertEquals(ensemble, zkClusterKey.quorumString);
394 }
395 assertEquals(port, zkClusterKey.clientPort);
396 assertEquals(znode, zkClusterKey.znodeParent);
397
398 ZKUtil.applyClusterKeyToConf(conf, key);
399 assertEquals(zkClusterKey.quorumString, conf.get(HConstants.ZOOKEEPER_QUORUM));
400 assertEquals(zkClusterKey.clientPort, conf.getInt(HConstants.ZOOKEEPER_CLIENT_PORT, -1));
401 assertEquals(zkClusterKey.znodeParent, conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT));
402
403 String reconstructedKey = ZKUtil.getZooKeeperClusterKey(conf);
404 if (multiplePortSupport) {
405 String key2 = ensemble2 + ":" + port + ":" + znode;
406 assertEquals(key2, reconstructedKey);
407 }
408 else {
409 assertEquals(key, reconstructedKey);
410 }
411 }
412
413
414
415
416
417
418
419
420 @Test
421 public void testCreateSilentIsReallySilent() throws InterruptedException,
422 KeeperException, IOException {
423 Configuration c = TEST_UTIL.getConfiguration();
424
425 String aclZnode = "/aclRoot";
426 String quorumServers = ZKConfig.getZKQuorumServersString(c);
427 int sessionTimeout = 5 * 1000;
428 ZooKeeper zk = new ZooKeeper(quorumServers, sessionTimeout, EmptyWatcher.instance);
429 zk.addAuthInfo("digest", "hbase:rox".getBytes());
430
431
432
433 ZooKeeperWatcher zk2 = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(),
434 "testCreateSilentIsReallySilent", null);
435
436
437 Stat s = null;
438 List<ACL> oldACL = null;
439 while (true) {
440 try {
441 s = new Stat();
442 oldACL = zk.getACL("/", s);
443 break;
444 } catch (KeeperException e) {
445 switch (e.code()) {
446 case CONNECTIONLOSS:
447 case SESSIONEXPIRED:
448 case OPERATIONTIMEOUT:
449 LOG.warn("Possibly transient ZooKeeper exception", e);
450 Threads.sleep(100);
451 break;
452 default:
453 throw e;
454 }
455 }
456 }
457
458
459
460 while (true) {
461 try {
462 zk.setACL("/", ZooDefs.Ids.CREATOR_ALL_ACL, -1);
463 break;
464 } catch (KeeperException e) {
465 switch (e.code()) {
466 case CONNECTIONLOSS:
467 case SESSIONEXPIRED:
468 case OPERATIONTIMEOUT:
469 LOG.warn("Possibly transient ZooKeeper exception: " + e);
470 Threads.sleep(100);
471 break;
472 default:
473 throw e;
474 }
475 }
476 }
477
478 while (true) {
479 try {
480 zk.create(aclZnode, null, ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
481 break;
482 } catch (KeeperException e) {
483 switch (e.code()) {
484 case CONNECTIONLOSS:
485 case SESSIONEXPIRED:
486 case OPERATIONTIMEOUT:
487 LOG.warn("Possibly transient ZooKeeper exception: " + e);
488 Threads.sleep(100);
489 break;
490 default:
491 throw e;
492 }
493 }
494 }
495 zk.close();
496 ZKUtil.createAndFailSilent(zk2, aclZnode);
497
498
499 ZooKeeper zk3 = new ZooKeeper(quorumServers, sessionTimeout, EmptyWatcher.instance);
500 zk3.addAuthInfo("digest", "hbase:rox".getBytes());
501 try {
502 zk3.setACL("/", oldACL, -1);
503 } finally {
504 zk3.close();
505 }
506 }
507
508
509
510
511
512 @Test
513 @SuppressWarnings("deprecation")
514 public void testGetChildDataAndWatchForNewChildrenShouldNotThrowNPE()
515 throws Exception {
516 ZooKeeperWatcher zkw = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(),
517 "testGetChildDataAndWatchForNewChildrenShouldNotThrowNPE", null);
518 ZKUtil.getChildDataAndWatchForNewChildren(zkw, "/wrongNode");
519 }
520
521
522
523
524
525
526 @Test(timeout = 300000)
527 public void testRegionAssignmentAfterMasterRecoveryDueToZKExpiry() throws Exception {
528 MiniHBaseCluster cluster = TEST_UTIL.getHBaseCluster();
529 cluster.startRegionServer();
530 cluster.waitForActiveAndReadyMaster(10000);
531 HMaster m = cluster.getMaster();
532 final ZooKeeperWatcher zkw = m.getZooKeeper();
533
534 try (Admin admin = TEST_UTIL.getHBaseAdmin()) {
535 byte[][] SPLIT_KEYS = new byte[][] { Bytes.toBytes("a"), Bytes.toBytes("b"),
536 Bytes.toBytes("c"), Bytes.toBytes("d"), Bytes.toBytes("e"), Bytes.toBytes("f"),
537 Bytes.toBytes("g"), Bytes.toBytes("h"), Bytes.toBytes("i"), Bytes.toBytes("j") };
538 String tableName = "testRegionAssignmentAfterMasterRecoveryDueToZKExpiry";
539 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
540 htd.addFamily(new HColumnDescriptor(HConstants.CATALOG_FAMILY));
541 admin.createTable(htd, SPLIT_KEYS);
542 TEST_UTIL.waitUntilNoRegionsInTransition(60000);
543 m.getZooKeeper().close();
544 MockLoadBalancer.retainAssignCalled = false;
545 final int expectedNumOfListeners = countPermanentListeners(zkw);
546 m.abort("Test recovery from zk session expired",
547 new KeeperException.SessionExpiredException());
548 assertTrue(m.isStopped());
549
550
551 assertFalse("Retain assignment should not be called", MockLoadBalancer.retainAssignCalled);
552
553
554 cluster.waitForActiveAndReadyMaster(120000);
555 final HMaster newMaster = cluster.getMasterThread().getMaster();
556 assertEquals(expectedNumOfListeners, countPermanentListeners(newMaster.getZooKeeper()));
557 }
558 }
559
560
561
562
563
564 private int countPermanentListeners(ZooKeeperWatcher watcher) {
565 return countListeners(watcher, ZkSplitLogWorkerCoordination.class);
566 }
567
568
569
570
571 private int countListeners(ZooKeeperWatcher watcher, Class<?>... exclude) {
572 int cnt = 0;
573 for (Object o : watcher.getListeners()) {
574 boolean skip = false;
575 for (Class<?> aClass : exclude) {
576 if (aClass.isAssignableFrom(o.getClass())) {
577 skip = true;
578 break;
579 }
580 }
581 if (!skip) {
582 cnt += 1;
583 }
584 }
585 return cnt;
586 }
587
588
589
590
591
592
593 @Test(timeout = 300000)
594 public void testLogSplittingAfterMasterRecoveryDueToZKExpiry() throws IOException,
595 KeeperException, InterruptedException {
596 MiniHBaseCluster cluster = TEST_UTIL.getHBaseCluster();
597 cluster.startRegionServer();
598 HMaster m = cluster.getMaster();
599
600 Admin admin = TEST_UTIL.getHBaseAdmin();
601 Table table = null;
602 try {
603 byte[][] SPLIT_KEYS = new byte[][] { Bytes.toBytes("1"), Bytes.toBytes("2"),
604 Bytes.toBytes("3"), Bytes.toBytes("4"), Bytes.toBytes("5") };
605
606 String tableName = "testLogSplittingAfterMasterRecoveryDueToZKExpiry";
607 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
608 HColumnDescriptor hcd = new HColumnDescriptor("col");
609 htd.addFamily(hcd);
610 admin.createTable(htd, SPLIT_KEYS);
611 ZooKeeperWatcher zooKeeperWatcher = HBaseTestingUtility.getZooKeeperWatcher(TEST_UTIL);
612 ZKAssign.blockUntilNoRIT(zooKeeperWatcher);
613 table = new HTable(TEST_UTIL.getConfiguration(), htd.getTableName());
614 Put p;
615 int numberOfPuts;
616 for (numberOfPuts = 0; numberOfPuts < 6; numberOfPuts++) {
617 p = new Put(Bytes.toBytes(numberOfPuts));
618 p.add(Bytes.toBytes("col"), Bytes.toBytes("ql"), Bytes.toBytes("value" + numberOfPuts));
619 table.put(p);
620 }
621 m.getZooKeeper().close();
622 m.abort("Test recovery from zk session expired",
623 new KeeperException.SessionExpiredException());
624 assertTrue(m.isStopped());
625 cluster.getRegionServer(0).abort("Aborting");
626
627
628 Scan scan = new Scan();
629 int numberOfRows = 0;
630 ResultScanner scanner = table.getScanner(scan);
631 Result[] result = scanner.next(1);
632 while (result != null && result.length > 0) {
633 numberOfRows++;
634 result = scanner.next(1);
635 }
636 assertEquals("Number of rows should be equal to number of puts.", numberOfPuts,
637 numberOfRows);
638 } finally {
639 if (table != null) table.close();
640 admin.close();
641 }
642 }
643
644 static class MockLoadBalancer extends SimpleLoadBalancer {
645 static boolean retainAssignCalled = false;
646
647 @Override
648 public Map<ServerName, List<HRegionInfo>> retainAssignment(
649 Map<HRegionInfo, ServerName> regions, List<ServerName> servers) {
650 retainAssignCalled = true;
651 return super.retainAssignment(regions, servers);
652 }
653 }
654
655 }
656