View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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.hbase.ServerName;
29  import org.apache.hadoop.hbase.zookeeper.ZKConfig;
30  import org.apache.zookeeper.KeeperException;
31  import org.junit.Before;
32  import org.junit.Test;
33  
34  /**
35   * White box testing for replication state interfaces. Implementations should extend this class, and
36   * initialize the interfaces properly.
37   */
38  public abstract class TestReplicationStateBasic {
39  
40    protected ReplicationQueues rq1;
41    protected ReplicationQueues rq2;
42    protected ReplicationQueues rq3;
43    protected ReplicationQueuesClient rqc;
44    protected String server1 = ServerName.valueOf("hostname1.example.org", 1234, -1L).toString();
45    protected String server2 = ServerName.valueOf("hostname2.example.org", 1234, -1L).toString();
46    protected String server3 = ServerName.valueOf("hostname3.example.org", 1234, -1L).toString();
47    protected ReplicationPeers rp;
48    protected static final String ID_ONE = "1";
49    protected static final String ID_TWO = "2";
50    protected static String KEY_ONE;
51    protected static String KEY_TWO;
52  
53    // For testing when we try to replicate to ourself
54    protected String OUR_ID = "3";
55    protected String OUR_KEY;
56  
57    protected static int zkTimeoutCount;
58    protected static final int ZK_MAX_COUNT = 300;
59    protected static final int ZK_SLEEP_INTERVAL = 100; // millis
60  
61    private static final Log LOG = LogFactory.getLog(TestReplicationStateBasic.class);
62  
63    @Before
64    public void setUp() {
65      zkTimeoutCount = 0;
66    }
67  
68    @Test
69    public void testReplicationQueuesClient() throws ReplicationException, KeeperException {
70      rqc.init();
71      // Test methods with empty state
72      assertEquals(0, rqc.getListOfReplicators().size());
73      assertNull(rqc.getLogsInQueue(server1, "qId1"));
74      assertNull(rqc.getAllQueues(server1));
75  
76      /*
77       * Set up data Two replicators: -- server1: three queues with 0, 1 and 2 log files each --
78       * server2: zero queues
79       */
80      rq1.init(server1);
81      rq2.init(server2);
82      rq1.addLog("qId1", "trash");
83      rq1.removeLog("qId1", "trash");
84      rq1.addLog("qId2", "filename1");
85      rq1.addLog("qId3", "filename2");
86      rq1.addLog("qId3", "filename3");
87      rq2.addLog("trash", "trash");
88      rq2.removeQueue("trash");
89  
90      List<String> reps = rqc.getListOfReplicators();
91      assertEquals(2, reps.size());
92      assertTrue(server1, reps.contains(server1));
93      assertTrue(server2, reps.contains(server2));
94  
95      assertNull(rqc.getLogsInQueue("bogus", "bogus"));
96      assertNull(rqc.getLogsInQueue(server1, "bogus"));
97      assertEquals(0, rqc.getLogsInQueue(server1, "qId1").size());
98      assertEquals(1, rqc.getLogsInQueue(server1, "qId2").size());
99      assertEquals("filename1", rqc.getLogsInQueue(server1, "qId2").get(0));
100 
101     assertNull(rqc.getAllQueues("bogus"));
102     assertEquals(0, rqc.getAllQueues(server2).size());
103     List<String> list = rqc.getAllQueues(server1);
104     assertEquals(3, list.size());
105     assertTrue(list.contains("qId2"));
106     assertTrue(list.contains("qId3"));
107   }
108 
109   @Test
110   public void testReplicationQueues() throws ReplicationException {
111     rq1.init(server1);
112     rq2.init(server2);
113     rq3.init(server3);
114     //Initialize ReplicationPeer so we can add peers (we don't transfer lone queues)
115     rp.init();
116 
117     // 3 replicators should exist
118     assertEquals(3, rq1.getListOfReplicators().size());
119     rq1.removeQueue("bogus");
120     rq1.removeLog("bogus", "bogus");
121     rq1.removeAllQueues();
122     assertNull(rq1.getAllQueues());
123     assertEquals(0, rq1.getLogPosition("bogus", "bogus"));
124     assertNull(rq1.getLogsInQueue("bogus"));
125     assertNull(rq1.getUnClaimedQueueIds(ServerName.valueOf("bogus", 1234, -1L).toString()));
126 
127     rq1.setLogPosition("bogus", "bogus", 5L);
128 
129     populateQueues();
130 
131     assertEquals(3, rq1.getListOfReplicators().size());
132     assertEquals(0, rq2.getLogsInQueue("qId1").size());
133     assertEquals(5, rq3.getLogsInQueue("qId5").size());
134     assertEquals(0, rq3.getLogPosition("qId1", "filename0"));
135     rq3.setLogPosition("qId5", "filename4", 354L);
136     assertEquals(354L, rq3.getLogPosition("qId5", "filename4"));
137 
138     assertEquals(5, rq3.getLogsInQueue("qId5").size());
139     assertEquals(0, rq2.getLogsInQueue("qId1").size());
140     assertEquals(0, rq1.getAllQueues().size());
141     assertEquals(1, rq2.getAllQueues().size());
142     assertEquals(5, rq3.getAllQueues().size());
143 
144     assertEquals(0, rq3.getUnClaimedQueueIds(server1).size());
145     rq3.removeReplicatorIfQueueIsEmpty(server1);
146     assertEquals(2, rq3.getListOfReplicators().size());
147 
148     List<String> queues = rq2.getUnClaimedQueueIds(server3);
149     assertEquals(5, queues.size());
150     for(String queue: queues) {
151       rq2.claimQueue(server3, queue);
152     }
153     rq2.removeReplicatorIfQueueIsEmpty(server3);
154     assertEquals(1, rq2.getListOfReplicators().size());
155 
156     // Try to claim our own queues
157     assertNull(rq2.getUnClaimedQueueIds(server2));
158     rq2.removeReplicatorIfQueueIsEmpty(server2);
159 
160     assertEquals(6, rq2.getAllQueues().size());
161 
162     rq2.removeAllQueues();
163 
164     assertEquals(0, rq2.getListOfReplicators().size());
165   }
166 
167   @Test
168   public void testHfileRefsReplicationQueues() throws ReplicationException, KeeperException {
169     rp.init();
170     rq1.init(server1);
171     rqc.init();
172 
173     List<String> files1 = new ArrayList<String>(3);
174     files1.add("file_1");
175     files1.add("file_2");
176     files1.add("file_3");
177     assertNull(rqc.getReplicableHFiles(ID_ONE));
178     assertEquals(0, rqc.getAllPeersFromHFileRefsQueue().size());
179     rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE), null);
180     rq1.addPeerToHFileRefs(ID_ONE);
181     rq1.addHFileRefs(ID_ONE, files1);
182     assertEquals(1, rqc.getAllPeersFromHFileRefsQueue().size());
183     assertEquals(3, rqc.getReplicableHFiles(ID_ONE).size());
184     List<String> files2 = new ArrayList<>(files1);
185     String removedString = files2.remove(0);
186     rq1.removeHFileRefs(ID_ONE, files2);
187     assertEquals(1, rqc.getReplicableHFiles(ID_ONE).size());
188     files2 = new ArrayList<>(1);
189     files2.add(removedString);
190     rq1.removeHFileRefs(ID_ONE, files2);
191     assertEquals(0, rqc.getReplicableHFiles(ID_ONE).size());
192     rp.removePeer(ID_ONE);
193   }
194 
195   @Test
196   public void testRemovePeerForHFileRefs() throws ReplicationException, KeeperException {
197     rq1.init(server1);
198     rqc.init();
199 
200     rp.init();
201     rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE), null);
202     rp.addPeer(ID_TWO, new ReplicationPeerConfig().setClusterKey(KEY_TWO), null);
203 
204     List<String> files1 = new ArrayList<String>(3);
205     files1.add("file_1");
206     files1.add("file_2");
207     files1.add("file_3");
208     rq1.addPeerToHFileRefs(ID_ONE);
209     rq1.addHFileRefs(ID_ONE, files1);
210     rq1.addPeerToHFileRefs(ID_TWO);
211     rq1.addHFileRefs(ID_TWO, files1);
212     assertEquals(2, rqc.getAllPeersFromHFileRefsQueue().size());
213     assertEquals(3, rqc.getReplicableHFiles(ID_ONE).size());
214     assertEquals(3, rqc.getReplicableHFiles(ID_TWO).size());
215 
216     rp.removePeer(ID_ONE);
217     rq1.removePeerFromHFileRefs(ID_ONE);
218     assertEquals(1, rqc.getAllPeersFromHFileRefsQueue().size());
219     assertNull(rqc.getReplicableHFiles(ID_ONE));
220     assertEquals(3, rqc.getReplicableHFiles(ID_TWO).size());
221 
222     rp.removePeer(ID_TWO);
223     rq1.removePeerFromHFileRefs(ID_TWO);
224     assertEquals(0, rqc.getAllPeersFromHFileRefsQueue().size());
225     assertNull(rqc.getReplicableHFiles(ID_TWO));
226   }
227 
228   @Test
229   public void testReplicationPeers() throws Exception {
230     rp.init();
231 
232     // Test methods with non-existent peer ids
233     try {
234       rp.removePeer("bogus");
235       fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
236     } catch (IllegalArgumentException e) {
237     }
238     try {
239       rp.enablePeer("bogus");
240       fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
241     } catch (IllegalArgumentException e) {
242     }
243     try {
244       rp.disablePeer("bogus");
245       fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
246     } catch (IllegalArgumentException e) {
247     }
248     try {
249       rp.getStatusOfPeer("bogus");
250       fail("Should have thrown an IllegalArgumentException when passed a bogus peerId");
251     } catch (IllegalArgumentException e) {
252     }
253     assertFalse(rp.peerAdded("bogus"));
254     rp.peerRemoved("bogus");
255 
256     assertNull(rp.getPeerConf("bogus"));
257     assertNumberOfPeers(0);
258 
259     // Add some peers
260     rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE), null);
261     assertNumberOfPeers(1);
262     rp.addPeer(ID_TWO, new ReplicationPeerConfig().setClusterKey(KEY_TWO), null);
263     assertNumberOfPeers(2);
264 
265     // Test methods with a peer that is added but not connected
266     try {
267       rp.getStatusOfPeer(ID_ONE);
268       fail("There are no connected peers, should have thrown an IllegalArgumentException");
269     } catch (IllegalArgumentException e) {
270     }
271     assertEquals(KEY_ONE, ZKConfig.getZooKeeperClusterKey(rp.getPeerConf(ID_ONE).getSecond()));
272     rp.removePeer(ID_ONE);
273     rp.peerRemoved(ID_ONE);
274     assertNumberOfPeers(1);
275 
276     // Add one peer
277     rp.addPeer(ID_ONE, new ReplicationPeerConfig().setClusterKey(KEY_ONE), null);
278     rp.peerAdded(ID_ONE);
279     assertNumberOfPeers(2);
280     assertTrue(rp.getStatusOfPeer(ID_ONE));
281     rp.disablePeer(ID_ONE);
282     assertConnectedPeerStatus(false, ID_ONE);
283     rp.enablePeer(ID_ONE);
284     assertConnectedPeerStatus(true, ID_ONE);
285 
286     // Disconnect peer
287     rp.peerRemoved(ID_ONE);
288     assertNumberOfPeers(2);
289     try {
290       rp.getStatusOfPeer(ID_ONE);
291       fail("There are no connected peers, should have thrown an IllegalArgumentException");
292     } catch (IllegalArgumentException e) {
293     }
294   }
295 
296   protected void assertConnectedPeerStatus(boolean status, String peerId) throws Exception {
297     // we can first check if the value was changed in the store, if it wasn't then fail right away
298     if (status != rp.getStatusOfPeerFromBackingStore(peerId)) {
299       fail("ConnectedPeerStatus was " + !status + " but expected " + status + " in ZK");
300     }
301     while (true) {
302       if (status == rp.getStatusOfPeer(peerId)) {
303         return;
304       }
305       if (zkTimeoutCount < ZK_MAX_COUNT) {
306         LOG.debug("ConnectedPeerStatus was " + !status + " but expected " + status
307             + ", sleeping and trying again.");
308         Thread.sleep(ZK_SLEEP_INTERVAL);
309       } else {
310         fail("Timed out waiting for ConnectedPeerStatus to be " + status);
311       }
312     }
313   }
314 
315   protected void assertNumberOfPeers(int total) {
316     assertEquals(total, rp.getAllPeerConfigs().size());
317     assertEquals(total, rp.getAllPeerIds().size());
318     assertEquals(total, rp.getAllPeerIds().size());
319   }
320 
321   /*
322    * three replicators: rq1 has 0 queues, rq2 has 1 queue with no logs, rq3 has 5 queues with 1, 2,
323    * 3, 4, 5 log files respectively
324    */
325   protected void populateQueues() throws ReplicationException {
326     rq1.addLog("trash", "trash");
327     rq1.removeQueue("trash");
328 
329     rq2.addLog("qId1", "trash");
330     rq2.removeLog("qId1", "trash");
331 
332     for (int i = 1; i < 6; i++) {
333       for (int j = 0; j < i; j++) {
334         rq3.addLog("qId" + i, "filename" + j);
335       }
336       //Add peers for the corresponding queues so they are not orphans
337       rp.addPeer("qId" + i, new ReplicationPeerConfig().setClusterKey("bogus" + i), null);
338     }
339   }
340 }
341