1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.replication;
20
21 import static org.junit.Assert.*;
22
23 import java.util.ArrayList;
24 import java.util.List;
25
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.fs.Path;
29 import org.apache.hadoop.hbase.ServerName;
30 import org.apache.hadoop.hbase.util.Pair;
31 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
32 import org.apache.zookeeper.KeeperException;
33 import org.junit.Before;
34 import org.junit.Test;
35
36
37
38
39
40 public abstract class TestReplicationStateBasic {
41
42 protected ReplicationQueues rq1;
43 protected ReplicationQueues rq2;
44 protected ReplicationQueues rq3;
45 protected ReplicationQueuesClient rqc;
46 protected String server1 = ServerName.valueOf("hostname1.example.org", 1234, -1L).toString();
47 protected String server2 = ServerName.valueOf("hostname2.example.org", 1234, -1L).toString();
48 protected String server3 = ServerName.valueOf("hostname3.example.org", 1234, -1L).toString();
49 protected ReplicationPeers rp;
50 protected static final String ID_ONE = "1";
51 protected static final String ID_TWO = "2";
52 protected static String KEY_ONE;
53 protected static String KEY_TWO;
54
55
56 protected String OUR_ID = "3";
57 protected String OUR_KEY;
58
59 protected static int zkTimeoutCount;
60 protected static final int ZK_MAX_COUNT = 300;
61 protected static final int ZK_SLEEP_INTERVAL = 100;
62
63 private static final Log LOG = LogFactory.getLog(TestReplicationStateBasic.class);
64
65 @Before
66 public void setUp() {
67 zkTimeoutCount = 0;
68 }
69
70 @Test
71 public void testReplicationQueuesClient() throws ReplicationException, KeeperException {
72 rqc.init();
73
74 assertEquals(0, rqc.getListOfReplicators().size());
75 assertNull(rqc.getLogsInQueue(server1, "qId1"));
76 assertNull(rqc.getAllQueues(server1));
77
78
79
80
81
82 rq1.init(server1);
83 rq2.init(server2);
84 rq1.addLog("qId1", "trash");
85 rq1.removeLog("qId1", "trash");
86 rq1.addLog("qId2", "filename1");
87 rq1.addLog("qId3", "filename2");
88 rq1.addLog("qId3", "filename3");
89 rq2.addLog("trash", "trash");
90 rq2.removeQueue("trash");
91
92 List<String> reps = rqc.getListOfReplicators();
93 assertEquals(2, reps.size());
94 assertTrue(server1, reps.contains(server1));
95 assertTrue(server2, reps.contains(server2));
96
97 assertNull(rqc.getLogsInQueue("bogus", "bogus"));
98 assertNull(rqc.getLogsInQueue(server1, "bogus"));
99 assertEquals(0, rqc.getLogsInQueue(server1, "qId1").size());
100 assertEquals(1, rqc.getLogsInQueue(server1, "qId2").size());
101 assertEquals("filename1", rqc.getLogsInQueue(server1, "qId2").get(0));
102
103 assertNull(rqc.getAllQueues("bogus"));
104 assertEquals(0, rqc.getAllQueues(server2).size());
105 List<String> list = rqc.getAllQueues(server1);
106 assertEquals(3, list.size());
107 assertTrue(list.contains("qId2"));
108 assertTrue(list.contains("qId3"));
109 }
110
111 @Test
112 public void testReplicationQueues() throws ReplicationException {
113 rq1.init(server1);
114 rq2.init(server2);
115 rq3.init(server3);
116
117 rp.init();
118
119
120 assertEquals(3, rq1.getListOfReplicators().size());
121 rq1.removeQueue("bogus");
122 rq1.removeLog("bogus", "bogus");
123 rq1.removeAllQueues();
124 assertNull(rq1.getAllQueues());
125 assertEquals(0, rq1.getLogPosition("bogus", "bogus"));
126 assertNull(rq1.getLogsInQueue("bogus"));
127 assertNull(rq1.getUnClaimedQueueIds(ServerName.valueOf("bogus", 1234, -1L).toString()));
128
129 rq1.setLogPosition("bogus", "bogus", 5L);
130
131 populateQueues();
132
133 assertEquals(3, rq1.getListOfReplicators().size());
134 assertEquals(0, rq2.getLogsInQueue("qId1").size());
135 assertEquals(5, rq3.getLogsInQueue("qId5").size());
136 assertEquals(0, rq3.getLogPosition("qId1", "filename0"));
137 rq3.setLogPosition("qId5", "filename4", 354L);
138 assertEquals(354L, rq3.getLogPosition("qId5", "filename4"));
139
140 assertEquals(5, rq3.getLogsInQueue("qId5").size());
141 assertEquals(0, rq2.getLogsInQueue("qId1").size());
142 assertEquals(0, rq1.getAllQueues().size());
143 assertEquals(1, rq2.getAllQueues().size());
144 assertEquals(5, rq3.getAllQueues().size());
145
146 assertEquals(0, rq3.getUnClaimedQueueIds(server1).size());
147 rq3.removeReplicatorIfQueueIsEmpty(server1);
148 assertEquals(2, rq3.getListOfReplicators().size());
149
150 List<String> queues = rq2.getUnClaimedQueueIds(server3);
151 assertEquals(5, queues.size());
152 for(String queue: queues) {
153 rq2.claimQueue(server3, queue);
154 }
155 rq2.removeReplicatorIfQueueIsEmpty(server3);
156 assertEquals(1, rq2.getListOfReplicators().size());
157
158
159 assertNull(rq2.getUnClaimedQueueIds(server2));
160 rq2.removeReplicatorIfQueueIsEmpty(server2);
161
162 assertEquals(6, rq2.getAllQueues().size());
163
164 rq2.removeAllQueues();
165
166 assertEquals(0, rq2.getListOfReplicators().size());
167 }
168
169 @Test
170 public void testInvalidClusterKeys() throws ReplicationException, KeeperException {
171 rp.init();
172
173 try {
174 rp.addPeer(ID_ONE,
175 new ReplicationPeerConfig().setClusterKey("hostname1.example.org:1234:hbase"));
176 fail("Should throw an IllegalArgumentException because "
177 + "zookeeper.znode.parent is missing leading '/'.");
178 } catch (IllegalArgumentException e) {
179
180 }
181
182 try {
183 rp.addPeer(ID_ONE,
184 new ReplicationPeerConfig().setClusterKey("hostname1.example.org:1234:/"));
185 fail("Should throw an IllegalArgumentException because zookeeper.znode.parent is missing.");
186 } catch (IllegalArgumentException e) {
187
188 }
189
190 try {
191 rp.addPeer(ID_ONE,
192 new ReplicationPeerConfig().setClusterKey("hostname1.example.org::/hbase"));
193 fail("Should throw an IllegalArgumentException because "
194 + "hbase.zookeeper.property.clientPort is missing.");
195 } catch (IllegalArgumentException e) {
196
197 }
198 }
199
200 @Test
201 public void testHfileRefsReplicationQueues() throws ReplicationException, KeeperException {
202 rp.init();
203 rq1.init(server1);
204 rqc.init();
205
206 List<Pair<Path, Path>> files1 = new ArrayList<>(3);
207 files1.add(new Pair<Path, Path>(null, new Path("file_1")));
208 files1.add(new Pair<Path, Path>(null, new Path("file_2")));
209 files1.add(new Pair<Path, Path>(null, new Path("file_3")));
210 assertNull(rqc.getReplicableHFiles(ID_ONE));
211 assertEquals(0, rqc.getAllPeersFromHFileRefsQueue().size());
212 rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE));
213 rq1.addPeerToHFileRefs(ID_ONE);
214 rq1.addHFileRefs(ID_ONE, files1);
215 assertEquals(1, rqc.getAllPeersFromHFileRefsQueue().size());
216 assertEquals(3, rqc.getReplicableHFiles(ID_ONE).size());
217 List<String> hfiles2 = new ArrayList<>();
218 for (Pair<Path, Path> p : files1) {
219 hfiles2.add(p.getSecond().getName());
220 }
221 String removedString = hfiles2.remove(0);
222 rq1.removeHFileRefs(ID_ONE, hfiles2);
223 assertEquals(1, rqc.getReplicableHFiles(ID_ONE).size());
224 hfiles2 = new ArrayList<>(1);
225 hfiles2.add(removedString);
226 rq1.removeHFileRefs(ID_ONE, hfiles2);
227 assertEquals(0, rqc.getReplicableHFiles(ID_ONE).size());
228 rp.removePeer(ID_ONE);
229 }
230
231 @Test
232 public void testRemovePeerForHFileRefs() throws ReplicationException, KeeperException {
233 rq1.init(server1);
234 rqc.init();
235
236 rp.init();
237 rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE));
238 rp.addPeer(ID_TWO, new ReplicationPeerConfig().setClusterKey(KEY_TWO));
239
240 List<Pair<Path, Path>> files1 = new ArrayList<>(3);
241 files1.add(new Pair<Path, Path>(null, new Path("file_1")));
242 files1.add(new Pair<Path, Path>(null, new Path("file_2")));
243 files1.add(new Pair<Path, Path>(null, new Path("file_3")));
244 rq1.addPeerToHFileRefs(ID_ONE);
245 rq1.addHFileRefs(ID_ONE, files1);
246 rq1.addPeerToHFileRefs(ID_TWO);
247 rq1.addHFileRefs(ID_TWO, files1);
248 assertEquals(2, rqc.getAllPeersFromHFileRefsQueue().size());
249 assertEquals(3, rqc.getReplicableHFiles(ID_ONE).size());
250 assertEquals(3, rqc.getReplicableHFiles(ID_TWO).size());
251
252 rp.removePeer(ID_ONE);
253 rq1.removePeerFromHFileRefs(ID_ONE);
254 assertEquals(1, rqc.getAllPeersFromHFileRefsQueue().size());
255 assertNull(rqc.getReplicableHFiles(ID_ONE));
256 assertEquals(3, rqc.getReplicableHFiles(ID_TWO).size());
257
258 rp.removePeer(ID_TWO);
259 rq1.removePeerFromHFileRefs(ID_TWO);
260 assertEquals(0, rqc.getAllPeersFromHFileRefsQueue().size());
261 assertNull(rqc.getReplicableHFiles(ID_TWO));
262 }
263
264 @Test
265 public void testReplicationPeers() throws Exception {
266 rp.init();
267
268
269 try {
270 rp.removePeer("bogus");
271 fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
272 } catch (IllegalArgumentException e) {
273 }
274 try {
275 rp.enablePeer("bogus");
276 fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
277 } catch (IllegalArgumentException e) {
278 }
279 try {
280 rp.disablePeer("bogus");
281 fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
282 } catch (IllegalArgumentException e) {
283 }
284 try {
285 rp.getStatusOfPeer("bogus");
286 fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
287 } catch (IllegalArgumentException e) {
288 }
289 assertFalse(rp.peerAdded("bogus"));
290 rp.peerRemoved("bogus");
291
292 assertNull(rp.getPeerConf("bogus"));
293 assertNumberOfPeers(0);
294
295
296 rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE));
297 assertNumberOfPeers(1);
298 rp.addPeer(ID_TWO, new ReplicationPeerConfig().setClusterKey(KEY_TWO));
299 assertNumberOfPeers(2);
300
301
302 try {
303 rp.getStatusOfPeer(ID_ONE);
304 fail("There are no connected peers, should have thrown an IllegalArgumentException");
305 } catch (IllegalArgumentException e) {
306 }
307 assertEquals(KEY_ONE, ZKConfig.getZooKeeperClusterKey(rp.getPeerConf(ID_ONE).getSecond()));
308 rp.removePeer(ID_ONE);
309 rp.peerRemoved(ID_ONE);
310 assertNumberOfPeers(1);
311
312
313 rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE));
314 rp.peerAdded(ID_ONE);
315 assertNumberOfPeers(2);
316 assertTrue(rp.getStatusOfPeer(ID_ONE));
317 rp.disablePeer(ID_ONE);
318 assertConnectedPeerStatus(false, ID_ONE);
319 rp.enablePeer(ID_ONE);
320 assertConnectedPeerStatus(true, ID_ONE);
321
322
323 rp.peerRemoved(ID_ONE);
324 assertNumberOfPeers(2);
325 try {
326 rp.getStatusOfPeer(ID_ONE);
327 fail("There are no connected peers, should have thrown an IllegalArgumentException");
328 } catch (IllegalArgumentException e) {
329 }
330 }
331
332 protected void assertConnectedPeerStatus(boolean status, String peerId) throws Exception {
333
334 if (status != rp.getStatusOfPeerFromBackingStore(peerId)) {
335 fail("ConnectedPeerStatus was " + !status + " but expected " + status + " in ZK");
336 }
337 while (true) {
338 if (status == rp.getStatusOfPeer(peerId)) {
339 return;
340 }
341 if (zkTimeoutCount < ZK_MAX_COUNT) {
342 LOG.debug("ConnectedPeerStatus was " + !status + " but expected " + status
343 + ", sleeping and trying again.");
344 Thread.sleep(ZK_SLEEP_INTERVAL);
345 } else {
346 fail("Timed out waiting for ConnectedPeerStatus to be " + status);
347 }
348 }
349 }
350
351 protected void assertNumberOfPeers(int total) {
352 assertEquals(total, rp.getAllPeerConfigs().size());
353 assertEquals(total, rp.getAllPeerIds().size());
354 assertEquals(total, rp.getAllPeerIds().size());
355 }
356
357
358
359
360
361 protected void populateQueues() throws ReplicationException {
362 rq1.addLog("trash", "trash");
363 rq1.removeQueue("trash");
364
365 rq2.addLog("qId1", "trash");
366 rq2.removeLog("qId1", "trash");
367
368 for (int i = 1; i < 6; i++) {
369 for (int j = 0; j < i; j++) {
370 rq3.addLog("qId" + i, "filename" + j);
371 }
372
373 rp.addPeer("qId" + i, new ReplicationPeerConfig().setClusterKey("localhost:2818:/bogus" + i));
374 }
375 }
376 }
377