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.regionserver;
20  
21  import static org.junit.Assert.*;
22  
23  import java.io.IOException;
24  import java.util.Arrays;
25  import java.util.Collection;
26  import java.util.concurrent.atomic.AtomicBoolean;
27  import java.util.concurrent.atomic.AtomicInteger;
28  import java.util.concurrent.atomic.AtomicReference;
29  
30  import org.apache.commons.logging.Log;
31  import org.apache.commons.logging.LogFactory;
32  import org.apache.commons.logging.impl.Log4JLogger;
33  import org.apache.hadoop.conf.Configuration;
34  import org.apache.hadoop.hbase.HBaseTestingUtility;
35  import org.apache.hadoop.hbase.HConstants;
36  import org.apache.hadoop.hbase.HTableDescriptor;
37  import org.apache.hadoop.hbase.TableName;
38  import org.apache.hadoop.hbase.Waiter.Predicate;
39  import org.apache.hadoop.hbase.client.Admin;
40  import org.apache.hadoop.hbase.client.Connection;
41  import org.apache.hadoop.hbase.client.ConnectionFactory;
42  import org.apache.hadoop.hbase.client.Consistency;
43  import org.apache.hadoop.hbase.client.Get;
44  import org.apache.hadoop.hbase.client.RpcRetryingCaller;
45  import org.apache.hadoop.hbase.client.Table;
46  import org.apache.hadoop.hbase.replication.regionserver.TestRegionReplicaReplicationEndpoint;
47  import org.apache.hadoop.hbase.testclassification.LargeTests;
48  import org.apache.hadoop.hbase.util.Bytes;
49  import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
50  import org.apache.hadoop.hbase.util.Threads;
51  import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
52  import org.apache.log4j.Level;
53  import org.junit.After;
54  import org.junit.Before;
55  import org.junit.Rule;
56  import org.junit.Test;
57  import org.junit.experimental.categories.Category;
58  import org.junit.rules.TestName;
59  import org.junit.runner.RunWith;
60  import org.junit.runners.Parameterized;
61  import org.junit.runners.Parameterized.Parameters;
62  
63  /**
64   * Tests failover of secondary region replicas.
65   */
66  @RunWith(Parameterized.class)
67  @Category(LargeTests.class)
68  public class TestRegionReplicaFailover {
69  
70    private static final Log LOG = LogFactory.getLog(TestRegionReplicaReplicationEndpoint.class);
71  
72    static {
73      ((Log4JLogger)RpcRetryingCaller.LOG).getLogger().setLevel(Level.ALL);
74    }
75  
76    private static final HBaseTestingUtility HTU = new HBaseTestingUtility();
77  
78    private static final int NB_SERVERS = 3;
79  
80    protected final byte[][] families = new byte[][] {HBaseTestingUtility.fam1,
81        HBaseTestingUtility.fam2, HBaseTestingUtility.fam3};
82    protected final byte[] fam = HBaseTestingUtility.fam1;
83    protected final byte[] qual1 = Bytes.toBytes("qual1");
84    protected final byte[] value1 = Bytes.toBytes("value1");
85    protected final byte[] row = Bytes.toBytes("rowA");
86    protected final byte[] row2 = Bytes.toBytes("rowB");
87  
88    @Rule public TestName name = new TestName();
89  
90    private HTableDescriptor htd;
91  
92    /*
93     * We are testing with dist log split and dist log replay separately
94     */
95    @Parameters
96    public static Collection<Object[]> getParameters() {
97      Object[][] params =
98          new Boolean[][] { {true}, {false} };
99      return Arrays.asList(params);
100   }
101 
102   @Parameterized.Parameter(0)
103   public boolean distributedLogReplay;
104 
105   @Before
106   public void before() throws Exception {
107     Configuration conf = HTU.getConfiguration();
108     conf.setBoolean(HConstants.REPLICATION_ENABLE_KEY, true);
109     conf.setBoolean(ServerRegionReplicaUtil.REGION_REPLICA_REPLICATION_CONF_KEY, true);
110     conf.setBoolean(ServerRegionReplicaUtil.REGION_REPLICA_WAIT_FOR_PRIMARY_FLUSH_CONF_KEY, true);
111     conf.setInt("replication.stats.thread.period.seconds", 5);
112     conf.setBoolean("hbase.tests.use.shortcircuit.reads", false);
113     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, distributedLogReplay);
114 
115     HTU.startMiniCluster(NB_SERVERS);
116     htd = HTU.createTableDescriptor(
117       name.getMethodName().substring(0, name.getMethodName().length()-3));
118     htd.setRegionReplication(3);
119     HTU.getHBaseAdmin().createTable(htd);
120   }
121 
122   @After
123   public void after() throws Exception {
124     HTU.deleteTableIfAny(htd.getTableName());
125     HTU.shutdownMiniCluster();
126   }
127 
128   /**
129    * Tests the case where a newly created table with region replicas and no data, the secondary
130    * region replicas are available to read immediately.
131    */
132   @Test(timeout = 60000)
133   public void testSecondaryRegionWithEmptyRegion() throws IOException {
134     // Create a new table with region replication, don't put any data. Test that the secondary
135     // region replica is available to read.
136     try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
137         Table table = connection.getTable(htd.getTableName())) {
138 
139       Get get = new Get(row);
140       get.setConsistency(Consistency.TIMELINE);
141       get.setReplicaId(1);
142       table.get(get); // this should not block
143     }
144   }
145 
146   /**
147    * Tests the case where if there is some data in the primary region, reopening the region replicas
148    * (enable/disable table, etc) makes the region replicas readable.
149    * @throws IOException
150    */
151   @Test(timeout = 60000)
152   public void testSecondaryRegionWithNonEmptyRegion() throws IOException {
153     // Create a new table with region replication and load some data
154     // than disable and enable the table again and verify the data from secondary
155     try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
156         Table table = connection.getTable(htd.getTableName())) {
157 
158       HTU.loadNumericRows(table, fam, 0, 1000);
159 
160       HTU.getHBaseAdmin().disableTable(htd.getTableName());
161       HTU.getHBaseAdmin().enableTable(htd.getTableName());
162 
163       HTU.verifyNumericRows(table, fam, 0, 1000, 1);
164     }
165   }
166 
167   /**
168    * Tests the case where killing a primary region with unflushed data recovers
169    */
170   @Test (timeout = 120000)
171   public void testPrimaryRegionKill() throws Exception {
172     try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
173         Table table = connection.getTable(htd.getTableName())) {
174 
175       HTU.loadNumericRows(table, fam, 0, 1000);
176 
177       // wal replication is async, we have to wait until the replication catches up, or we timeout
178       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
179       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
180 
181       // we should not have flushed files now, but data in memstores of primary and secondary
182       // kill the primary region replica now, and ensure that when it comes back up, we can still
183       // read from it the same data from primary and secondaries
184       boolean aborted = false;
185       for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
186         for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
187           if (r.getRegionInfo().getReplicaId() == 0) {
188             LOG.info("Aborting region server hosting primary region replica");
189             rs.getRegionServer().abort("for test");
190             aborted = true;
191           }
192         }
193       }
194       assertTrue(aborted);
195 
196       // wal replication is async, we have to wait until the replication catches up, or we timeout
197       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 0, 30000);
198       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
199       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
200     }
201 
202     // restart the region server
203     HTU.getMiniHBaseCluster().startRegionServer();
204   }
205 
206   /** wal replication is async, we have to wait until the replication catches up, or we timeout
207    */
208   private void verifyNumericRowsWithTimeout(final Table table, final byte[] f, final int startRow,
209       final int endRow, final int replicaId, final long timeout) throws Exception {
210     try {
211       HTU.waitFor(timeout, new Predicate<Exception>() {
212         @Override
213         public boolean evaluate() throws Exception {
214           try {
215             HTU.verifyNumericRows(table, f, startRow, endRow, replicaId);
216             return true;
217           } catch (AssertionError ae) {
218             return false;
219           }
220         }
221       });
222     } catch (Throwable t) {
223       // ignore this, but redo the verify do get the actual exception
224       HTU.verifyNumericRows(table, f, startRow, endRow, replicaId);
225     }
226   }
227 
228   /**
229    * Tests the case where killing a secondary region with unflushed data recovers, and the replica
230    * becomes available to read again shortly.
231    */
232   @Test (timeout = 120000)
233   public void testSecondaryRegionKill() throws Exception {
234     try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
235         Table table = connection.getTable(htd.getTableName())) {
236       HTU.loadNumericRows(table, fam, 0, 1000);
237 
238       // wait for some time to ensure that async wal replication does it's magic
239       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
240       verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
241 
242       // we should not have flushed files now, but data in memstores of primary and secondary
243       // kill the secondary region replica now, and ensure that when it comes back up, we can still
244       // read from it the same data
245       boolean aborted = false;
246       for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
247         for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
248           if (r.getRegionInfo().getReplicaId() == 1) {
249             LOG.info("Aborting region server hosting secondary region replica");
250             rs.getRegionServer().abort("for test");
251             aborted = true;
252           }
253         }
254       }
255       assertTrue(aborted);
256 
257       Threads.sleep(5000);
258 
259       HTU.verifyNumericRows(table, fam, 0, 1000, 1);
260       HTU.verifyNumericRows(table, fam, 0, 1000, 2);
261     }
262 
263     // restart the region server
264     HTU.getMiniHBaseCluster().startRegionServer();
265   }
266 
267   /**
268    * Tests the case where there are 3 region replicas and the primary is continuously accepting
269    * new writes while one of the secondaries is killed. Verification is done for both of the
270    * secondary replicas.
271    */
272   @Test (timeout = 120000)
273   public void testSecondaryRegionKillWhilePrimaryIsAcceptingWrites() throws Exception {
274     try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
275         Table table = connection.getTable(htd.getTableName());
276         Admin admin = connection.getAdmin()) {
277       // start a thread to do the loading of primary
278       HTU.loadNumericRows(table, fam, 0, 1000); // start with some base
279       admin.flush(table.getName());
280       HTU.loadNumericRows(table, fam, 1000, 2000);
281 
282       final AtomicReference<Throwable> ex = new AtomicReference<Throwable>(null);
283       final AtomicBoolean done = new AtomicBoolean(false);
284       final AtomicInteger key = new AtomicInteger(2000);
285 
286       Thread loader = new Thread() {
287         @Override
288         public void run() {
289           while (!done.get()) {
290             try {
291               HTU.loadNumericRows(table, fam, key.get(), key.get()+1000);
292               key.addAndGet(1000);
293             } catch (Throwable e) {
294               ex.compareAndSet(null, e);
295             }
296           }
297         }
298       };
299       loader.start();
300 
301       Thread aborter = new Thread() {
302         @Override
303         public void run() {
304           try {
305             boolean aborted = false;
306             for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
307               for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
308                 if (r.getRegionInfo().getReplicaId() == 1) {
309                   LOG.info("Aborting region server hosting secondary region replica");
310                   rs.getRegionServer().abort("for test");
311                   aborted = true;
312                 }
313               }
314             }
315             assertTrue(aborted);
316           } catch (Throwable e) {
317             ex.compareAndSet(null, e);
318           }
319         };
320       };
321 
322       aborter.start();
323       aborter.join();
324       done.set(true);
325       loader.join();
326 
327       assertNull(ex.get());
328 
329       assertTrue(key.get() > 1000); // assert that the test is working as designed
330       LOG.info("Loaded up to key :" + key.get());
331       verifyNumericRowsWithTimeout(table, fam, 0, key.get(), 0, 30000);
332       verifyNumericRowsWithTimeout(table, fam, 0, key.get(), 1, 30000);
333       verifyNumericRowsWithTimeout(table, fam, 0, key.get(), 2, 30000);
334     }
335 
336     // restart the region server
337     HTU.getMiniHBaseCluster().startRegionServer();
338   }
339 
340   /**
341    * Tests the case where we are creating a table with a lot of regions and replicas. Opening region
342    * replicas should not block handlers on RS indefinitely.
343    */
344   @Test (timeout = 120000)
345   public void testLotsOfRegionReplicas() throws IOException {
346     int numRegions = NB_SERVERS * 20;
347     int regionReplication = 10;
348     String tableName = htd.getTableName().getNameAsString() + "2";
349     htd = HTU.createTableDescriptor(tableName);
350     htd.setRegionReplication(regionReplication);
351 
352     // dont care about splits themselves too much
353     byte[] startKey = Bytes.toBytes("aaa");
354     byte[] endKey = Bytes.toBytes("zzz");
355     byte[][] splits = HTU.getRegionSplitStartKeys(startKey, endKey, numRegions);
356     HTU.getHBaseAdmin().createTable(htd, startKey, endKey, numRegions);
357 
358     try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
359         Table table = connection.getTable(htd.getTableName())) {
360 
361       for (int i = 1; i < splits.length; i++) {
362         for (int j = 0; j < regionReplication; j++) {
363           Get get = new Get(splits[i]);
364           get.setConsistency(Consistency.TIMELINE);
365           get.setReplicaId(j);
366           table.get(get); // this should not block. Regions should be coming online
367         }
368       }
369     }
370 
371     HTU.deleteTableIfAny(TableName.valueOf(tableName));
372   }
373 }