1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.zookeeper;
19
20 import static org.junit.Assert.assertTrue;
21
22 import java.io.IOException;
23 import java.lang.reflect.Field;
24 import java.util.Properties;
25
26 import org.apache.hadoop.conf.Configuration;
27 import org.apache.hadoop.hbase.Abortable;
28 import org.apache.hadoop.hbase.HBaseTestingUtility;
29 import org.apache.hadoop.hbase.HConstants;
30 import org.apache.hadoop.hbase.testclassification.MediumTests;
31 import org.apache.hadoop.hbase.util.Bytes;
32 import org.apache.zookeeper.CreateMode;
33 import org.apache.zookeeper.KeeperException;
34 import org.apache.zookeeper.Watcher;
35 import org.apache.zookeeper.ZooDefs.Ids;
36 import org.apache.zookeeper.ZooKeeper;
37 import org.apache.zookeeper.data.Stat;
38 import org.junit.AfterClass;
39 import org.junit.BeforeClass;
40 import org.junit.Test;
41 import org.junit.experimental.categories.Category;
42
43 @Category(MediumTests.class)
44 public class TestRecoverableZooKeeper {
45
46 private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
47
48 Abortable abortable = new Abortable() {
49 @Override
50 public void abort(String why, Throwable e) {
51
52 }
53
54 @Override
55 public boolean isAborted() {
56 return false;
57 }
58 };
59
60 @BeforeClass
61 public static void setUpBeforeClass() throws Exception {
62 TEST_UTIL.startMiniZKCluster();
63 }
64
65 @AfterClass
66 public static void tearDownAfterClass() throws Exception {
67 TEST_UTIL.shutdownMiniZKCluster();
68 }
69
70 @Test
71 public void testSetDataVersionMismatchInLoop() throws Exception {
72 String znode = "/hbase/splitWAL/9af7cfc9b15910a0b3d714bf40a3248f";
73 Configuration conf = TEST_UTIL.getConfiguration();
74 ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "testSetDataVersionMismatchInLoop",
75 abortable, true);
76 String ensemble = ZKConfig.getZKQuorumServersString(conf);
77 RecoverableZooKeeper rzk = ZKUtil.connect(conf, ensemble, zkw);
78 rzk.create(znode, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
79 rzk.setData(znode, "OPENING".getBytes(), 0);
80 Field zkField = RecoverableZooKeeper.class.getDeclaredField("zk");
81 zkField.setAccessible(true);
82 int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
83 ZookeeperStub zkStub = new ZookeeperStub(ensemble, timeout, zkw);
84 zkStub.setThrowExceptionInNumOperations(1);
85 zkField.set(rzk, zkStub);
86 byte[] opened = "OPENED".getBytes();
87 rzk.setData(znode, opened, 1);
88 byte[] data = rzk.getData(znode, false, new Stat());
89 assertTrue(Bytes.equals(opened, data));
90 }
91
92 class ZookeeperStub extends ZooKeeper {
93
94 private int throwExceptionInNumOperations;
95
96 public ZookeeperStub(String connectString, int sessionTimeout, Watcher watcher)
97 throws IOException {
98 super(connectString, sessionTimeout, watcher);
99 }
100
101 public void setThrowExceptionInNumOperations(int throwExceptionInNumOperations) {
102 this.throwExceptionInNumOperations = throwExceptionInNumOperations;
103 }
104
105 private void checkThrowKeeperException() throws KeeperException {
106 if (throwExceptionInNumOperations == 1) {
107 throwExceptionInNumOperations = 0;
108 throw new KeeperException.ConnectionLossException();
109 }
110 if (throwExceptionInNumOperations > 0)
111 throwExceptionInNumOperations--;
112 }
113
114 @Override
115 public Stat setData(String path, byte[] data, int version) throws KeeperException,
116 InterruptedException {
117 Stat stat = super.setData(path, data, version);
118 checkThrowKeeperException();
119 return stat;
120 }
121 }
122 }