View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.master;
20  
21  import static org.apache.hadoop.hbase.SplitLogCounters.tot_mgr_wait_for_zk_delete;
22  import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_final_transition_failed;
23  import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_preempt_task;
24  import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_acquired;
25  import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_done;
26  import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_err;
27  import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_resigned;
28  import static org.junit.Assert.assertEquals;
29  import static org.junit.Assert.assertFalse;
30  import static org.junit.Assert.assertNotNull;
31  import static org.junit.Assert.assertTrue;
32  import static org.junit.Assert.fail;
33  
34  import java.io.IOException;
35  import java.util.ArrayList;
36  import java.util.Arrays;
37  import java.util.HashSet;
38  import java.util.Iterator;
39  import java.util.LinkedList;
40  import java.util.List;
41  import java.util.NavigableSet;
42  import java.util.Set;
43  import java.util.concurrent.ExecutorService;
44  import java.util.concurrent.Executors;
45  import java.util.concurrent.Future;
46  import java.util.concurrent.TimeUnit;
47  import java.util.concurrent.TimeoutException;
48  import java.util.concurrent.atomic.AtomicLong;
49  
50  import org.apache.commons.logging.Log;
51  import org.apache.commons.logging.LogFactory;
52  import org.apache.hadoop.conf.Configuration;
53  import org.apache.hadoop.fs.FSDataOutputStream;
54  import org.apache.hadoop.fs.FileStatus;
55  import org.apache.hadoop.fs.FileSystem;
56  import org.apache.hadoop.fs.Path;
57  import org.apache.hadoop.fs.PathFilter;
58  import org.apache.hadoop.hbase.HBaseConfiguration;
59  import org.apache.hadoop.hbase.HBaseTestingUtility;
60  import org.apache.hadoop.hbase.HColumnDescriptor;
61  import org.apache.hadoop.hbase.HConstants;
62  import org.apache.hadoop.hbase.HRegionInfo;
63  import org.apache.hadoop.hbase.HTableDescriptor;
64  import org.apache.hadoop.hbase.KeyValue;
65  import org.apache.hadoop.hbase.MiniHBaseCluster;
66  import org.apache.hadoop.hbase.NamespaceDescriptor;
67  import org.apache.hadoop.hbase.ServerName;
68  import org.apache.hadoop.hbase.SplitLogCounters;
69  import org.apache.hadoop.hbase.TableName;
70  import org.apache.hadoop.hbase.Waiter;
71  import org.apache.hadoop.hbase.client.ClusterConnection;
72  import org.apache.hadoop.hbase.client.ConnectionUtils;
73  import org.apache.hadoop.hbase.client.Delete;
74  import org.apache.hadoop.hbase.client.Get;
75  import org.apache.hadoop.hbase.client.Increment;
76  import org.apache.hadoop.hbase.client.NonceGenerator;
77  import org.apache.hadoop.hbase.client.PerClientRandomNonceGenerator;
78  import org.apache.hadoop.hbase.client.Put;
79  import org.apache.hadoop.hbase.client.RegionLocator;
80  import org.apache.hadoop.hbase.client.Result;
81  import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
82  import org.apache.hadoop.hbase.client.Table;
83  import org.apache.hadoop.hbase.coordination.BaseCoordinatedStateManager;
84  import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination;
85  import org.apache.hadoop.hbase.exceptions.OperationConflictException;
86  import org.apache.hadoop.hbase.exceptions.RegionInRecoveryException;
87  import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
88  import org.apache.hadoop.hbase.master.SplitLogManager.TaskBatch;
89  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
90  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
91  import org.apache.hadoop.hbase.regionserver.HRegion;
92  import org.apache.hadoop.hbase.regionserver.HRegionServer;
93  import org.apache.hadoop.hbase.regionserver.Region;
94  import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
95  import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
96  import org.apache.hadoop.hbase.testclassification.LargeTests;
97  import org.apache.hadoop.hbase.util.Bytes;
98  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
99  import org.apache.hadoop.hbase.util.FSUtils;
100 import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread;
101 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
102 import org.apache.hadoop.hbase.util.Threads;
103 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
104 import org.apache.hadoop.hbase.wal.WAL;
105 import org.apache.hadoop.hbase.wal.WALFactory;
106 import org.apache.hadoop.hbase.wal.WALSplitter;
107 import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
108 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
109 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
110 import org.apache.hadoop.hdfs.MiniDFSCluster;
111 import org.junit.After;
112 import org.junit.AfterClass;
113 import org.junit.Assert;
114 import org.junit.Before;
115 import org.junit.BeforeClass;
116 import org.junit.Ignore;
117 import org.junit.Test;
118 import org.junit.experimental.categories.Category;
119 
120 @Category({LargeTests.class})
121 @SuppressWarnings("deprecation")
122 public class TestDistributedLogSplitting {
123   private static final Log LOG = LogFactory.getLog(TestSplitLogManager.class);
124   static {
125     // Uncomment the following line if more verbosity is needed for
126     // debugging (see HBASE-12285 for details).
127     //Logger.getLogger("org.apache.hadoop.hbase").setLevel(Level.DEBUG);
128 
129     // test ThreeRSAbort fails under hadoop2 (2.0.2-alpha) if shortcircuit-read (scr) is on. this
130     // turns it off for this test.  TODO: Figure out why scr breaks recovery.
131     System.setProperty("hbase.tests.use.shortcircuit.reads", "false");
132 
133   }
134 
135   // Start a cluster with 2 masters and 6 regionservers
136   static final int NUM_MASTERS = 2;
137   static final int NUM_RS = 6;
138 
139   MiniHBaseCluster cluster;
140   HMaster master;
141   Configuration conf;
142   static Configuration originalConf;
143   static HBaseTestingUtility TEST_UTIL;
144   static MiniDFSCluster dfsCluster;
145   static MiniZooKeeperCluster zkCluster;
146 
147   @BeforeClass
148   public static void setup() throws Exception {
149     TEST_UTIL = new HBaseTestingUtility(HBaseConfiguration.create());
150     dfsCluster = TEST_UTIL.startMiniDFSCluster(1);
151     zkCluster = TEST_UTIL.startMiniZKCluster();
152     originalConf = TEST_UTIL.getConfiguration();
153   }
154 
155   @AfterClass
156   public static void tearDown() throws IOException {
157     TEST_UTIL.shutdownMiniZKCluster();
158     TEST_UTIL.shutdownMiniDFSCluster();
159     TEST_UTIL.shutdownMiniHBaseCluster();
160   }
161 
162   private void startCluster(int num_rs) throws Exception {
163     SplitLogCounters.resetCounters();
164     LOG.info("Starting cluster");
165     conf.getLong("hbase.splitlog.max.resubmit", 0);
166     // Make the failure test faster
167     conf.setInt("zookeeper.recovery.retry", 0);
168     conf.setInt(HConstants.REGIONSERVER_INFO_PORT, -1);
169     conf.setFloat(HConstants.LOAD_BALANCER_SLOP_KEY, (float) 100.0); // no load balancing
170     conf.setInt("hbase.regionserver.wal.max.splitters", 3);
171     conf.setInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT, 10);
172     TEST_UTIL.shutdownMiniHBaseCluster();
173     TEST_UTIL = new HBaseTestingUtility(conf);
174     TEST_UTIL.setDFSCluster(dfsCluster);
175     TEST_UTIL.setZkCluster(zkCluster);
176     TEST_UTIL.startMiniHBaseCluster(NUM_MASTERS, num_rs);
177     cluster = TEST_UTIL.getHBaseCluster();
178     LOG.info("Waiting for active/ready master");
179     cluster.waitForActiveAndReadyMaster();
180     master = cluster.getMaster();
181     while (cluster.getLiveRegionServerThreads().size() < num_rs) {
182       Threads.sleep(10);
183     }
184   }
185 
186   @Before
187   public void before() throws Exception {
188     // refresh configuration
189     conf = HBaseConfiguration.create(originalConf);
190   }
191 
192   @After
193   public void after() throws Exception {
194     try {
195       if (TEST_UTIL.getHBaseCluster() != null) {
196         for (MasterThread mt : TEST_UTIL.getHBaseCluster().getLiveMasterThreads()) {
197           mt.getMaster().abort("closing...", null);
198         }
199       }
200       TEST_UTIL.shutdownMiniHBaseCluster();
201     } finally {
202       TEST_UTIL.getTestFileSystem().delete(FSUtils.getRootDir(TEST_UTIL.getConfiguration()), true);
203       ZKUtil.deleteNodeRecursively(TEST_UTIL.getZooKeeperWatcher(), "/hbase");
204     }
205   }
206 
207   @Ignore("DLR is broken by HBASE-12751") @Test (timeout=300000)
208   public void testRecoveredEdits() throws Exception {
209     LOG.info("testRecoveredEdits");
210     conf.setLong("hbase.regionserver.hlog.blocksize", 30 * 1024); // create more than one wal
211     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false);
212     startCluster(NUM_RS);
213 
214     final int NUM_LOG_LINES = 1000;
215     final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
216     // turn off load balancing to prevent regions from moving around otherwise
217     // they will consume recovered.edits
218     master.balanceSwitch(false);
219     FileSystem fs = master.getMasterFileSystem().getFileSystem();
220 
221     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
222 
223     Path rootdir = FSUtils.getRootDir(conf);
224 
225     Table t = installTable(new ZooKeeperWatcher(conf, "table-creation", null),
226         "table", "family", 40);
227     try {
228       TableName table = t.getName();
229       List<HRegionInfo> regions = null;
230       HRegionServer hrs = null;
231       for (int i = 0; i < NUM_RS; i++) {
232         boolean foundRs = false;
233         hrs = rsts.get(i).getRegionServer();
234         regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
235         for (HRegionInfo region : regions) {
236           if (region.getTable().getNameAsString().equalsIgnoreCase("table")) {
237             foundRs = true;
238             break;
239           }
240         }
241         if (foundRs) break;
242       }
243       final Path logDir = new Path(rootdir, DefaultWALProvider.getWALDirectoryName(hrs
244           .getServerName().toString()));
245 
246       LOG.info("#regions = " + regions.size());
247       Iterator<HRegionInfo> it = regions.iterator();
248       while (it.hasNext()) {
249         HRegionInfo region = it.next();
250         if (region.getTable().getNamespaceAsString()
251             .equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) {
252           it.remove();
253         }
254       }
255 
256       makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
257 
258       slm.splitLogDistributed(logDir);
259 
260       int count = 0;
261       for (HRegionInfo hri : regions) {
262 
263         Path tdir = FSUtils.getTableDir(rootdir, table);
264         Path editsdir =
265             WALSplitter.getRegionDirRecoveredEditsDir(
266                 HRegion.getRegionDir(tdir, hri.getEncodedName()));
267         LOG.debug("checking edits dir " + editsdir);
268         FileStatus[] files = fs.listStatus(editsdir, new PathFilter() {
269           @Override
270           public boolean accept(Path p) {
271             if (WALSplitter.isSequenceIdFile(p)) {
272               return false;
273             }
274             return true;
275           }
276         });
277         assertTrue(
278             "edits dir should have more than a single file in it. instead has " + files.length,
279             files.length > 1);
280         for (int i = 0; i < files.length; i++) {
281           int c = countWAL(files[i].getPath(), fs, conf);
282           count += c;
283         }
284         LOG.info(count + " edits in " + files.length + " recovered edits files.");
285       }
286 
287       // check that the log file is moved
288       assertFalse(fs.exists(logDir));
289 
290       assertEquals(NUM_LOG_LINES, count);
291     } finally {
292       if (t != null) t.close();
293     }
294   }
295 
296   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
297   public void testLogReplayWithNonMetaRSDown() throws Exception {
298     LOG.info("testLogReplayWithNonMetaRSDown");
299     conf.setLong("hbase.regionserver.hlog.blocksize", 30 * 1024); // create more than one wal
300     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
301     startCluster(NUM_RS);
302     final int NUM_REGIONS_TO_CREATE = 40;
303     final int NUM_LOG_LINES = 1000;
304     // turn off load balancing to prevent regions from moving around otherwise
305     // they will consume recovered.edits
306     master.balanceSwitch(false);
307 
308     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
309     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
310     try {
311       HRegionServer hrs = findRSToKill(false, "table");
312       List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
313       makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
314 
315       // wait for abort completes
316       this.abortRSAndVerifyRecovery(hrs, ht, zkw, NUM_REGIONS_TO_CREATE, NUM_LOG_LINES);
317     } finally {
318       if (ht != null) ht.close();
319       if (zkw != null) zkw.close();
320     }
321   }
322 
323   private static class NonceGeneratorWithDups extends PerClientRandomNonceGenerator {
324     private boolean isDups = false;
325     private LinkedList<Long> nonces = new LinkedList<Long>();
326 
327     public void startDups() {
328       isDups = true;
329     }
330 
331     @Override
332     public long newNonce() {
333       long nonce = isDups ? nonces.removeFirst() : super.newNonce();
334       if (!isDups) {
335         nonces.add(nonce);
336       }
337       return nonce;
338     }
339   }
340 
341   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
342   public void testNonceRecovery() throws Exception {
343     LOG.info("testNonceRecovery");
344     final String TABLE_NAME = "table";
345     final String FAMILY_NAME = "family";
346     final int NUM_REGIONS_TO_CREATE = 40;
347 
348     conf.setLong("hbase.regionserver.hlog.blocksize", 100*1024);
349     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
350     startCluster(NUM_RS);
351     master.balanceSwitch(false);
352 
353     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
354     Table ht = installTable(zkw, TABLE_NAME, FAMILY_NAME, NUM_REGIONS_TO_CREATE);
355     NonceGeneratorWithDups ng = new NonceGeneratorWithDups();
356     NonceGenerator oldNg =
357         ConnectionUtils.injectNonceGeneratorForTesting(
358             (ClusterConnection)TEST_UTIL.getConnection(), ng);
359 
360     try {
361       List<Increment> reqs = new ArrayList<Increment>();
362       for (RegionServerThread rst : cluster.getLiveRegionServerThreads()) {
363         HRegionServer hrs = rst.getRegionServer();
364         List<HRegionInfo> hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
365         for (HRegionInfo hri : hris) {
366           if (TABLE_NAME.equalsIgnoreCase(hri.getTable().getNameAsString())) {
367             byte[] key = hri.getStartKey();
368             if (key == null || key.length == 0) {
369               key = Bytes.copy(hri.getEndKey());
370               --(key[key.length - 1]);
371             }
372             Increment incr = new Increment(key);
373             incr.addColumn(Bytes.toBytes(FAMILY_NAME), Bytes.toBytes("q"), 1);
374             ht.increment(incr);
375             reqs.add(incr);
376           }
377         }
378       }
379 
380       HRegionServer hrs = findRSToKill(false, "table");
381       abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
382       ng.startDups();
383       for (Increment incr : reqs) {
384         try {
385           ht.increment(incr);
386           fail("should have thrown");
387         } catch (OperationConflictException ope) {
388           LOG.debug("Caught as expected: " + ope.getMessage());
389         }
390       }
391     } finally {
392       ConnectionUtils.injectNonceGeneratorForTesting((ClusterConnection)
393           TEST_UTIL.getConnection(), oldNg);
394       if (ht != null) ht.close();
395       if (zkw != null) zkw.close();
396     }
397   }
398 
399   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
400   public void testLogReplayWithMetaRSDown() throws Exception {
401     LOG.info("testRecoveredEditsReplayWithMetaRSDown");
402     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
403     startCluster(NUM_RS);
404     final int NUM_REGIONS_TO_CREATE = 40;
405     final int NUM_LOG_LINES = 1000;
406     // turn off load balancing to prevent regions from moving around otherwise
407     // they will consume recovered.edits
408     master.balanceSwitch(false);
409 
410     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
411     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
412     try {
413       HRegionServer hrs = findRSToKill(true, "table");
414       List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
415       makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
416 
417       this.abortRSAndVerifyRecovery(hrs, ht, zkw, NUM_REGIONS_TO_CREATE, NUM_LOG_LINES);
418     } finally {
419       if (ht != null) ht.close();
420       if (zkw != null) zkw.close();
421     }
422   }
423 
424   private void abortRSAndVerifyRecovery(HRegionServer hrs, Table ht, final ZooKeeperWatcher zkw,
425       final int numRegions, final int numofLines) throws Exception {
426 
427     abortRSAndWaitForRecovery(hrs, zkw, numRegions);
428     assertEquals(numofLines, TEST_UTIL.countRows(ht));
429   }
430 
431   private void abortRSAndWaitForRecovery(HRegionServer hrs, final ZooKeeperWatcher zkw,
432       final int numRegions) throws Exception {
433     final MiniHBaseCluster tmpCluster = this.cluster;
434 
435     // abort RS
436     LOG.info("Aborting region server: " + hrs.getServerName());
437     hrs.abort("testing");
438 
439     // wait for abort completes
440     TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
441       @Override
442       public boolean evaluate() throws Exception {
443         return (tmpCluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
444       }
445     });
446 
447     // wait for regions come online
448     TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
449       @Override
450       public boolean evaluate() throws Exception {
451         return (HBaseTestingUtility.getAllOnlineRegions(tmpCluster).size()
452             >= (numRegions + 1));
453       }
454     });
455 
456     // wait for all regions are fully recovered
457     TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
458       @Override
459       public boolean evaluate() throws Exception {
460         List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
461             zkw.recoveringRegionsZNode, false);
462         return (recoveringRegions != null && recoveringRegions.size() == 0);
463       }
464     });
465   }
466 
467   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
468   public void testMasterStartsUpWithLogSplittingWork() throws Exception {
469     LOG.info("testMasterStartsUpWithLogSplittingWork");
470     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false);
471     conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, NUM_RS - 1);
472     startCluster(NUM_RS);
473 
474     final int NUM_REGIONS_TO_CREATE = 40;
475     final int NUM_LOG_LINES = 1000;
476     // turn off load balancing to prevent regions from moving around otherwise
477     // they will consume recovered.edits
478     master.balanceSwitch(false);
479 
480     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
481     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
482     try {
483       HRegionServer hrs = findRSToKill(false, "table");
484       List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
485       makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
486 
487       // abort master
488       abortMaster(cluster);
489 
490       // abort RS
491       LOG.info("Aborting region server: " + hrs.getServerName());
492       hrs.abort("testing");
493 
494       // wait for abort completes
495       TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
496         @Override
497         public boolean evaluate() throws Exception {
498           return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
499         }
500       });
501 
502       Thread.sleep(2000);
503       LOG.info("Current Open Regions:"
504           + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
505 
506       // wait for abort completes
507       TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
508         @Override
509         public boolean evaluate() throws Exception {
510           return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
511               >= (NUM_REGIONS_TO_CREATE + 1));
512         }
513       });
514 
515       LOG.info("Current Open Regions After Master Node Starts Up:"
516           + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
517 
518       assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
519     } finally {
520       if (ht != null) ht.close();
521       if (zkw != null) zkw.close();
522     }
523   }
524 
525   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
526   public void testMasterStartsUpWithLogReplayWork() throws Exception {
527     LOG.info("testMasterStartsUpWithLogReplayWork");
528     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
529     conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, NUM_RS - 1);
530     startCluster(NUM_RS);
531 
532     final int NUM_REGIONS_TO_CREATE = 40;
533     final int NUM_LOG_LINES = 1000;
534     // turn off load balancing to prevent regions from moving around otherwise
535     // they will consume recovered.edits
536     master.balanceSwitch(false);
537 
538     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
539     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
540     try {
541       HRegionServer hrs = findRSToKill(false, "table");
542       List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
543       makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
544 
545       // abort master
546       abortMaster(cluster);
547 
548       // abort RS
549       LOG.info("Aborting region server: " + hrs.getServerName());
550       hrs.abort("testing");
551 
552       // wait for the RS dies
553       TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
554         @Override
555         public boolean evaluate() throws Exception {
556           return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
557         }
558       });
559 
560       Thread.sleep(2000);
561       LOG.info("Current Open Regions:" + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
562 
563       // wait for all regions are fully recovered
564       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
565         @Override
566         public boolean evaluate() throws Exception {
567           List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
568               zkw.recoveringRegionsZNode, false);
569           boolean done = recoveringRegions != null && recoveringRegions.size() == 0;
570           if (!done) {
571             LOG.info("Recovering regions: " + recoveringRegions);
572           }
573           return done;
574         }
575       });
576 
577       LOG.info("Current Open Regions After Master Node Starts Up:"
578           + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
579 
580       assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
581     } finally {
582       if (ht != null) ht.close();
583       if (zkw != null) zkw.close();
584     }
585   }
586 
587 
588   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
589   public void testLogReplayTwoSequentialRSDown() throws Exception {
590     LOG.info("testRecoveredEditsReplayTwoSequentialRSDown");
591     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
592     startCluster(NUM_RS);
593     final int NUM_REGIONS_TO_CREATE = 40;
594     final int NUM_LOG_LINES = 1000;
595     // turn off load balancing to prevent regions from moving around otherwise
596     // they will consume recovered.edits
597     master.balanceSwitch(false);
598 
599     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
600     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
601     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
602     try {
603       List<HRegionInfo> regions = null;
604       HRegionServer hrs1 = findRSToKill(false, "table");
605       regions = ProtobufUtil.getOnlineRegions(hrs1.getRSRpcServices());
606 
607       makeWAL(hrs1, regions, "table", "family", NUM_LOG_LINES, 100);
608 
609       // abort RS1
610       LOG.info("Aborting region server: " + hrs1.getServerName());
611       hrs1.abort("testing");
612 
613       // wait for abort completes
614       TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
615         @Override
616         public boolean evaluate() throws Exception {
617           return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
618         }
619       });
620 
621       // wait for regions come online
622       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
623         @Override
624         public boolean evaluate() throws Exception {
625           return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
626               >= (NUM_REGIONS_TO_CREATE + 1));
627         }
628       });
629 
630       // sleep a little bit in order to interrupt recovering in the middle
631       Thread.sleep(300);
632       // abort second region server
633       rsts = cluster.getLiveRegionServerThreads();
634       HRegionServer hrs2 = rsts.get(0).getRegionServer();
635       LOG.info("Aborting one more region server: " + hrs2.getServerName());
636       hrs2.abort("testing");
637 
638       // wait for abort completes
639       TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
640         @Override
641         public boolean evaluate() throws Exception {
642           return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 2));
643         }
644       });
645 
646       // wait for regions come online
647       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
648         @Override
649         public boolean evaluate() throws Exception {
650           return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
651               >= (NUM_REGIONS_TO_CREATE + 1));
652         }
653       });
654 
655       // wait for all regions are fully recovered
656       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
657         @Override
658         public boolean evaluate() throws Exception {
659           List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
660               zkw.recoveringRegionsZNode, false);
661           return (recoveringRegions != null && recoveringRegions.size() == 0);
662         }
663       });
664 
665       assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
666     } finally {
667       if (ht != null) ht.close();
668       if (zkw != null) zkw.close();
669     }
670   }
671 
672   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
673   public void testMarkRegionsRecoveringInZK() throws Exception {
674     LOG.info("testMarkRegionsRecoveringInZK");
675     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
676     startCluster(NUM_RS);
677     master.balanceSwitch(false);
678     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
679     final ZooKeeperWatcher zkw = master.getZooKeeper();
680     Table ht = installTable(zkw, "table", "family", 40);
681     try {
682       final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
683 
684       Set<HRegionInfo> regionSet = new HashSet<HRegionInfo>();
685       HRegionInfo region = null;
686       HRegionServer hrs = null;
687       ServerName firstFailedServer = null;
688       ServerName secondFailedServer = null;
689       for (int i = 0; i < NUM_RS; i++) {
690         hrs = rsts.get(i).getRegionServer();
691         List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
692         if (regions.isEmpty()) continue;
693         region = regions.get(0);
694         regionSet.add(region);
695         firstFailedServer = hrs.getServerName();
696         secondFailedServer = rsts.get((i + 1) % NUM_RS).getRegionServer().getServerName();
697         break;
698       }
699 
700       slm.markRegionsRecovering(firstFailedServer, regionSet);
701       slm.markRegionsRecovering(secondFailedServer, regionSet);
702 
703       assertNotNull(zkw.recoveringRegionsZNode);
704       List<String> recoveringRegions = ZKUtil.listChildrenNoWatch(zkw,
705           ZKUtil.joinZNode(zkw.recoveringRegionsZNode, region.getEncodedName()));
706 
707       assertEquals(recoveringRegions.size(), 2);
708 
709       // wait for splitLogWorker to mark them up because there is no WAL files recorded in ZK
710       final HRegionServer tmphrs = hrs;
711       TEST_UTIL.waitFor(60000, 1000, new Waiter.Predicate<Exception>() {
712         @Override
713         public boolean evaluate() throws Exception {
714           return (tmphrs.getRecoveringRegions().size() == 0);
715         }
716       });
717     } finally {
718       if (ht != null) ht.close();
719       if (zkw != null) zkw.close();
720     }
721   }
722 
723   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
724   public void testReplayCmd() throws Exception {
725     LOG.info("testReplayCmd");
726     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
727     startCluster(NUM_RS);
728     final int NUM_REGIONS_TO_CREATE = 40;
729     // turn off load balancing to prevent regions from moving around otherwise
730     // they will consume recovered.edits
731     master.balanceSwitch(false);
732 
733     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
734     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
735     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
736     try {
737       List<HRegionInfo> regions = null;
738       HRegionServer hrs = null;
739       for (int i = 0; i < NUM_RS; i++) {
740         boolean isCarryingMeta = false;
741         hrs = rsts.get(i).getRegionServer();
742         regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
743         for (HRegionInfo region : regions) {
744           if (region.isMetaRegion()) {
745             isCarryingMeta = true;
746             break;
747           }
748         }
749         if (isCarryingMeta) {
750           continue;
751         }
752         if (regions.size() > 0) break;
753       }
754 
755       this.prepareData(ht, Bytes.toBytes("family"), Bytes.toBytes("c1"));
756       String originalCheckSum = TEST_UTIL.checksumRows(ht);
757 
758       // abort RA and trigger replay
759       abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
760 
761       assertEquals("Data should remain after reopening of regions", originalCheckSum,
762           TEST_UTIL.checksumRows(ht));
763     } finally {
764       if (ht != null) ht.close();
765       if (zkw != null) zkw.close();
766     }
767   }
768 
769   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
770   public void testLogReplayForDisablingTable() throws Exception {
771     LOG.info("testLogReplayForDisablingTable");
772     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
773     startCluster(NUM_RS);
774     final int NUM_REGIONS_TO_CREATE = 40;
775     final int NUM_LOG_LINES = 1000;
776 
777     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
778     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
779     Table disablingHT = installTable(zkw, "disableTable", "family", NUM_REGIONS_TO_CREATE);
780     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE, NUM_REGIONS_TO_CREATE);
781     try {
782       // turn off load balancing to prevent regions from moving around otherwise
783       // they will consume recovered.edits
784       master.balanceSwitch(false);
785 
786       List<HRegionInfo> regions = null;
787       HRegionServer hrs = null;
788       boolean hasRegionsForBothTables = false;
789       String tableName = null;
790       for (int i = 0; i < NUM_RS; i++) {
791         tableName = null;
792         hasRegionsForBothTables = false;
793         boolean isCarryingSystem = false;
794         hrs = rsts.get(i).getRegionServer();
795         regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
796         for (HRegionInfo region : regions) {
797           if (region.getTable().isSystemTable()) {
798             isCarryingSystem = true;
799             break;
800           }
801           if (tableName != null &&
802               !tableName.equalsIgnoreCase(region.getTable().getNameAsString())) {
803             // make sure that we find a RS has online regions for both "table" and "disableTable"
804             hasRegionsForBothTables = true;
805             break;
806           } else if (tableName == null) {
807             tableName = region.getTable().getNameAsString();
808           }
809         }
810         if (isCarryingSystem) {
811           continue;
812         }
813         if (hasRegionsForBothTables) {
814           break;
815         }
816       }
817 
818       // make sure we found a good RS
819       Assert.assertTrue(hasRegionsForBothTables);
820 
821       LOG.info("#regions = " + regions.size());
822       Iterator<HRegionInfo> it = regions.iterator();
823       while (it.hasNext()) {
824         HRegionInfo region = it.next();
825         if (region.isMetaTable()) {
826           it.remove();
827         }
828       }
829       makeWAL(hrs, regions, "disableTable", "family", NUM_LOG_LINES, 100, false);
830       makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
831 
832       LOG.info("Disabling table\n");
833       TEST_UTIL.getHBaseAdmin().disableTable(TableName.valueOf("disableTable"));
834       TEST_UTIL.waitTableDisabled(TableName.valueOf("disableTable").getName());
835 
836       // abort RS
837       LOG.info("Aborting region server: " + hrs.getServerName());
838       hrs.abort("testing");
839 
840       // wait for abort completes
841       TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
842         @Override
843         public boolean evaluate() throws Exception {
844           return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
845         }
846       });
847 
848       // wait for regions come online
849       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
850         @Override
851         public boolean evaluate() throws Exception {
852           return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
853               >= (NUM_REGIONS_TO_CREATE + 1));
854         }
855       });
856 
857       // wait for all regions are fully recovered
858       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
859         @Override
860         public boolean evaluate() throws Exception {
861           List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
862               zkw.recoveringRegionsZNode, false);
863           ServerManager serverManager = master.getServerManager();
864           return (!serverManager.areDeadServersInProgress() &&
865               recoveringRegions != null && recoveringRegions.size() == 0);
866         }
867       });
868 
869       int count = 0;
870       FileSystem fs = master.getMasterFileSystem().getFileSystem();
871       Path rootdir = FSUtils.getRootDir(conf);
872       Path tdir = FSUtils.getTableDir(rootdir, TableName.valueOf("disableTable"));
873       for (HRegionInfo hri : regions) {
874         Path editsdir =
875             WALSplitter.getRegionDirRecoveredEditsDir(
876                 HRegion.getRegionDir(tdir, hri.getEncodedName()));
877         LOG.debug("checking edits dir " + editsdir);
878         if(!fs.exists(editsdir)) continue;
879         FileStatus[] files = fs.listStatus(editsdir, new PathFilter() {
880           @Override
881           public boolean accept(Path p) {
882             if (WALSplitter.isSequenceIdFile(p)) {
883               return false;
884             }
885             return true;
886           }
887         });
888         if(files != null) {
889           for(FileStatus file : files) {
890             int c = countWAL(file.getPath(), fs, conf);
891             count += c;
892             LOG.info(c + " edits in " + file.getPath());
893           }
894         }
895       }
896 
897       LOG.info("Verify edits in recovered.edits files");
898       assertEquals(NUM_LOG_LINES, count);
899       LOG.info("Verify replayed edits");
900       assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
901 
902       // clean up
903       for (HRegionInfo hri : regions) {
904         Path editsdir =
905             WALSplitter.getRegionDirRecoveredEditsDir(
906                 HRegion.getRegionDir(tdir, hri.getEncodedName()));
907         fs.delete(editsdir, true);
908       }
909       disablingHT.close();
910     } finally {
911       if (ht != null) ht.close();
912       if (zkw != null) zkw.close();
913     }
914   }
915 
916   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
917   public void testDisallowWritesInRecovering() throws Exception {
918     LOG.info("testDisallowWritesInRecovering");
919     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
920     conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 3);
921     conf.setBoolean(HConstants.DISALLOW_WRITES_IN_RECOVERING, true);
922     startCluster(NUM_RS);
923     final int NUM_REGIONS_TO_CREATE = 40;
924     // turn off load balancing to prevent regions from moving around otherwise
925     // they will consume recovered.edits
926     master.balanceSwitch(false);
927 
928     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
929     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
930     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
931     try {
932       final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
933 
934       Set<HRegionInfo> regionSet = new HashSet<HRegionInfo>();
935       HRegionInfo region = null;
936       HRegionServer hrs = null;
937       HRegionServer dstRS = null;
938       for (int i = 0; i < NUM_RS; i++) {
939         hrs = rsts.get(i).getRegionServer();
940         List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
941         if (regions.isEmpty()) continue;
942         region = regions.get(0);
943         regionSet.add(region);
944         dstRS = rsts.get((i+1) % NUM_RS).getRegionServer();
945         break;
946       }
947 
948       assertNotNull(region);
949       assertNotNull(dstRS);
950 
951       slm.markRegionsRecovering(hrs.getServerName(), regionSet);
952       // move region in order for the region opened in recovering state
953       final HRegionInfo hri = region;
954       final HRegionServer tmpRS = dstRS;
955       TEST_UTIL.getHBaseAdmin().move(region.getEncodedNameAsBytes(),
956           Bytes.toBytes(dstRS.getServerName().getServerName()));
957       // wait for region move completes
958       final RegionStates regionStates =
959           TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager().getRegionStates();
960       TEST_UTIL.waitFor(45000, 200, new Waiter.Predicate<Exception>() {
961         @Override
962         public boolean evaluate() throws Exception {
963           ServerName sn = regionStates.getRegionServerOfRegion(hri);
964           return (sn != null && sn.equals(tmpRS.getServerName()));
965         }
966       });
967 
968       try {
969         byte[] key = region.getStartKey();
970         if (key == null || key.length == 0) {
971           key = new byte[] { 0, 0, 0, 0, 1 };
972         }
973         Put put = new Put(key);
974         put.add(Bytes.toBytes("family"), Bytes.toBytes("c1"), new byte[]{'b'});
975         ht.put(put);
976       } catch (IOException ioe) {
977         Assert.assertTrue(ioe instanceof RetriesExhaustedWithDetailsException);
978         RetriesExhaustedWithDetailsException re = (RetriesExhaustedWithDetailsException) ioe;
979         boolean foundRegionInRecoveryException = false;
980         for (Throwable t : re.getCauses()) {
981           if (t instanceof RegionInRecoveryException) {
982             foundRegionInRecoveryException = true;
983             break;
984           }
985         }
986         Assert.assertTrue(
987             "No RegionInRecoveryException. Following exceptions returned=" + re.getCauses(),
988             foundRegionInRecoveryException);
989       }
990     } finally {
991       if (ht != null) ht.close();
992       if (ht != null) zkw.close();
993     }
994   }
995 
996   /**
997    * The original intention of this test was to force an abort of a region
998    * server and to make sure that the failure path in the region servers is
999    * properly evaluated. But it is difficult to ensure that the region server
1000    * doesn't finish the log splitting before it aborts. Also now, there is
1001    * this code path where the master will preempt the region server when master
1002    * detects that the region server has aborted.
1003    * @throws Exception
1004    */
1005   @Ignore ("Disabled because flakey") @Test (timeout=300000)
1006   public void testWorkerAbort() throws Exception {
1007     LOG.info("testWorkerAbort");
1008     startCluster(3);
1009     final int NUM_LOG_LINES = 10000;
1010     final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
1011     FileSystem fs = master.getMasterFileSystem().getFileSystem();
1012 
1013     final List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1014     HRegionServer hrs = findRSToKill(false, "table");
1015     Path rootdir = FSUtils.getRootDir(conf);
1016     final Path logDir = new Path(rootdir,
1017         DefaultWALProvider.getWALDirectoryName(hrs.getServerName().toString()));
1018 
1019     Table t = installTable(new ZooKeeperWatcher(conf, "table-creation", null),
1020         "table", "family", 40);
1021     try {
1022       makeWAL(hrs, ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices()),
1023           "table", "family", NUM_LOG_LINES, 100);
1024 
1025       new Thread() {
1026         @Override
1027         public void run() {
1028           waitForCounter(tot_wkr_task_acquired, 0, 1, 1000);
1029           for (RegionServerThread rst : rsts) {
1030             rst.getRegionServer().abort("testing");
1031             break;
1032           }
1033         }
1034       }.start();
1035       // slm.splitLogDistributed(logDir);
1036       FileStatus[] logfiles = fs.listStatus(logDir);
1037       TaskBatch batch = new TaskBatch();
1038       slm.enqueueSplitTask(logfiles[0].getPath().toString(), batch);
1039       //waitForCounter but for one of the 2 counters
1040       long curt = System.currentTimeMillis();
1041       long waitTime = 80000;
1042       long endt = curt + waitTime;
1043       while (curt < endt) {
1044         if ((tot_wkr_task_resigned.get() + tot_wkr_task_err.get() +
1045             tot_wkr_final_transition_failed.get() + tot_wkr_task_done.get() +
1046             tot_wkr_preempt_task.get()) == 0) {
1047           Thread.yield();
1048           curt = System.currentTimeMillis();
1049         } else {
1050           assertTrue(1 <= (tot_wkr_task_resigned.get() + tot_wkr_task_err.get() +
1051               tot_wkr_final_transition_failed.get() + tot_wkr_task_done.get() +
1052               tot_wkr_preempt_task.get()));
1053           return;
1054         }
1055       }
1056       fail("none of the following counters went up in " + waitTime +
1057           " milliseconds - " +
1058           "tot_wkr_task_resigned, tot_wkr_task_err, " +
1059           "tot_wkr_final_transition_failed, tot_wkr_task_done, " +
1060           "tot_wkr_preempt_task");
1061     } finally {
1062       if (t != null) t.close();
1063     }
1064   }
1065 
1066   @Test (timeout=300000)
1067   public void testThreeRSAbort() throws Exception {
1068     LOG.info("testThreeRSAbort");
1069     final int NUM_REGIONS_TO_CREATE = 40;
1070     final int NUM_ROWS_PER_REGION = 100;
1071 
1072     startCluster(NUM_RS); // NUM_RS=6.
1073 
1074     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf,
1075         "distributed log splitting test", null);
1076 
1077     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
1078     try {
1079       populateDataInTable(NUM_ROWS_PER_REGION, "family");
1080 
1081 
1082       List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1083       assertEquals(NUM_RS, rsts.size());
1084       rsts.get(0).getRegionServer().abort("testing");
1085       rsts.get(1).getRegionServer().abort("testing");
1086       rsts.get(2).getRegionServer().abort("testing");
1087 
1088       long start = EnvironmentEdgeManager.currentTime();
1089       while (cluster.getLiveRegionServerThreads().size() > (NUM_RS - 3)) {
1090         if (EnvironmentEdgeManager.currentTime() - start > 60000) {
1091           assertTrue(false);
1092         }
1093         Thread.sleep(200);
1094       }
1095 
1096       start = EnvironmentEdgeManager.currentTime();
1097       while (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
1098           < (NUM_REGIONS_TO_CREATE + 1)) {
1099         if (EnvironmentEdgeManager.currentTime() - start > 60000) {
1100           assertTrue("Timedout", false);
1101         }
1102         Thread.sleep(200);
1103       }
1104 
1105       // wait for all regions are fully recovered
1106       TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
1107         @Override
1108         public boolean evaluate() throws Exception {
1109           List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
1110               zkw.recoveringRegionsZNode, false);
1111           return (recoveringRegions != null && recoveringRegions.size() == 0);
1112         }
1113       });
1114 
1115       assertEquals(NUM_REGIONS_TO_CREATE * NUM_ROWS_PER_REGION,
1116           TEST_UTIL.countRows(ht));
1117     } finally {
1118       if (ht != null) ht.close();
1119       if (zkw != null) zkw.close();
1120     }
1121   }
1122 
1123 
1124 
1125   @Test(timeout=30000)
1126   public void testDelayedDeleteOnFailure() throws Exception {
1127     LOG.info("testDelayedDeleteOnFailure");
1128     startCluster(1);
1129     final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
1130     final FileSystem fs = master.getMasterFileSystem().getFileSystem();
1131     final Path logDir = new Path(FSUtils.getRootDir(conf), "x");
1132     fs.mkdirs(logDir);
1133     ExecutorService executor = null;
1134     try {
1135       final Path corruptedLogFile = new Path(logDir, "x");
1136       FSDataOutputStream out;
1137       out = fs.create(corruptedLogFile);
1138       out.write(0);
1139       out.write(Bytes.toBytes("corrupted bytes"));
1140       out.close();
1141       ZKSplitLogManagerCoordination coordination =
1142           (ZKSplitLogManagerCoordination) ((BaseCoordinatedStateManager) master
1143               .getCoordinatedStateManager()).getSplitLogManagerCoordination();
1144       coordination.setIgnoreDeleteForTesting(true);
1145       executor = Executors.newSingleThreadExecutor();
1146       Runnable runnable = new Runnable() {
1147         @Override
1148         public void run() {
1149           try {
1150             // since the logDir is a fake, corrupted one, so the split log worker
1151             // will finish it quickly with error, and this call will fail and throw
1152             // an IOException.
1153             slm.splitLogDistributed(logDir);
1154           } catch (IOException ioe) {
1155             try {
1156               assertTrue(fs.exists(corruptedLogFile));
1157               // this call will block waiting for the task to be removed from the
1158               // tasks map which is not going to happen since ignoreZKDeleteForTesting
1159               // is set to true, until it is interrupted.
1160               slm.splitLogDistributed(logDir);
1161             } catch (IOException e) {
1162               assertTrue(Thread.currentThread().isInterrupted());
1163               return;
1164             }
1165             fail("did not get the expected IOException from the 2nd call");
1166           }
1167           fail("did not get the expected IOException from the 1st call");
1168         }
1169       };
1170       Future<?> result = executor.submit(runnable);
1171       try {
1172         result.get(2000, TimeUnit.MILLISECONDS);
1173       } catch (TimeoutException te) {
1174         // it is ok, expected.
1175       }
1176       waitForCounter(tot_mgr_wait_for_zk_delete, 0, 1, 10000);
1177       executor.shutdownNow();
1178       executor = null;
1179 
1180       // make sure the runnable is finished with no exception thrown.
1181       result.get();
1182     } finally {
1183       if (executor != null) {
1184         // interrupt the thread in case the test fails in the middle.
1185         // it has no effect if the thread is already terminated.
1186         executor.shutdownNow();
1187       }
1188       fs.delete(logDir, true);
1189     }
1190   }
1191 
1192   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
1193   public void testMetaRecoveryInZK() throws Exception {
1194     LOG.info("testMetaRecoveryInZK");
1195     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
1196     startCluster(NUM_RS);
1197 
1198     // turn off load balancing to prevent regions from moving around otherwise
1199     // they will consume recovered.edits
1200     master.balanceSwitch(false);
1201     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1202 
1203     // only testing meta recovery in ZK operation
1204     HRegionServer hrs = findRSToKill(true, null);
1205     List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1206 
1207     LOG.info("#regions = " + regions.size());
1208     Set<HRegionInfo> tmpRegions = new HashSet<HRegionInfo>();
1209     tmpRegions.add(HRegionInfo.FIRST_META_REGIONINFO);
1210     master.getMasterFileSystem().prepareLogReplay(hrs.getServerName(), tmpRegions);
1211     Set<HRegionInfo> userRegionSet = new HashSet<HRegionInfo>();
1212     userRegionSet.addAll(regions);
1213     master.getMasterFileSystem().prepareLogReplay(hrs.getServerName(), userRegionSet);
1214     boolean isMetaRegionInRecovery = false;
1215     List<String> recoveringRegions =
1216         zkw.getRecoverableZooKeeper().getChildren(zkw.recoveringRegionsZNode, false);
1217     for (String curEncodedRegionName : recoveringRegions) {
1218       if (curEncodedRegionName.equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1219         isMetaRegionInRecovery = true;
1220         break;
1221       }
1222     }
1223     assertTrue(isMetaRegionInRecovery);
1224 
1225     master.getMasterFileSystem().splitMetaLog(hrs.getServerName());
1226 
1227     isMetaRegionInRecovery = false;
1228     recoveringRegions =
1229         zkw.getRecoverableZooKeeper().getChildren(zkw.recoveringRegionsZNode, false);
1230     for (String curEncodedRegionName : recoveringRegions) {
1231       if (curEncodedRegionName.equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1232         isMetaRegionInRecovery = true;
1233         break;
1234       }
1235     }
1236     // meta region should be recovered
1237     assertFalse(isMetaRegionInRecovery);
1238     zkw.close();
1239   }
1240 
1241   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
1242   public void testSameVersionUpdatesRecovery() throws Exception {
1243     LOG.info("testSameVersionUpdatesRecovery");
1244     conf.setLong("hbase.regionserver.hlog.blocksize", 15 * 1024);
1245     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
1246     startCluster(NUM_RS);
1247     final int NUM_REGIONS_TO_CREATE = 40;
1248     final int NUM_LOG_LINES = 1000;
1249     // turn off load balancing to prevent regions from moving around otherwise
1250     // they will consume recovered.edits
1251     master.balanceSwitch(false);
1252 
1253     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1254     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1255     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
1256     try {
1257       List<HRegionInfo> regions = null;
1258       HRegionServer hrs = null;
1259       for (int i = 0; i < NUM_RS; i++) {
1260         boolean isCarryingMeta = false;
1261         hrs = rsts.get(i).getRegionServer();
1262         regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1263         for (HRegionInfo region : regions) {
1264           if (region.isMetaRegion()) {
1265             isCarryingMeta = true;
1266             break;
1267           }
1268         }
1269         if (isCarryingMeta) {
1270           continue;
1271         }
1272         break;
1273       }
1274 
1275       LOG.info("#regions = " + regions.size());
1276       Iterator<HRegionInfo> it = regions.iterator();
1277       while (it.hasNext()) {
1278         HRegionInfo region = it.next();
1279         if (region.isMetaTable()
1280             || region.getEncodedName().equals(
1281             HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1282           it.remove();
1283         }
1284       }
1285       if (regions.size() == 0) return;
1286       HRegionInfo curRegionInfo = regions.get(0);
1287       byte[] startRow = curRegionInfo.getStartKey();
1288       if (startRow == null || startRow.length == 0) {
1289         startRow = new byte[] { 0, 0, 0, 0, 1 };
1290       }
1291       byte[] row = Bytes.incrementBytes(startRow, 1);
1292       // use last 5 bytes because HBaseTestingUtility.createMultiRegions use 5 bytes key
1293       row = Arrays.copyOfRange(row, 3, 8);
1294       long value = 0;
1295       TableName tableName = TableName.valueOf("table");
1296       byte[] family = Bytes.toBytes("family");
1297       byte[] qualifier = Bytes.toBytes("c1");
1298       long timeStamp = System.currentTimeMillis();
1299       HTableDescriptor htd = new HTableDescriptor(tableName);
1300       htd.addFamily(new HColumnDescriptor(family));
1301       final WAL wal = hrs.getWAL(curRegionInfo);
1302       for (int i = 0; i < NUM_LOG_LINES; i += 1) {
1303         WALEdit e = new WALEdit();
1304         value++;
1305         e.add(new KeyValue(row, family, qualifier, timeStamp, Bytes.toBytes(value)));
1306         wal.append(htd, curRegionInfo,
1307             new HLogKey(curRegionInfo.getEncodedNameAsBytes(), tableName, System.currentTimeMillis()),
1308             e, true);
1309       }
1310       wal.sync();
1311       wal.shutdown();
1312 
1313       // wait for abort completes
1314       this.abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
1315 
1316       // verify we got the last value
1317       LOG.info("Verification Starts...");
1318       Get g = new Get(row);
1319       Result r = ht.get(g);
1320       long theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1321       assertEquals(value, theStoredVal);
1322 
1323       // after flush
1324       LOG.info("Verification after flush...");
1325       TEST_UTIL.getHBaseAdmin().flush(tableName);
1326       r = ht.get(g);
1327       theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1328       assertEquals(value, theStoredVal);
1329     } finally {
1330       if (ht != null) ht.close();
1331       if (zkw != null) zkw.close();
1332     }
1333   }
1334 
1335   @Ignore("DLR is broken by HBASE-12751") @Test(timeout = 300000)
1336   public void testSameVersionUpdatesRecoveryWithCompaction() throws Exception {
1337     LOG.info("testSameVersionUpdatesRecoveryWithWrites");
1338     conf.setLong("hbase.regionserver.hlog.blocksize", 15 * 1024);
1339     conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
1340     conf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 30 * 1024);
1341     conf.setInt("hbase.hstore.compactionThreshold", 3);
1342     startCluster(NUM_RS);
1343     final int NUM_REGIONS_TO_CREATE = 40;
1344     final int NUM_LOG_LINES = 2000;
1345     // turn off load balancing to prevent regions from moving around otherwise
1346     // they will consume recovered.edits
1347     master.balanceSwitch(false);
1348 
1349     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1350     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1351     Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
1352     try {
1353       List<HRegionInfo> regions = null;
1354       HRegionServer hrs = null;
1355       for (int i = 0; i < NUM_RS; i++) {
1356         boolean isCarryingMeta = false;
1357         hrs = rsts.get(i).getRegionServer();
1358         regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1359         for (HRegionInfo region : regions) {
1360           if (region.isMetaRegion()) {
1361             isCarryingMeta = true;
1362             break;
1363           }
1364         }
1365         if (isCarryingMeta) {
1366           continue;
1367         }
1368         break;
1369       }
1370 
1371       LOG.info("#regions = " + regions.size());
1372       Iterator<HRegionInfo> it = regions.iterator();
1373       while (it.hasNext()) {
1374         HRegionInfo region = it.next();
1375         if (region.isMetaTable()
1376             || region.getEncodedName().equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1377           it.remove();
1378         }
1379       }
1380       if (regions.size() == 0) return;
1381       HRegionInfo curRegionInfo = regions.get(0);
1382       byte[] startRow = curRegionInfo.getStartKey();
1383       if (startRow == null || startRow.length == 0) {
1384         startRow = new byte[] { 0, 0, 0, 0, 1 };
1385       }
1386       byte[] row = Bytes.incrementBytes(startRow, 1);
1387       // use last 5 bytes because HBaseTestingUtility.createMultiRegions use 5 bytes key
1388       row = Arrays.copyOfRange(row, 3, 8);
1389       long value = 0;
1390       final TableName tableName = TableName.valueOf("table");
1391       byte[] family = Bytes.toBytes("family");
1392       byte[] qualifier = Bytes.toBytes("c1");
1393       long timeStamp = System.currentTimeMillis();
1394       HTableDescriptor htd = new HTableDescriptor(tableName);
1395       htd.addFamily(new HColumnDescriptor(family));
1396       final WAL wal = hrs.getWAL(curRegionInfo);
1397       for (int i = 0; i < NUM_LOG_LINES; i += 1) {
1398         WALEdit e = new WALEdit();
1399         value++;
1400         e.add(new KeyValue(row, family, qualifier, timeStamp, Bytes.toBytes(value)));
1401         wal.append(htd, curRegionInfo, new HLogKey(curRegionInfo.getEncodedNameAsBytes(),
1402             tableName, System.currentTimeMillis()), e, true);
1403       }
1404       wal.sync();
1405       wal.shutdown();
1406 
1407       // wait for abort completes
1408       this.abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
1409 
1410       // verify we got the last value
1411       LOG.info("Verification Starts...");
1412       Get g = new Get(row);
1413       Result r = ht.get(g);
1414       long theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1415       assertEquals(value, theStoredVal);
1416 
1417       // after flush & compaction
1418       LOG.info("Verification after flush...");
1419       TEST_UTIL.getHBaseAdmin().flush(tableName);
1420       TEST_UTIL.getHBaseAdmin().compact(tableName);
1421 
1422       // wait for compaction completes
1423       TEST_UTIL.waitFor(30000, 200, new Waiter.Predicate<Exception>() {
1424         @Override
1425         public boolean evaluate() throws Exception {
1426           return (TEST_UTIL.getHBaseAdmin().getCompactionState(tableName) == CompactionState.NONE);
1427         }
1428       });
1429 
1430       r = ht.get(g);
1431       theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1432       assertEquals(value, theStoredVal);
1433     } finally {
1434       if (ht != null) ht.close();
1435       if (zkw != null) zkw.close();
1436     }
1437   }
1438 
1439   @Test(timeout = 300000)
1440   public void testReadWriteSeqIdFiles() throws Exception {
1441     LOG.info("testReadWriteSeqIdFiles");
1442     startCluster(2);
1443     final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1444     Table ht = installTable(zkw, "table", "family", 10);
1445     try {
1446       FileSystem fs = master.getMasterFileSystem().getFileSystem();
1447       Path tableDir = FSUtils.getTableDir(FSUtils.getRootDir(conf), TableName.valueOf("table"));
1448       List<Path> regionDirs = FSUtils.getRegionDirs(fs, tableDir);
1449       long newSeqId = WALSplitter.writeRegionSequenceIdFile(fs, regionDirs.get(0), 1L, 1000L);
1450       WALSplitter.writeRegionSequenceIdFile(fs, regionDirs.get(0) , 1L, 1000L);
1451       assertEquals(newSeqId + 2000,
1452           WALSplitter.writeRegionSequenceIdFile(fs, regionDirs.get(0), 3L, 1000L));
1453 
1454       Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(regionDirs.get(0));
1455       FileStatus[] files = FSUtils.listStatus(fs, editsdir, new PathFilter() {
1456         @Override
1457         public boolean accept(Path p) {
1458           return WALSplitter.isSequenceIdFile(p);
1459         }
1460       });
1461       // only one seqid file should exist
1462       assertEquals(1, files.length);
1463 
1464       // verify all seqId files aren't treated as recovered.edits files
1465       NavigableSet<Path> recoveredEdits =
1466           WALSplitter.getSplitEditFilesSorted(fs, regionDirs.get(0));
1467       assertEquals(0, recoveredEdits.size());
1468     } finally {
1469       if (ht != null) ht.close();
1470       if (zkw != null) zkw.close();
1471     }
1472   }
1473 
1474   Table installTable(ZooKeeperWatcher zkw, String tname, String fname, int nrs) throws Exception {
1475     return installTable(zkw, tname, fname, nrs, 0);
1476   }
1477 
1478   Table installTable(ZooKeeperWatcher zkw, String tname, String fname, int nrs,
1479       int existingRegions) throws Exception {
1480     // Create a table with regions
1481     TableName table = TableName.valueOf(tname);
1482     byte [] family = Bytes.toBytes(fname);
1483     LOG.info("Creating table with " + nrs + " regions");
1484     Table ht = TEST_UTIL.createMultiRegionTable(table, family, nrs);
1485     int numRegions = -1;
1486     try (RegionLocator r = TEST_UTIL.getConnection().getRegionLocator(table)) {
1487       numRegions = r.getStartKeys().length;
1488     }
1489     assertEquals(nrs, numRegions);
1490     LOG.info("Waiting for no more RIT\n");
1491     blockUntilNoRIT(zkw, master);
1492     // disable-enable cycle to get rid of table's dead regions left behind
1493     // by createMultiRegions
1494     LOG.debug("Disabling table\n");
1495     TEST_UTIL.getHBaseAdmin().disableTable(table);
1496     LOG.debug("Waiting for no more RIT\n");
1497     blockUntilNoRIT(zkw, master);
1498     NavigableSet<String> regions = HBaseTestingUtility.getAllOnlineRegions(cluster);
1499     LOG.debug("Verifying only catalog and namespace regions are assigned\n");
1500     if (regions.size() != 2) {
1501       for (String oregion : regions)
1502         LOG.debug("Region still online: " + oregion);
1503     }
1504     assertEquals(2 + existingRegions, regions.size());
1505     LOG.debug("Enabling table\n");
1506     TEST_UTIL.getHBaseAdmin().enableTable(table);
1507     LOG.debug("Waiting for no more RIT\n");
1508     blockUntilNoRIT(zkw, master);
1509     LOG.debug("Verifying there are " + numRegions + " assigned on cluster\n");
1510     regions = HBaseTestingUtility.getAllOnlineRegions(cluster);
1511     assertEquals(numRegions + 2 + existingRegions, regions.size());
1512     return ht;
1513   }
1514 
1515   void populateDataInTable(int nrows, String fname) throws Exception {
1516     byte [] family = Bytes.toBytes(fname);
1517 
1518     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1519     assertEquals(NUM_RS, rsts.size());
1520 
1521     for (RegionServerThread rst : rsts) {
1522       HRegionServer hrs = rst.getRegionServer();
1523       List<HRegionInfo> hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1524       for (HRegionInfo hri : hris) {
1525         if (hri.getTable().isSystemTable()) {
1526           continue;
1527         }
1528         LOG.debug("adding data to rs = " + rst.getName() +
1529             " region = "+ hri.getRegionNameAsString());
1530         Region region = hrs.getOnlineRegion(hri.getRegionName());
1531         assertTrue(region != null);
1532         putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
1533       }
1534     }
1535 
1536     for (MasterThread mt : cluster.getLiveMasterThreads()) {
1537       HRegionServer hrs = mt.getMaster();
1538       List<HRegionInfo> hris;
1539       try {
1540         hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1541       } catch (ServerNotRunningYetException e) {
1542         // It's ok: this master may be a backup. Ignored.
1543         continue;
1544       }
1545       for (HRegionInfo hri : hris) {
1546         if (hri.getTable().isSystemTable()) {
1547           continue;
1548         }
1549         LOG.debug("adding data to rs = " + mt.getName() +
1550             " region = "+ hri.getRegionNameAsString());
1551         Region region = hrs.getOnlineRegion(hri.getRegionName());
1552         assertTrue(region != null);
1553         putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
1554       }
1555     }
1556   }
1557 
1558   public void makeWAL(HRegionServer hrs, List<HRegionInfo> regions, String tname, String fname,
1559       int num_edits, int edit_size) throws IOException {
1560     makeWAL(hrs, regions, tname, fname, num_edits, edit_size, true);
1561   }
1562 
1563   public void makeWAL(HRegionServer hrs, List<HRegionInfo> regions, String tname, String fname,
1564       int num_edits, int edit_size, boolean cleanShutdown) throws IOException {
1565     TableName fullTName = TableName.valueOf(tname);
1566     // remove root and meta region
1567     regions.remove(HRegionInfo.FIRST_META_REGIONINFO);
1568     // using one sequenceId for edits across all regions is ok.
1569 
1570     for(Iterator<HRegionInfo> iter = regions.iterator(); iter.hasNext(); ) {
1571       HRegionInfo regionInfo = iter.next();
1572       if(regionInfo.getTable().isSystemTable()) {
1573         iter.remove();
1574       }
1575     }
1576     HTableDescriptor htd = new HTableDescriptor(fullTName);
1577     byte[] family = Bytes.toBytes(fname);
1578     htd.addFamily(new HColumnDescriptor(family));
1579     byte[] value = new byte[edit_size];
1580 
1581     List<HRegionInfo> hris = new ArrayList<HRegionInfo>();
1582     for (HRegionInfo region : regions) {
1583       if (!region.getTable().getNameAsString().equalsIgnoreCase(tname)) {
1584         continue;
1585       }
1586       hris.add(region);
1587     }
1588     LOG.info("Creating wal edits across " + hris.size() + " regions.");
1589     for (int i = 0; i < edit_size; i++) {
1590       value[i] = (byte) ('a' + (i % 26));
1591     }
1592     int n = hris.size();
1593     int[] counts = new int[n];
1594     // sync every ~30k to line up with desired wal rolls
1595     final int syncEvery = 30 * 1024 / edit_size;
1596     if (n > 0) {
1597       for (int i = 0; i < num_edits; i += 1) {
1598         WALEdit e = new WALEdit();
1599         HRegionInfo curRegionInfo = hris.get(i % n);
1600         final WAL log = hrs.getWAL(curRegionInfo);
1601         byte[] startRow = curRegionInfo.getStartKey();
1602         if (startRow == null || startRow.length == 0) {
1603           startRow = new byte[] { 0, 0, 0, 0, 1 };
1604         }
1605         byte[] row = Bytes.incrementBytes(startRow, counts[i % n]);
1606         row = Arrays.copyOfRange(row, 3, 8); // use last 5 bytes because
1607         // HBaseTestingUtility.createMultiRegions use 5 bytes
1608         // key
1609         byte[] qualifier = Bytes.toBytes("c" + Integer.toString(i));
1610         e.add(new KeyValue(row, family, qualifier, System.currentTimeMillis(), value));
1611         log.append(htd, curRegionInfo,
1612             new HLogKey(curRegionInfo.getEncodedNameAsBytes(), fullTName,
1613                 System.currentTimeMillis()), e, true);
1614         if (0 == i % syncEvery) {
1615           log.sync();
1616         }
1617         counts[i % n] += 1;
1618       }
1619     }
1620     // done as two passes because the regions might share logs. shutdown is idempotent, but sync
1621     // will cause errors if done after.
1622     for (HRegionInfo info : hris) {
1623       final WAL log = hrs.getWAL(info);
1624       log.sync();
1625     }
1626     if (cleanShutdown) {
1627       for (HRegionInfo info : hris) {
1628         final WAL log = hrs.getWAL(info);
1629         log.shutdown();
1630       }
1631     }
1632     for (int i = 0; i < n; i++) {
1633       LOG.info("region " + hris.get(i).getRegionNameAsString() + " has " + counts[i] + " edits");
1634     }
1635     return;
1636   }
1637 
1638   private int countWAL(Path log, FileSystem fs, Configuration conf)
1639       throws IOException {
1640     int count = 0;
1641     WAL.Reader in = WALFactory.createReader(fs, log, conf);
1642     try {
1643       WAL.Entry e;
1644       while ((e = in.next()) != null) {
1645         if (!WALEdit.isMetaEditFamily(e.getEdit().getCells().get(0))) {
1646           count++;
1647         }
1648       }
1649     } finally {
1650       try {
1651         in.close();
1652       } catch (IOException exception) {
1653         LOG.warn("Problem closing wal: " + exception.getMessage());
1654         LOG.debug("exception details.", exception);
1655       }
1656     }
1657     return count;
1658   }
1659 
1660   private void blockUntilNoRIT(ZooKeeperWatcher zkw, HMaster master) throws Exception {
1661     TEST_UTIL.waitUntilNoRegionsInTransition(60000);
1662   }
1663 
1664   private void putData(Region region, byte[] startRow, int numRows, byte [] qf,
1665       byte [] ...families)
1666       throws IOException {
1667     for(int i = 0; i < numRows; i++) {
1668       Put put = new Put(Bytes.add(startRow, Bytes.toBytes(i)));
1669       for(byte [] family : families) {
1670         put.add(family, qf, null);
1671       }
1672       region.put(put);
1673     }
1674   }
1675 
1676   /**
1677    * Load table with puts and deletes with expected values so that we can verify later
1678    */
1679   private void prepareData(final Table t, final byte[] f, final byte[] column) throws IOException {
1680     byte[] k = new byte[3];
1681 
1682     // add puts
1683     List<Put> puts = new ArrayList<>();
1684     for (byte b1 = 'a'; b1 <= 'z'; b1++) {
1685       for (byte b2 = 'a'; b2 <= 'z'; b2++) {
1686         for (byte b3 = 'a'; b3 <= 'z'; b3++) {
1687           k[0] = b1;
1688           k[1] = b2;
1689           k[2] = b3;
1690           Put put = new Put(k);
1691           put.add(f, column, k);
1692           puts.add(put);
1693         }
1694       }
1695     }
1696     t.put(puts);
1697     // add deletes
1698     for (byte b3 = 'a'; b3 <= 'z'; b3++) {
1699       k[0] = 'a';
1700       k[1] = 'a';
1701       k[2] = b3;
1702       Delete del = new Delete(k);
1703       t.delete(del);
1704     }
1705   }
1706 
1707   private void waitForCounter(AtomicLong ctr, long oldval, long newval,
1708       long timems) {
1709     long curt = System.currentTimeMillis();
1710     long endt = curt + timems;
1711     while (curt < endt) {
1712       if (ctr.get() == oldval) {
1713         Thread.yield();
1714         curt = System.currentTimeMillis();
1715       } else {
1716         assertEquals(newval, ctr.get());
1717         return;
1718       }
1719     }
1720     assertTrue(false);
1721   }
1722 
1723   private void abortMaster(MiniHBaseCluster cluster) throws InterruptedException {
1724     for (MasterThread mt : cluster.getLiveMasterThreads()) {
1725       if (mt.getMaster().isActiveMaster()) {
1726         mt.getMaster().abort("Aborting for tests", new Exception("Trace info"));
1727         mt.join();
1728         break;
1729       }
1730     }
1731     LOG.debug("Master is aborted");
1732   }
1733 
1734   /**
1735    * Find a RS that has regions of a table.
1736    * @param hasMetaRegion when true, the returned RS has hbase:meta region as well
1737    * @param tableName
1738    * @return
1739    * @throws Exception
1740    */
1741   private HRegionServer findRSToKill(boolean hasMetaRegion, String tableName) throws Exception {
1742     List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1743     List<HRegionInfo> regions = null;
1744     HRegionServer hrs = null;
1745 
1746     for (RegionServerThread rst: rsts) {
1747       hrs = rst.getRegionServer();
1748       while (rst.isAlive() && !hrs.isOnline()) {
1749         Thread.sleep(100);
1750       }
1751       if (!rst.isAlive()) {
1752         continue;
1753       }
1754       boolean isCarryingMeta = false;
1755       boolean foundTableRegion = false;
1756       regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1757       for (HRegionInfo region : regions) {
1758         if (region.isMetaRegion()) {
1759           isCarryingMeta = true;
1760         }
1761         if (tableName == null || region.getTable().getNameAsString().equals(tableName)) {
1762           foundTableRegion = true;
1763         }
1764         if (foundTableRegion && (isCarryingMeta || !hasMetaRegion)) {
1765           break;
1766         }
1767       }
1768       if (isCarryingMeta && hasMetaRegion) {
1769         // clients ask for a RS with META
1770         if (!foundTableRegion) {
1771           final HRegionServer destRS = hrs;
1772           // the RS doesn't have regions of the specified table so we need move one to this RS
1773           List<HRegionInfo> tableRegions =
1774               TEST_UTIL.getHBaseAdmin().getTableRegions(TableName.valueOf(tableName));
1775           final HRegionInfo hri = tableRegions.get(0);
1776           TEST_UTIL.getHBaseAdmin().move(hri.getEncodedNameAsBytes(),
1777               Bytes.toBytes(destRS.getServerName().getServerName()));
1778           // wait for region move completes
1779           final RegionStates regionStates =
1780               TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager().getRegionStates();
1781           TEST_UTIL.waitFor(45000, 200, new Waiter.Predicate<Exception>() {
1782             @Override
1783             public boolean evaluate() throws Exception {
1784               ServerName sn = regionStates.getRegionServerOfRegion(hri);
1785               return (sn != null && sn.equals(destRS.getServerName()));
1786             }
1787           });
1788         }
1789         return hrs;
1790       } else if (hasMetaRegion || isCarryingMeta) {
1791         continue;
1792       }
1793       if (foundTableRegion) break;
1794     }
1795 
1796     return hrs;
1797   }
1798 }