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  package org.apache.hadoop.hbase.util;
19  
20  import java.io.Closeable;
21  import java.io.FileNotFoundException;
22  import java.io.IOException;
23  import java.io.InterruptedIOException;
24  import java.io.PrintWriter;
25  import java.io.StringWriter;
26  import java.net.InetAddress;
27  import java.net.URI;
28  import java.util.ArrayList;
29  import java.util.Arrays;
30  import java.util.Collection;
31  import java.util.Collections;
32  import java.util.Comparator;
33  import java.util.HashMap;
34  import java.util.HashSet;
35  import java.util.Iterator;
36  import java.util.List;
37  import java.util.Map;
38  import java.util.Map.Entry;
39  import java.util.Set;
40  import java.util.SortedMap;
41  import java.util.SortedSet;
42  import java.util.TreeMap;
43  import java.util.TreeSet;
44  import java.util.concurrent.Callable;
45  import java.util.concurrent.ConcurrentSkipListMap;
46  import java.util.concurrent.ExecutionException;
47  import java.util.concurrent.ExecutorService;
48  import java.util.concurrent.Executors;
49  import java.util.concurrent.Future;
50  import java.util.concurrent.FutureTask;
51  import java.util.concurrent.ScheduledThreadPoolExecutor;
52  import java.util.concurrent.TimeUnit;
53  import java.util.concurrent.TimeoutException;
54  import java.util.concurrent.atomic.AtomicBoolean;
55  import java.util.concurrent.atomic.AtomicInteger;
56  
57  import com.google.common.base.Joiner;
58  import com.google.common.base.Preconditions;
59  import com.google.common.collect.ImmutableList;
60  import com.google.common.collect.Lists;
61  import com.google.common.collect.Multimap;
62  import com.google.common.collect.Ordering;
63  import com.google.common.collect.TreeMultimap;
64  import com.google.protobuf.ServiceException;
65  import org.apache.commons.io.IOUtils;
66  import org.apache.commons.lang.RandomStringUtils;
67  import org.apache.commons.lang.StringUtils;
68  import org.apache.commons.logging.Log;
69  import org.apache.commons.logging.LogFactory;
70  import org.apache.hadoop.conf.Configuration;
71  import org.apache.hadoop.conf.Configured;
72  import org.apache.hadoop.fs.FSDataOutputStream;
73  import org.apache.hadoop.fs.FileStatus;
74  import org.apache.hadoop.fs.FileSystem;
75  import org.apache.hadoop.fs.Path;
76  import org.apache.hadoop.fs.permission.FsAction;
77  import org.apache.hadoop.fs.permission.FsPermission;
78  import org.apache.hadoop.hbase.Abortable;
79  import org.apache.hadoop.hbase.Cell;
80  import org.apache.hadoop.hbase.ClusterStatus;
81  import org.apache.hadoop.hbase.HBaseConfiguration;
82  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
83  import org.apache.hadoop.hbase.HColumnDescriptor;
84  import org.apache.hadoop.hbase.HConstants;
85  import org.apache.hadoop.hbase.HRegionInfo;
86  import org.apache.hadoop.hbase.HRegionLocation;
87  import org.apache.hadoop.hbase.HTableDescriptor;
88  import org.apache.hadoop.hbase.KeyValue;
89  import org.apache.hadoop.hbase.KeyValueUtil;
90  import org.apache.hadoop.hbase.MasterNotRunningException;
91  import org.apache.hadoop.hbase.MetaTableAccessor;
92  import org.apache.hadoop.hbase.RegionLocations;
93  import org.apache.hadoop.hbase.ServerName;
94  import org.apache.hadoop.hbase.TableDescriptor;
95  import org.apache.hadoop.hbase.TableName;
96  import org.apache.hadoop.hbase.ZooKeeperConnectionException;
97  import org.apache.hadoop.hbase.classification.InterfaceAudience;
98  import org.apache.hadoop.hbase.classification.InterfaceStability;
99  import org.apache.hadoop.hbase.client.Admin;
100 import org.apache.hadoop.hbase.client.ClusterConnection;
101 import org.apache.hadoop.hbase.client.Connection;
102 import org.apache.hadoop.hbase.client.ConnectionFactory;
103 import org.apache.hadoop.hbase.client.Delete;
104 import org.apache.hadoop.hbase.client.Get;
105 import org.apache.hadoop.hbase.client.HConnection;
106 import org.apache.hadoop.hbase.client.Put;
107 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
108 import org.apache.hadoop.hbase.client.Result;
109 import org.apache.hadoop.hbase.client.RowMutations;
110 import org.apache.hadoop.hbase.client.Table;
111 import org.apache.hadoop.hbase.client.TableState;
112 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
113 import org.apache.hadoop.hbase.io.hfile.HFile;
114 import org.apache.hadoop.hbase.master.MasterFileSystem;
115 import org.apache.hadoop.hbase.master.RegionState;
116 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
117 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService.BlockingInterface;
118 import org.apache.hadoop.hbase.regionserver.HRegion;
119 import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
120 import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
121 import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL;
122 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
123 import org.apache.hadoop.hbase.security.AccessDeniedException;
124 import org.apache.hadoop.hbase.security.UserProvider;
125 import org.apache.hadoop.hbase.util.Bytes.ByteArrayComparator;
126 import org.apache.hadoop.hbase.util.HBaseFsck.ErrorReporter.ERROR_CODE;
127 import org.apache.hadoop.hbase.util.hbck.HFileCorruptionChecker;
128 import org.apache.hadoop.hbase.util.hbck.TableIntegrityErrorHandler;
129 import org.apache.hadoop.hbase.util.hbck.TableIntegrityErrorHandlerImpl;
130 import org.apache.hadoop.hbase.util.hbck.TableLockChecker;
131 import org.apache.hadoop.hbase.wal.WAL;
132 import org.apache.hadoop.hbase.wal.WALFactory;
133 import org.apache.hadoop.hbase.wal.WALSplitter;
134 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
135 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
136 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
137 import org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException;
138 import org.apache.hadoop.ipc.RemoteException;
139 import org.apache.hadoop.security.UserGroupInformation;
140 import org.apache.hadoop.util.ReflectionUtils;
141 import org.apache.hadoop.util.Tool;
142 import org.apache.hadoop.util.ToolRunner;
143 import org.apache.zookeeper.KeeperException;
144 
145 /**
146  * HBaseFsck (hbck) is a tool for checking and repairing region consistency and
147  * table integrity problems in a corrupted HBase.
148  * <p>
149  * Region consistency checks verify that hbase:meta, region deployment on region
150  * servers and the state of data in HDFS (.regioninfo files) all are in
151  * accordance.
152  * <p>
153  * Table integrity checks verify that all possible row keys resolve to exactly
154  * one region of a table.  This means there are no individual degenerate
155  * or backwards regions; no holes between regions; and that there are no
156  * overlapping regions.
157  * <p>
158  * The general repair strategy works in two phases:
159  * <ol>
160  * <li> Repair Table Integrity on HDFS. (merge or fabricate regions)
161  * <li> Repair Region Consistency with hbase:meta and assignments
162  * </ol>
163  * <p>
164  * For table integrity repairs, the tables' region directories are scanned
165  * for .regioninfo files.  Each table's integrity is then verified.  If there
166  * are any orphan regions (regions with no .regioninfo files) or holes, new
167  * regions are fabricated.  Backwards regions are sidelined as well as empty
168  * degenerate (endkey==startkey) regions.  If there are any overlapping regions,
169  * a new region is created and all data is merged into the new region.
170  * <p>
171  * Table integrity repairs deal solely with HDFS and could potentially be done
172  * offline -- the hbase region servers or master do not need to be running.
173  * This phase can eventually be used to completely reconstruct the hbase:meta table in
174  * an offline fashion.
175  * <p>
176  * Region consistency requires three conditions -- 1) valid .regioninfo file
177  * present in an HDFS region dir,  2) valid row with .regioninfo data in META,
178  * and 3) a region is deployed only at the regionserver that was assigned to
179  * with proper state in the master.
180  * <p>
181  * Region consistency repairs require hbase to be online so that hbck can
182  * contact the HBase master and region servers.  The hbck#connect() method must
183  * first be called successfully.  Much of the region consistency information
184  * is transient and less risky to repair.
185  * <p>
186  * If hbck is run from the command line, there are a handful of arguments that
187  * can be used to limit the kinds of repairs hbck will do.  See the code in
188  * {@link #printUsageAndExit()} for more details.
189  */
190 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
191 @InterfaceStability.Evolving
192 public class HBaseFsck extends Configured implements Closeable {
193   public static final long DEFAULT_TIME_LAG = 60000; // default value of 1 minute
194   public static final long DEFAULT_SLEEP_BEFORE_RERUN = 10000;
195   private static final int MAX_NUM_THREADS = 50; // #threads to contact regions
196   private static boolean rsSupportsOffline = true;
197   private static final int DEFAULT_OVERLAPS_TO_SIDELINE = 2;
198   private static final int DEFAULT_MAX_MERGE = 5;
199   private static final String TO_BE_LOADED = "to_be_loaded";
200   private static final String HBCK_LOCK_FILE = "hbase-hbck.lock";
201   private static final int DEFAULT_MAX_LOCK_FILE_ATTEMPTS = 5;
202   private static final int DEFAULT_LOCK_FILE_ATTEMPT_SLEEP_INTERVAL = 200; // milliseconds
203   private static final int DEFAULT_LOCK_FILE_ATTEMPT_MAX_SLEEP_TIME = 5000; // milliseconds
204   // We have to set the timeout value > HdfsConstants.LEASE_SOFTLIMIT_PERIOD.
205   // In HADOOP-2.6 and later, the Namenode proxy now created with custom RetryPolicy for
206   // AlreadyBeingCreatedException which is implies timeout on this operations up to
207   // HdfsConstants.LEASE_SOFTLIMIT_PERIOD (60 seconds).
208   private static final int DEFAULT_WAIT_FOR_LOCK_TIMEOUT = 80; // seconds
209 
210   /**********************
211    * Internal resources
212    **********************/
213   private static final Log LOG = LogFactory.getLog(HBaseFsck.class.getName());
214   private ClusterStatus status;
215   private ClusterConnection connection;
216   private Admin admin;
217   private Table meta;
218   // threads to do ||izable tasks: retrieve data from regionservers, handle overlapping regions
219   protected ExecutorService executor;
220   private long startMillis = EnvironmentEdgeManager.currentTime();
221   private HFileCorruptionChecker hfcc;
222   private int retcode = 0;
223   private Path HBCK_LOCK_PATH;
224   private FSDataOutputStream hbckOutFd;
225   // This lock is to prevent cleanup of balancer resources twice between
226   // ShutdownHook and the main code. We cleanup only if the connect() is
227   // successful
228   private final AtomicBoolean hbckLockCleanup = new AtomicBoolean(false);
229 
230   /***********
231    * Options
232    ***********/
233   private static boolean details = false; // do we display the full report
234   private long timelag = DEFAULT_TIME_LAG; // tables whose modtime is older
235   private boolean fixAssignments = false; // fix assignment errors?
236   private boolean fixMeta = false; // fix meta errors?
237   private boolean checkHdfs = true; // load and check fs consistency?
238   private boolean fixHdfsHoles = false; // fix fs holes?
239   private boolean fixHdfsOverlaps = false; // fix fs overlaps (risky)
240   private boolean fixHdfsOrphans = false; // fix fs holes (missing .regioninfo)
241   private boolean fixTableOrphans = false; // fix fs holes (missing .tableinfo)
242   private boolean fixVersionFile = false; // fix missing hbase.version file in hdfs
243   private boolean fixSplitParents = false; // fix lingering split parents
244   private boolean fixReferenceFiles = false; // fix lingering reference store file
245   private boolean fixEmptyMetaCells = false; // fix (remove) empty REGIONINFO_QUALIFIER rows
246   private boolean fixTableLocks = false; // fix table locks which are expired
247   private boolean fixAny = false; // Set to true if any of the fix is required.
248 
249   // limit checking/fixes to listed tables, if empty attempt to check/fix all
250   // hbase:meta are always checked
251   private Set<TableName> tablesIncluded = new HashSet<TableName>();
252   private int maxMerge = DEFAULT_MAX_MERGE; // maximum number of overlapping regions to merge
253   // maximum number of overlapping regions to sideline
254   private int maxOverlapsToSideline = DEFAULT_OVERLAPS_TO_SIDELINE;
255   private boolean sidelineBigOverlaps = false; // sideline overlaps with >maxMerge regions
256   private Path sidelineDir = null;
257 
258   private boolean rerun = false; // if we tried to fix something, rerun hbck
259   private static boolean summary = false; // if we want to print less output
260   private boolean checkMetaOnly = false;
261   private boolean checkRegionBoundaries = false;
262   private boolean ignorePreCheckPermission = false; // if pre-check permission
263 
264   /*********
265    * State
266    *********/
267   final private ErrorReporter errors;
268   int fixes = 0;
269 
270   /**
271    * This map contains the state of all hbck items.  It maps from encoded region
272    * name to HbckInfo structure.  The information contained in HbckInfo is used
273    * to detect and correct consistency (hdfs/meta/deployment) problems.
274    */
275   private TreeMap<String, HbckInfo> regionInfoMap = new TreeMap<String, HbckInfo>();
276   // Empty regioninfo qualifiers in hbase:meta
277   private Set<Result> emptyRegionInfoQualifiers = new HashSet<Result>();
278 
279   /**
280    * This map from Tablename -> TableInfo contains the structures necessary to
281    * detect table consistency problems (holes, dupes, overlaps).  It is sorted
282    * to prevent dupes.
283    *
284    * If tablesIncluded is empty, this map contains all tables.
285    * Otherwise, it contains only meta tables and tables in tablesIncluded,
286    * unless checkMetaOnly is specified, in which case, it contains only
287    * the meta table
288    */
289   private SortedMap<TableName, TableInfo> tablesInfo =
290       new ConcurrentSkipListMap<TableName, TableInfo>();
291 
292   /**
293    * When initially looking at HDFS, we attempt to find any orphaned data.
294    */
295   private List<HbckInfo> orphanHdfsDirs = Collections.synchronizedList(new ArrayList<HbckInfo>());
296 
297   private Map<TableName, Set<String>> orphanTableDirs =
298       new HashMap<TableName, Set<String>>();
299   private Map<TableName, TableState> tableStates =
300       new HashMap<TableName, TableState>();
301   private final RetryCounterFactory lockFileRetryCounterFactory;
302 
303   private Map<TableName, Set<String>> skippedRegions = new HashMap<TableName, Set<String>>();
304 
305   /**
306    * Constructor
307    *
308    * @param conf Configuration object
309    * @throws MasterNotRunningException if the master is not running
310    * @throws ZooKeeperConnectionException if unable to connect to ZooKeeper
311    */
312   public HBaseFsck(Configuration conf) throws MasterNotRunningException,
313       ZooKeeperConnectionException, IOException, ClassNotFoundException {
314     super(conf);
315     // make a copy, just to be sure we're not overriding someone else's config
316     setConf(HBaseConfiguration.create(getConf()));
317     // disable blockcache for tool invocation, see HBASE-10500
318     getConf().setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0);
319     // Disable usage of meta replicas in hbck
320     getConf().setBoolean(HConstants.USE_META_REPLICAS, false);
321     errors = getErrorReporter(conf);
322 
323     int numThreads = conf.getInt("hbasefsck.numthreads", MAX_NUM_THREADS);
324     executor = new ScheduledThreadPoolExecutor(numThreads, Threads.newDaemonThreadFactory("hbasefsck"));
325     lockFileRetryCounterFactory = new RetryCounterFactory(
326       getConf().getInt("hbase.hbck.lockfile.attempts", DEFAULT_MAX_LOCK_FILE_ATTEMPTS),
327       getConf().getInt(
328         "hbase.hbck.lockfile.attempt.sleep.interval", DEFAULT_LOCK_FILE_ATTEMPT_SLEEP_INTERVAL),
329       getConf().getInt(
330         "hbase.hbck.lockfile.attempt.maxsleeptime", DEFAULT_LOCK_FILE_ATTEMPT_MAX_SLEEP_TIME));
331   }
332 
333   /**
334    * Constructor
335    *
336    * @param conf
337    *          Configuration object
338    * @throws MasterNotRunningException
339    *           if the master is not running
340    * @throws ZooKeeperConnectionException
341    *           if unable to connect to ZooKeeper
342    */
343   public HBaseFsck(Configuration conf, ExecutorService exec) throws MasterNotRunningException,
344       ZooKeeperConnectionException, IOException, ClassNotFoundException {
345     super(conf);
346     errors = getErrorReporter(getConf());
347     this.executor = exec;
348     lockFileRetryCounterFactory = new RetryCounterFactory(
349       getConf().getInt("hbase.hbck.lockfile.attempts", DEFAULT_MAX_LOCK_FILE_ATTEMPTS),
350       getConf().getInt(
351         "hbase.hbck.lockfile.attempt.sleep.interval", DEFAULT_LOCK_FILE_ATTEMPT_SLEEP_INTERVAL),
352       getConf().getInt(
353         "hbase.hbck.lockfile.attempt.maxsleeptime", DEFAULT_LOCK_FILE_ATTEMPT_MAX_SLEEP_TIME));
354   }
355 
356   private class FileLockCallable implements Callable<FSDataOutputStream> {
357     RetryCounter retryCounter;
358 
359     public FileLockCallable(RetryCounter retryCounter) {
360       this.retryCounter = retryCounter;
361     }
362     @Override
363     public FSDataOutputStream call() throws IOException {
364       try {
365         FileSystem fs = FSUtils.getCurrentFileSystem(getConf());
366         FsPermission defaultPerms = FSUtils.getFilePermissions(fs, getConf(),
367             HConstants.DATA_FILE_UMASK_KEY);
368         Path tmpDir = new Path(FSUtils.getRootDir(getConf()), HConstants.HBASE_TEMP_DIRECTORY);
369         fs.mkdirs(tmpDir);
370         HBCK_LOCK_PATH = new Path(tmpDir, HBCK_LOCK_FILE);
371         final FSDataOutputStream out = createFileWithRetries(fs, HBCK_LOCK_PATH, defaultPerms);
372         out.writeBytes(InetAddress.getLocalHost().toString());
373         out.flush();
374         return out;
375       } catch(RemoteException e) {
376         if(AlreadyBeingCreatedException.class.getName().equals(e.getClassName())){
377           return null;
378         } else {
379           throw e;
380         }
381       }
382     }
383 
384     private FSDataOutputStream createFileWithRetries(final FileSystem fs,
385         final Path hbckLockFilePath, final FsPermission defaultPerms)
386         throws IOException {
387 
388       IOException exception = null;
389       do {
390         try {
391           return FSUtils.create(fs, hbckLockFilePath, defaultPerms, false);
392         } catch (IOException ioe) {
393           LOG.info("Failed to create lock file " + hbckLockFilePath.getName()
394               + ", try=" + (retryCounter.getAttemptTimes() + 1) + " of "
395               + retryCounter.getMaxAttempts());
396           LOG.debug("Failed to create lock file " + hbckLockFilePath.getName(), 
397               ioe);
398           try {
399             exception = ioe;
400             retryCounter.sleepUntilNextRetry();
401           } catch (InterruptedException ie) {
402             throw (InterruptedIOException) new InterruptedIOException(
403                 "Can't create lock file " + hbckLockFilePath.getName())
404             .initCause(ie);
405           }
406         }
407       } while (retryCounter.shouldRetry());
408 
409       throw exception;
410     }
411   }
412 
413   /**
414    * This method maintains a lock using a file. If the creation fails we return null
415    *
416    * @return FSDataOutputStream object corresponding to the newly opened lock file
417    * @throws IOException
418    */
419   private FSDataOutputStream checkAndMarkRunningHbck() throws IOException {
420     RetryCounter retryCounter = lockFileRetryCounterFactory.create();
421     FileLockCallable callable = new FileLockCallable(retryCounter);
422     ExecutorService executor = Executors.newFixedThreadPool(1);
423     FutureTask<FSDataOutputStream> futureTask = new FutureTask<FSDataOutputStream>(callable);
424     executor.execute(futureTask);
425     final int timeoutInSeconds = getConf().getInt(
426       "hbase.hbck.lockfile.maxwaittime", DEFAULT_WAIT_FOR_LOCK_TIMEOUT);
427     FSDataOutputStream stream = null;
428     try {
429       stream = futureTask.get(timeoutInSeconds, TimeUnit.SECONDS);
430     } catch (ExecutionException ee) {
431       LOG.warn("Encountered exception when opening lock file", ee);
432     } catch (InterruptedException ie) {
433       LOG.warn("Interrupted when opening lock file", ie);
434       Thread.currentThread().interrupt();
435     } catch (TimeoutException exception) {
436       // took too long to obtain lock
437       LOG.warn("Took more than " + timeoutInSeconds + " seconds in obtaining lock");
438       futureTask.cancel(true);
439     } finally {
440       executor.shutdownNow();
441     }
442     return stream;
443   }
444 
445   private void unlockHbck() {
446     if (hbckLockCleanup.compareAndSet(true, false)) {
447       RetryCounter retryCounter = lockFileRetryCounterFactory.create();
448       do {
449         try {
450           IOUtils.closeQuietly(hbckOutFd);
451           FSUtils.delete(FSUtils.getCurrentFileSystem(getConf()),
452               HBCK_LOCK_PATH, true);
453           LOG.info("Finishing hbck");
454           return;
455         } catch (IOException ioe) {
456           LOG.info("Failed to delete " + HBCK_LOCK_PATH + ", try="
457               + (retryCounter.getAttemptTimes() + 1) + " of "
458               + retryCounter.getMaxAttempts());
459           LOG.debug("Failed to delete " + HBCK_LOCK_PATH, ioe);
460           try {
461             retryCounter.sleepUntilNextRetry();
462           } catch (InterruptedException ie) {
463             Thread.currentThread().interrupt();
464             LOG.warn("Interrupted while deleting lock file" +
465                 HBCK_LOCK_PATH);
466             return;
467           }
468         }
469       } while (retryCounter.shouldRetry());
470     }
471   }
472 
473   /**
474    * To repair region consistency, one must call connect() in order to repair
475    * online state.
476    */
477   public void connect() throws IOException {
478 
479     // Check if another instance of balancer is running
480     hbckOutFd = checkAndMarkRunningHbck();
481     if (hbckOutFd == null) {
482       setRetCode(-1);
483       LOG.error("Another instance of hbck is running, exiting this instance.[If you are sure" +
484           " no other instance is running, delete the lock file " +
485           HBCK_LOCK_PATH + " and rerun the tool]");
486       throw new IOException("Duplicate hbck - Abort");
487     }
488 
489     // Make sure to cleanup the lock
490     hbckLockCleanup.set(true);
491 
492     // Add a shutdown hook to this thread, in case user tries to
493     // kill the hbck with a ctrl-c, we want to cleanup the lock so that
494     // it is available for further calls
495     Runtime.getRuntime().addShutdownHook(new Thread() {
496       @Override
497       public void run() {
498         IOUtils.closeQuietly(HBaseFsck.this);
499         unlockHbck();
500       }
501     });
502 
503     LOG.info("Launching hbck");
504 
505     connection = (ClusterConnection)ConnectionFactory.createConnection(getConf());
506     admin = connection.getAdmin();
507     meta = connection.getTable(TableName.META_TABLE_NAME);
508     status = admin.getClusterStatus();
509   }
510 
511   /**
512    * Get deployed regions according to the region servers.
513    */
514   private void loadDeployedRegions() throws IOException, InterruptedException {
515     // From the master, get a list of all known live region servers
516     Collection<ServerName> regionServers = status.getServers();
517     errors.print("Number of live region servers: " + regionServers.size());
518     if (details) {
519       for (ServerName rsinfo: regionServers) {
520         errors.print("  " + rsinfo.getServerName());
521       }
522     }
523 
524     // From the master, get a list of all dead region servers
525     Collection<ServerName> deadRegionServers = status.getDeadServerNames();
526     errors.print("Number of dead region servers: " + deadRegionServers.size());
527     if (details) {
528       for (ServerName name: deadRegionServers) {
529         errors.print("  " + name);
530       }
531     }
532 
533     // Print the current master name and state
534     errors.print("Master: " + status.getMaster());
535 
536     // Print the list of all backup masters
537     Collection<ServerName> backupMasters = status.getBackupMasters();
538     errors.print("Number of backup masters: " + backupMasters.size());
539     if (details) {
540       for (ServerName name: backupMasters) {
541         errors.print("  " + name);
542       }
543     }
544 
545     errors.print("Average load: " + status.getAverageLoad());
546     errors.print("Number of requests: " + status.getRequestsCount());
547     errors.print("Number of regions: " + status.getRegionsCount());
548 
549     Map<String, RegionState> rits = status.getRegionsInTransition();
550     errors.print("Number of regions in transition: " + rits.size());
551     if (details) {
552       for (RegionState state: rits.values()) {
553         errors.print("  " + state.toDescriptiveString());
554       }
555     }
556 
557     // Determine what's deployed
558     processRegionServers(regionServers);
559   }
560 
561   /**
562    * Clear the current state of hbck.
563    */
564   private void clearState() {
565     // Make sure regionInfo is empty before starting
566     fixes = 0;
567     regionInfoMap.clear();
568     emptyRegionInfoQualifiers.clear();
569     tableStates.clear();
570     errors.clear();
571     tablesInfo.clear();
572     orphanHdfsDirs.clear();
573     skippedRegions.clear();
574   }
575 
576   /**
577    * This repair method analyzes hbase data in hdfs and repairs it to satisfy
578    * the table integrity rules.  HBase doesn't need to be online for this
579    * operation to work.
580    */
581   public void offlineHdfsIntegrityRepair() throws IOException, InterruptedException {
582     // Initial pass to fix orphans.
583     if (shouldCheckHdfs() && (shouldFixHdfsOrphans() || shouldFixHdfsHoles()
584         || shouldFixHdfsOverlaps() || shouldFixTableOrphans())) {
585       LOG.info("Loading regioninfos HDFS");
586       // if nothing is happening this should always complete in two iterations.
587       int maxIterations = getConf().getInt("hbase.hbck.integrityrepair.iterations.max", 3);
588       int curIter = 0;
589       do {
590         clearState(); // clears hbck state and reset fixes to 0 and.
591         // repair what's on HDFS
592         restoreHdfsIntegrity();
593         curIter++;// limit the number of iterations.
594       } while (fixes > 0 && curIter <= maxIterations);
595 
596       // Repairs should be done in the first iteration and verification in the second.
597       // If there are more than 2 passes, something funny has happened.
598       if (curIter > 2) {
599         if (curIter == maxIterations) {
600           LOG.warn("Exiting integrity repairs after max " + curIter + " iterations. "
601               + "Tables integrity may not be fully repaired!");
602         } else {
603           LOG.info("Successfully exiting integrity repairs after " + curIter + " iterations");
604         }
605       }
606     }
607   }
608 
609   /**
610    * This repair method requires the cluster to be online since it contacts
611    * region servers and the masters.  It makes each region's state in HDFS, in
612    * hbase:meta, and deployments consistent.
613    *
614    * @return If &gt; 0 , number of errors detected, if &lt; 0 there was an unrecoverable
615    * error.  If 0, we have a clean hbase.
616    */
617   public int onlineConsistencyRepair() throws IOException, KeeperException,
618     InterruptedException {
619     clearState();
620 
621     // get regions according to what is online on each RegionServer
622     loadDeployedRegions();
623     // check whether hbase:meta is deployed and online
624     recordMetaRegion();
625     // Check if hbase:meta is found only once and in the right place
626     if (!checkMetaRegion()) {
627       String errorMsg = "hbase:meta table is not consistent. ";
628       if (shouldFixAssignments()) {
629         errorMsg += "HBCK will try fixing it. Rerun once hbase:meta is back to consistent state.";
630       } else {
631         errorMsg += "Run HBCK with proper fix options to fix hbase:meta inconsistency.";
632       }
633       errors.reportError(errorMsg + " Exiting...");
634       return -2;
635     }
636     // Not going with further consistency check for tables when hbase:meta itself is not consistent.
637     LOG.info("Loading regionsinfo from the hbase:meta table");
638     boolean success = loadMetaEntries();
639     if (!success) return -1;
640 
641     // Empty cells in hbase:meta?
642     reportEmptyMetaCells();
643 
644     // Check if we have to cleanup empty REGIONINFO_QUALIFIER rows from hbase:meta
645     if (shouldFixEmptyMetaCells()) {
646       fixEmptyMetaCells();
647     }
648 
649     // get a list of all tables that have not changed recently.
650     if (!checkMetaOnly) {
651       reportTablesInFlux();
652     }
653 
654     // Get disabled tables states
655     loadTableStates();
656 
657     // load regiondirs and regioninfos from HDFS
658     if (shouldCheckHdfs()) {
659       LOG.info("Loading region directories from HDFS");
660       loadHdfsRegionDirs();
661       LOG.info("Loading region information from HDFS");
662       loadHdfsRegionInfos();
663     }
664 
665     // fix the orphan tables
666     fixOrphanTables();
667 
668     LOG.info("Checking and fixing region consistency");
669 
670     // Check and fix consistency
671     checkAndFixConsistency();
672 
673     // Check integrity (does not fix)
674     checkIntegrity();
675     return errors.getErrorList().size();
676   }
677 
678   /**
679    * Contacts the master and prints out cluster-wide information
680    * @return 0 on success, non-zero on failure
681    */
682   public int onlineHbck() throws IOException, KeeperException, InterruptedException, ServiceException {
683     // print hbase server version
684     errors.print("Version: " + status.getHBaseVersion());
685     offlineHdfsIntegrityRepair();
686 
687     // turn the balancer off
688     boolean oldBalancer = admin.setBalancerRunning(false, true);
689     try {
690       onlineConsistencyRepair();
691     }
692     finally {
693       admin.setBalancerRunning(oldBalancer, false);
694     }
695 
696     if (checkRegionBoundaries) {
697       checkRegionBoundaries();
698     }
699 
700     offlineReferenceFileRepair();
701 
702     checkAndFixTableLocks();
703 
704     // Remove the hbck lock
705     unlockHbck();
706 
707     // Print table summary
708     printTableSummary(tablesInfo);
709     return errors.summarize();
710   }
711 
712   public static byte[] keyOnly (byte[] b) {
713     if (b == null)
714       return b;
715     int rowlength = Bytes.toShort(b, 0);
716     byte[] result = new byte[rowlength];
717     System.arraycopy(b, Bytes.SIZEOF_SHORT, result, 0, rowlength);
718     return result;
719   }
720 
721   @Override
722   public void close() throws IOException {
723     try {
724       unlockHbck();
725     } catch (Exception io) {
726       LOG.warn(io);
727     } finally {
728       IOUtils.closeQuietly(admin);
729       IOUtils.closeQuietly(meta);
730       IOUtils.closeQuietly(connection);
731     }
732   }
733 
734   private static class RegionBoundariesInformation {
735     public byte [] regionName;
736     public byte [] metaFirstKey;
737     public byte [] metaLastKey;
738     public byte [] storesFirstKey;
739     public byte [] storesLastKey;
740     @Override
741     public String toString () {
742       return "regionName=" + Bytes.toStringBinary(regionName) +
743              "\nmetaFirstKey=" + Bytes.toStringBinary(metaFirstKey) +
744              "\nmetaLastKey=" + Bytes.toStringBinary(metaLastKey) +
745              "\nstoresFirstKey=" + Bytes.toStringBinary(storesFirstKey) +
746              "\nstoresLastKey=" + Bytes.toStringBinary(storesLastKey);
747     }
748   }
749 
750   public void checkRegionBoundaries() {
751     try {
752       ByteArrayComparator comparator = new ByteArrayComparator();
753       List<HRegionInfo> regions = MetaTableAccessor.getAllRegions(connection, true);
754       final RegionBoundariesInformation currentRegionBoundariesInformation =
755           new RegionBoundariesInformation();
756       Path hbaseRoot = FSUtils.getRootDir(getConf());
757       for (HRegionInfo regionInfo : regions) {
758         Path tableDir = FSUtils.getTableDir(hbaseRoot, regionInfo.getTable());
759         currentRegionBoundariesInformation.regionName = regionInfo.getRegionName();
760         // For each region, get the start and stop key from the META and compare them to the
761         // same information from the Stores.
762         Path path = new Path(tableDir, regionInfo.getEncodedName());
763         FileSystem fs = path.getFileSystem(getConf());
764         FileStatus[] files = fs.listStatus(path);
765         // For all the column families in this region...
766         byte[] storeFirstKey = null;
767         byte[] storeLastKey = null;
768         for (FileStatus file : files) {
769           String fileName = file.getPath().toString();
770           fileName = fileName.substring(fileName.lastIndexOf("/") + 1);
771           if (!fileName.startsWith(".") && !fileName.endsWith("recovered.edits")) {
772             FileStatus[] storeFiles = fs.listStatus(file.getPath());
773             // For all the stores in this column family.
774             for (FileStatus storeFile : storeFiles) {
775               HFile.Reader reader = HFile.createReader(fs, storeFile.getPath(), new CacheConfig(
776                   getConf()), getConf());
777               if ((reader.getFirstKey() != null)
778                   && ((storeFirstKey == null) || (comparator.compare(storeFirstKey,
779                       ((KeyValue.KeyOnlyKeyValue) reader.getFirstKey()).getKey()) > 0))) {
780                 storeFirstKey = ((KeyValue.KeyOnlyKeyValue)reader.getFirstKey()).getKey();
781               }
782               if ((reader.getLastKey() != null)
783                   && ((storeLastKey == null) || (comparator.compare(storeLastKey,
784                       reader.getLastKey())) < 0)) {
785                 storeLastKey = reader.getLastKey();
786               }
787               reader.close();
788             }
789           }
790         }
791         currentRegionBoundariesInformation.metaFirstKey = regionInfo.getStartKey();
792         currentRegionBoundariesInformation.metaLastKey = regionInfo.getEndKey();
793         currentRegionBoundariesInformation.storesFirstKey = storeFirstKey;
794         currentRegionBoundariesInformation.storesLastKey = keyOnly(storeLastKey);
795         if (currentRegionBoundariesInformation.metaFirstKey.length == 0)
796           currentRegionBoundariesInformation.metaFirstKey = null;
797         if (currentRegionBoundariesInformation.metaLastKey.length == 0)
798           currentRegionBoundariesInformation.metaLastKey = null;
799 
800         // For a region to be correct, we need the META start key to be smaller or equal to the
801         // smallest start key from all the stores, and the start key from the next META entry to
802         // be bigger than the last key from all the current stores. First region start key is null;
803         // Last region end key is null; some regions can be empty and not have any store.
804 
805         boolean valid = true;
806         // Checking start key.
807         if ((currentRegionBoundariesInformation.storesFirstKey != null)
808             && (currentRegionBoundariesInformation.metaFirstKey != null)) {
809           valid = valid
810               && comparator.compare(currentRegionBoundariesInformation.storesFirstKey,
811                 currentRegionBoundariesInformation.metaFirstKey) >= 0;
812         }
813         // Checking stop key.
814         if ((currentRegionBoundariesInformation.storesLastKey != null)
815             && (currentRegionBoundariesInformation.metaLastKey != null)) {
816           valid = valid
817               && comparator.compare(currentRegionBoundariesInformation.storesLastKey,
818                 currentRegionBoundariesInformation.metaLastKey) < 0;
819         }
820         if (!valid) {
821           errors.reportError(ERROR_CODE.BOUNDARIES_ERROR, "Found issues with regions boundaries",
822             tablesInfo.get(regionInfo.getTable()));
823           LOG.warn("Region's boundaries not alligned between stores and META for:");
824           LOG.warn(currentRegionBoundariesInformation);
825         }
826       }
827     } catch (IOException e) {
828       LOG.error(e);
829     }
830   }
831 
832   /**
833    * Iterates through the list of all orphan/invalid regiondirs.
834    */
835   private void adoptHdfsOrphans(Collection<HbckInfo> orphanHdfsDirs) throws IOException {
836     for (HbckInfo hi : orphanHdfsDirs) {
837       LOG.info("Attempting to handle orphan hdfs dir: " + hi.getHdfsRegionDir());
838       adoptHdfsOrphan(hi);
839     }
840   }
841 
842   /**
843    * Orphaned regions are regions without a .regioninfo file in them.  We "adopt"
844    * these orphans by creating a new region, and moving the column families,
845    * recovered edits, WALs, into the new region dir.  We determine the region
846    * startkey and endkeys by looking at all of the hfiles inside the column
847    * families to identify the min and max keys. The resulting region will
848    * likely violate table integrity but will be dealt with by merging
849    * overlapping regions.
850    */
851   @SuppressWarnings("deprecation")
852   private void adoptHdfsOrphan(HbckInfo hi) throws IOException {
853     Path p = hi.getHdfsRegionDir();
854     FileSystem fs = p.getFileSystem(getConf());
855     FileStatus[] dirs = fs.listStatus(p);
856     if (dirs == null) {
857       LOG.warn("Attempt to adopt ophan hdfs region skipped becuase no files present in " +
858           p + ". This dir could probably be deleted.");
859       return ;
860     }
861 
862     TableName tableName = hi.getTableName();
863     TableInfo tableInfo = tablesInfo.get(tableName);
864     Preconditions.checkNotNull(tableInfo, "Table '" + tableName + "' not present!");
865     HTableDescriptor template = tableInfo.getHTD();
866 
867     // find min and max key values
868     Pair<byte[],byte[]> orphanRegionRange = null;
869     for (FileStatus cf : dirs) {
870       String cfName= cf.getPath().getName();
871       // TODO Figure out what the special dirs are
872       if (cfName.startsWith(".") || cfName.equals(HConstants.SPLIT_LOGDIR_NAME)) continue;
873 
874       FileStatus[] hfiles = fs.listStatus(cf.getPath());
875       for (FileStatus hfile : hfiles) {
876         byte[] start, end;
877         HFile.Reader hf = null;
878         try {
879           CacheConfig cacheConf = new CacheConfig(getConf());
880           hf = HFile.createReader(fs, hfile.getPath(), cacheConf, getConf());
881           hf.loadFileInfo();
882           Cell startKv = hf.getFirstKey();
883           start = startKv.getRow();
884           KeyValue endKv = KeyValueUtil.createKeyValueFromKey(hf.getLastKey());
885           end = endKv.getRow();
886         } catch (IOException ioe) {
887           LOG.warn("Problem reading orphan file " + hfile + ", skipping");
888           continue;
889         } catch (NullPointerException ioe) {
890           LOG.warn("Orphan file " + hfile + " is possibly corrupted HFile, skipping");
891           continue;
892         } finally {
893           if (hf != null) {
894             hf.close();
895           }
896         }
897 
898         // expand the range to include the range of all hfiles
899         if (orphanRegionRange == null) {
900           // first range
901           orphanRegionRange = new Pair<byte[], byte[]>(start, end);
902         } else {
903           // TODO add test
904 
905           // expand range only if the hfile is wider.
906           if (Bytes.compareTo(orphanRegionRange.getFirst(), start) > 0) {
907             orphanRegionRange.setFirst(start);
908           }
909           if (Bytes.compareTo(orphanRegionRange.getSecond(), end) < 0 ) {
910             orphanRegionRange.setSecond(end);
911           }
912         }
913       }
914     }
915     if (orphanRegionRange == null) {
916       LOG.warn("No data in dir " + p + ", sidelining data");
917       fixes++;
918       sidelineRegionDir(fs, hi);
919       return;
920     }
921     LOG.info("Min max keys are : [" + Bytes.toString(orphanRegionRange.getFirst()) + ", " +
922         Bytes.toString(orphanRegionRange.getSecond()) + ")");
923 
924     // create new region on hdfs.  move data into place.
925     HRegionInfo hri = new HRegionInfo(template.getTableName(), orphanRegionRange.getFirst(), orphanRegionRange.getSecond());
926     LOG.info("Creating new region : " + hri);
927     HRegion region = HBaseFsckRepair.createHDFSRegionDir(getConf(), hri, template);
928     Path target = region.getRegionFileSystem().getRegionDir();
929 
930     // rename all the data to new region
931     mergeRegionDirs(target, hi);
932     fixes++;
933   }
934 
935   /**
936    * This method determines if there are table integrity errors in HDFS.  If
937    * there are errors and the appropriate "fix" options are enabled, the method
938    * will first correct orphan regions making them into legit regiondirs, and
939    * then reload to merge potentially overlapping regions.
940    *
941    * @return number of table integrity errors found
942    */
943   private int restoreHdfsIntegrity() throws IOException, InterruptedException {
944     // Determine what's on HDFS
945     LOG.info("Loading HBase regioninfo from HDFS...");
946     loadHdfsRegionDirs(); // populating regioninfo table.
947 
948     int errs = errors.getErrorList().size();
949     // First time just get suggestions.
950     tablesInfo = loadHdfsRegionInfos(); // update tableInfos based on region info in fs.
951     checkHdfsIntegrity(false, false);
952 
953     if (errors.getErrorList().size() == errs) {
954       LOG.info("No integrity errors.  We are done with this phase. Glorious.");
955       return 0;
956     }
957 
958     if (shouldFixHdfsOrphans() && orphanHdfsDirs.size() > 0) {
959       adoptHdfsOrphans(orphanHdfsDirs);
960       // TODO optimize by incrementally adding instead of reloading.
961     }
962 
963     // Make sure there are no holes now.
964     if (shouldFixHdfsHoles()) {
965       clearState(); // this also resets # fixes.
966       loadHdfsRegionDirs();
967       tablesInfo = loadHdfsRegionInfos(); // update tableInfos based on region info in fs.
968       tablesInfo = checkHdfsIntegrity(shouldFixHdfsHoles(), false);
969     }
970 
971     // Now we fix overlaps
972     if (shouldFixHdfsOverlaps()) {
973       // second pass we fix overlaps.
974       clearState(); // this also resets # fixes.
975       loadHdfsRegionDirs();
976       tablesInfo = loadHdfsRegionInfos(); // update tableInfos based on region info in fs.
977       tablesInfo = checkHdfsIntegrity(false, shouldFixHdfsOverlaps());
978     }
979 
980     return errors.getErrorList().size();
981   }
982 
983   /**
984    * Scan all the store file names to find any lingering reference files,
985    * which refer to some none-exiting files. If "fix" option is enabled,
986    * any lingering reference file will be sidelined if found.
987    * <p>
988    * Lingering reference file prevents a region from opening. It has to
989    * be fixed before a cluster can start properly.
990    */
991   private void offlineReferenceFileRepair() throws IOException {
992     Configuration conf = getConf();
993     Path hbaseRoot = FSUtils.getRootDir(conf);
994     FileSystem fs = hbaseRoot.getFileSystem(conf);
995     LOG.info("Computing mapping of all store files");
996     Map<String, Path> allFiles = FSUtils.getTableStoreFilePathMap(fs, hbaseRoot, errors);
997     errors.print("");
998     LOG.info("Validating mapping using HDFS state");
999     for (Path path: allFiles.values()) {
1000       boolean isReference = false;
1001       try {
1002         isReference = StoreFileInfo.isReference(path);
1003       } catch (Throwable t) {
1004         // Ignore. Some files may not be store files at all.
1005         // For example, files under .oldlogs folder in hbase:meta
1006         // Warning message is already logged by
1007         // StoreFile#isReference.
1008       }
1009       if (!isReference) continue;
1010 
1011       Path referredToFile = StoreFileInfo.getReferredToFile(path);
1012       if (fs.exists(referredToFile)) continue;  // good, expected
1013 
1014       // Found a lingering reference file
1015       errors.reportError(ERROR_CODE.LINGERING_REFERENCE_HFILE,
1016         "Found lingering reference file " + path);
1017       if (!shouldFixReferenceFiles()) continue;
1018 
1019       // Now, trying to fix it since requested
1020       boolean success = false;
1021       String pathStr = path.toString();
1022 
1023       // A reference file path should be like
1024       // ${hbase.rootdir}/data/namespace/table_name/region_id/family_name/referred_file.region_name
1025       // Up 5 directories to get the root folder.
1026       // So the file will be sidelined to a similar folder structure.
1027       int index = pathStr.lastIndexOf(Path.SEPARATOR_CHAR);
1028       for (int i = 0; index > 0 && i < 5; i++) {
1029         index = pathStr.lastIndexOf(Path.SEPARATOR_CHAR, index - 1);
1030       }
1031       if (index > 0) {
1032         Path rootDir = getSidelineDir();
1033         Path dst = new Path(rootDir, pathStr.substring(index + 1));
1034         fs.mkdirs(dst.getParent());
1035         LOG.info("Trying to sildeline reference file "
1036           + path + " to " + dst);
1037         setShouldRerun();
1038 
1039         success = fs.rename(path, dst);
1040       }
1041       if (!success) {
1042         LOG.error("Failed to sideline reference file " + path);
1043       }
1044     }
1045   }
1046 
1047   /**
1048    * TODO -- need to add tests for this.
1049    */
1050   private void reportEmptyMetaCells() {
1051     errors.print("Number of empty REGIONINFO_QUALIFIER rows in hbase:meta: " +
1052       emptyRegionInfoQualifiers.size());
1053     if (details) {
1054       for (Result r: emptyRegionInfoQualifiers) {
1055         errors.print("  " + r);
1056       }
1057     }
1058   }
1059 
1060   /**
1061    * TODO -- need to add tests for this.
1062    */
1063   private void reportTablesInFlux() {
1064     AtomicInteger numSkipped = new AtomicInteger(0);
1065     HTableDescriptor[] allTables = getTables(numSkipped);
1066     errors.print("Number of Tables: " + allTables.length);
1067     if (details) {
1068       if (numSkipped.get() > 0) {
1069         errors.detail("Number of Tables in flux: " + numSkipped.get());
1070       }
1071       for (HTableDescriptor td : allTables) {
1072         errors.detail("  Table: " + td.getTableName() + "\t" +
1073                            (td.isReadOnly() ? "ro" : "rw") + "\t" +
1074                             (td.isMetaRegion() ? "META" : "    ") + "\t" +
1075                            " families: " + td.getFamilies().size());
1076       }
1077     }
1078   }
1079 
1080   public ErrorReporter getErrors() {
1081     return errors;
1082   }
1083 
1084   /**
1085    * Read the .regioninfo file from the file system.  If there is no
1086    * .regioninfo, add it to the orphan hdfs region list.
1087    */
1088   private void loadHdfsRegioninfo(HbckInfo hbi) throws IOException {
1089     Path regionDir = hbi.getHdfsRegionDir();
1090     if (regionDir == null) {
1091       LOG.warn("No HDFS region dir found: " + hbi + " meta=" + hbi.metaEntry);
1092       return;
1093     }
1094 
1095     if (hbi.hdfsEntry.hri != null) {
1096       // already loaded data
1097       return;
1098     }
1099 
1100     FileSystem fs = FileSystem.get(getConf());
1101     HRegionInfo hri = HRegionFileSystem.loadRegionInfoFileContent(fs, regionDir);
1102     LOG.debug("HRegionInfo read: " + hri.toString());
1103     hbi.hdfsEntry.hri = hri;
1104   }
1105 
1106   /**
1107    * Exception thrown when a integrity repair operation fails in an
1108    * unresolvable way.
1109    */
1110   public static class RegionRepairException extends IOException {
1111     private static final long serialVersionUID = 1L;
1112     final IOException ioe;
1113     public RegionRepairException(String s, IOException ioe) {
1114       super(s);
1115       this.ioe = ioe;
1116     }
1117   }
1118 
1119   /**
1120    * Populate hbi's from regionInfos loaded from file system.
1121    */
1122   private SortedMap<TableName, TableInfo> loadHdfsRegionInfos()
1123       throws IOException, InterruptedException {
1124     tablesInfo.clear(); // regenerating the data
1125     // generate region split structure
1126     Collection<HbckInfo> hbckInfos = regionInfoMap.values();
1127 
1128     // Parallelized read of .regioninfo files.
1129     List<WorkItemHdfsRegionInfo> hbis = new ArrayList<WorkItemHdfsRegionInfo>(hbckInfos.size());
1130     List<Future<Void>> hbiFutures;
1131 
1132     for (HbckInfo hbi : hbckInfos) {
1133       WorkItemHdfsRegionInfo work = new WorkItemHdfsRegionInfo(hbi, this, errors);
1134       hbis.add(work);
1135     }
1136 
1137     // Submit and wait for completion
1138     hbiFutures = executor.invokeAll(hbis);
1139 
1140     for(int i=0; i<hbiFutures.size(); i++) {
1141       WorkItemHdfsRegionInfo work = hbis.get(i);
1142       Future<Void> f = hbiFutures.get(i);
1143       try {
1144         f.get();
1145       } catch(ExecutionException e) {
1146         LOG.warn("Failed to read .regioninfo file for region " +
1147               work.hbi.getRegionNameAsString(), e.getCause());
1148       }
1149     }
1150 
1151     Path hbaseRoot = FSUtils.getRootDir(getConf());
1152     FileSystem fs = hbaseRoot.getFileSystem(getConf());
1153     // serialized table info gathering.
1154     for (HbckInfo hbi: hbckInfos) {
1155 
1156       if (hbi.getHdfsHRI() == null) {
1157         // was an orphan
1158         continue;
1159       }
1160 
1161 
1162       // get table name from hdfs, populate various HBaseFsck tables.
1163       TableName tableName = hbi.getTableName();
1164       if (tableName == null) {
1165         // There was an entry in hbase:meta not in the HDFS?
1166         LOG.warn("tableName was null for: " + hbi);
1167         continue;
1168       }
1169 
1170       TableInfo modTInfo = tablesInfo.get(tableName);
1171       if (modTInfo == null) {
1172         // only executed once per table.
1173         modTInfo = new TableInfo(tableName);
1174         tablesInfo.put(tableName, modTInfo);
1175         try {
1176           TableDescriptor htd =
1177               FSTableDescriptors.getTableDescriptorFromFs(fs, hbaseRoot, tableName);
1178           modTInfo.htds.add(htd.getHTableDescriptor());
1179         } catch (IOException ioe) {
1180           if (!orphanTableDirs.containsKey(tableName)) {
1181             LOG.warn("Unable to read .tableinfo from " + hbaseRoot, ioe);
1182             //should only report once for each table
1183             errors.reportError(ERROR_CODE.NO_TABLEINFO_FILE,
1184                 "Unable to read .tableinfo from " + hbaseRoot + "/" + tableName);
1185             Set<String> columns = new HashSet<String>();
1186             orphanTableDirs.put(tableName, getColumnFamilyList(columns, hbi));
1187           }
1188         }
1189       }
1190       if (!hbi.isSkipChecks()) {
1191         modTInfo.addRegionInfo(hbi);
1192       }
1193     }
1194 
1195     loadTableInfosForTablesWithNoRegion();
1196     errors.print("");
1197 
1198     return tablesInfo;
1199   }
1200 
1201   /**
1202    * To get the column family list according to the column family dirs
1203    * @param columns
1204    * @param hbi
1205    * @return a set of column families
1206    * @throws IOException
1207    */
1208   private Set<String> getColumnFamilyList(Set<String> columns, HbckInfo hbi) throws IOException {
1209     Path regionDir = hbi.getHdfsRegionDir();
1210     FileSystem fs = regionDir.getFileSystem(getConf());
1211     FileStatus[] subDirs = fs.listStatus(regionDir, new FSUtils.FamilyDirFilter(fs));
1212     for (FileStatus subdir : subDirs) {
1213       String columnfamily = subdir.getPath().getName();
1214       columns.add(columnfamily);
1215     }
1216     return columns;
1217   }
1218 
1219   /**
1220    * To fabricate a .tableinfo file with following contents<br>
1221    * 1. the correct tablename <br>
1222    * 2. the correct colfamily list<br>
1223    * 3. the default properties for both {@link HTableDescriptor} and {@link HColumnDescriptor}<br>
1224    * @throws IOException
1225    */
1226   private boolean fabricateTableInfo(FSTableDescriptors fstd, TableName tableName,
1227       Set<String> columns) throws IOException {
1228     if (columns ==null || columns.isEmpty()) return false;
1229     HTableDescriptor htd = new HTableDescriptor(tableName);
1230     for (String columnfamimly : columns) {
1231       htd.addFamily(new HColumnDescriptor(columnfamimly));
1232     }
1233     fstd.createTableDescriptor(new TableDescriptor(htd), true);
1234     return true;
1235   }
1236 
1237   /**
1238    * To fix the empty REGIONINFO_QUALIFIER rows from hbase:meta <br>
1239    * @throws IOException
1240    */
1241   public void fixEmptyMetaCells() throws IOException {
1242     if (shouldFixEmptyMetaCells() && !emptyRegionInfoQualifiers.isEmpty()) {
1243       LOG.info("Trying to fix empty REGIONINFO_QUALIFIER hbase:meta rows.");
1244       for (Result region : emptyRegionInfoQualifiers) {
1245         deleteMetaRegion(region.getRow());
1246         errors.getErrorList().remove(ERROR_CODE.EMPTY_META_CELL);
1247       }
1248       emptyRegionInfoQualifiers.clear();
1249     }
1250   }
1251 
1252   /**
1253    * To fix orphan table by creating a .tableinfo file under tableDir <br>
1254    * 1. if TableInfo is cached, to recover the .tableinfo accordingly <br>
1255    * 2. else create a default .tableinfo file with following items<br>
1256    * &nbsp;2.1 the correct tablename <br>
1257    * &nbsp;2.2 the correct colfamily list<br>
1258    * &nbsp;2.3 the default properties for both {@link HTableDescriptor} and {@link HColumnDescriptor}<br>
1259    * @throws IOException
1260    */
1261   public void fixOrphanTables() throws IOException {
1262     if (shouldFixTableOrphans() && !orphanTableDirs.isEmpty()) {
1263 
1264       List<TableName> tmpList = new ArrayList<TableName>();
1265       tmpList.addAll(orphanTableDirs.keySet());
1266       HTableDescriptor[] htds = getHTableDescriptors(tmpList);
1267       Iterator<Entry<TableName, Set<String>>> iter =
1268           orphanTableDirs.entrySet().iterator();
1269       int j = 0;
1270       int numFailedCase = 0;
1271       FSTableDescriptors fstd = new FSTableDescriptors(getConf());
1272       while (iter.hasNext()) {
1273         Entry<TableName, Set<String>> entry =
1274             iter.next();
1275         TableName tableName = entry.getKey();
1276         LOG.info("Trying to fix orphan table error: " + tableName);
1277         if (j < htds.length) {
1278           if (tableName.equals(htds[j].getTableName())) {
1279             HTableDescriptor htd = htds[j];
1280             LOG.info("fixing orphan table: " + tableName + " from cache");
1281             fstd.createTableDescriptor(new TableDescriptor(htd), true);
1282             j++;
1283             iter.remove();
1284           }
1285         } else {
1286           if (fabricateTableInfo(fstd, tableName, entry.getValue())) {
1287             LOG.warn("fixing orphan table: " + tableName + " with a default .tableinfo file");
1288             LOG.warn("Strongly recommend to modify the HTableDescriptor if necessary for: " + tableName);
1289             iter.remove();
1290           } else {
1291             LOG.error("Unable to create default .tableinfo for " + tableName + " while missing column family information");
1292             numFailedCase++;
1293           }
1294         }
1295         fixes++;
1296       }
1297 
1298       if (orphanTableDirs.isEmpty()) {
1299         // all orphanTableDirs are luckily recovered
1300         // re-run doFsck after recovering the .tableinfo file
1301         setShouldRerun();
1302         LOG.warn("Strongly recommend to re-run manually hfsck after all orphanTableDirs being fixed");
1303       } else if (numFailedCase > 0) {
1304         LOG.error("Failed to fix " + numFailedCase
1305             + " OrphanTables with default .tableinfo files");
1306       }
1307 
1308     }
1309     //cleanup the list
1310     orphanTableDirs.clear();
1311 
1312   }
1313 
1314   /**
1315    * This borrows code from MasterFileSystem.bootstrap(). Explicitly creates it's own WAL, so be
1316    * sure to close it as well as the region when you're finished.
1317    *
1318    * @return an open hbase:meta HRegion
1319    */
1320   private HRegion createNewMeta() throws IOException {
1321     Path rootdir = FSUtils.getRootDir(getConf());
1322     Configuration c = getConf();
1323     HRegionInfo metaHRI = new HRegionInfo(HRegionInfo.FIRST_META_REGIONINFO);
1324     HTableDescriptor metaDescriptor = new FSTableDescriptors(c).get(TableName.META_TABLE_NAME);
1325     MasterFileSystem.setInfoFamilyCachingForMeta(metaDescriptor, false);
1326     // The WAL subsystem will use the default rootDir rather than the passed in rootDir
1327     // unless I pass along via the conf.
1328     Configuration confForWAL = new Configuration(c);
1329     confForWAL.set(HConstants.HBASE_DIR, rootdir.toString());
1330     WAL wal = (new WALFactory(confForWAL,
1331         Collections.<WALActionsListener>singletonList(new MetricsWAL()),
1332         "hbck-meta-recovery-" + RandomStringUtils.randomNumeric(8))).
1333         getWAL(metaHRI.getEncodedNameAsBytes());
1334     HRegion meta = HRegion.createHRegion(metaHRI, rootdir, c, metaDescriptor, wal);
1335     MasterFileSystem.setInfoFamilyCachingForMeta(metaDescriptor, true);
1336     return meta;
1337   }
1338 
1339   /**
1340    * Generate set of puts to add to new meta.  This expects the tables to be
1341    * clean with no overlaps or holes.  If there are any problems it returns null.
1342    *
1343    * @return An array list of puts to do in bulk, null if tables have problems
1344    */
1345   private ArrayList<Put> generatePuts(
1346       SortedMap<TableName, TableInfo> tablesInfo) throws IOException {
1347     ArrayList<Put> puts = new ArrayList<Put>();
1348     boolean hasProblems = false;
1349     for (Entry<TableName, TableInfo> e : tablesInfo.entrySet()) {
1350       TableName name = e.getKey();
1351 
1352       // skip "hbase:meta"
1353       if (name.compareTo(TableName.META_TABLE_NAME) == 0) {
1354         continue;
1355       }
1356 
1357       TableInfo ti = e.getValue();
1358       puts.add(MetaTableAccessor
1359           .makePutFromTableState(new TableState(ti.tableName, TableState.State.ENABLED)));
1360       for (Entry<byte[], Collection<HbckInfo>> spl : ti.sc.getStarts().asMap()
1361           .entrySet()) {
1362         Collection<HbckInfo> his = spl.getValue();
1363         int sz = his.size();
1364         if (sz != 1) {
1365           // problem
1366           LOG.error("Split starting at " + Bytes.toStringBinary(spl.getKey())
1367               + " had " +  sz + " regions instead of exactly 1." );
1368           hasProblems = true;
1369           continue;
1370         }
1371 
1372         // add the row directly to meta.
1373         HbckInfo hi = his.iterator().next();
1374         HRegionInfo hri = hi.getHdfsHRI(); // hi.metaEntry;
1375         Put p = MetaTableAccessor.makePutFromRegionInfo(hri);
1376         puts.add(p);
1377       }
1378     }
1379     return hasProblems ? null : puts;
1380   }
1381 
1382   /**
1383    * Suggest fixes for each table
1384    */
1385   private void suggestFixes(
1386       SortedMap<TableName, TableInfo> tablesInfo) throws IOException {
1387     logParallelMerge();
1388     for (TableInfo tInfo : tablesInfo.values()) {
1389       TableIntegrityErrorHandler handler = tInfo.new IntegrityFixSuggester(tInfo, errors);
1390       tInfo.checkRegionChain(handler);
1391     }
1392   }
1393 
1394   /**
1395    * Rebuilds meta from information in hdfs/fs.  Depends on configuration settings passed into
1396    * hbck constructor to point to a particular fs/dir. Assumes HBase is OFFLINE.
1397    *
1398    * @param fix flag that determines if method should attempt to fix holes
1399    * @return true if successful, false if attempt failed.
1400    */
1401   public boolean rebuildMeta(boolean fix) throws IOException,
1402       InterruptedException {
1403 
1404     // TODO check to make sure hbase is offline. (or at least the table
1405     // currently being worked on is off line)
1406 
1407     // Determine what's on HDFS
1408     LOG.info("Loading HBase regioninfo from HDFS...");
1409     loadHdfsRegionDirs(); // populating regioninfo table.
1410 
1411     int errs = errors.getErrorList().size();
1412     tablesInfo = loadHdfsRegionInfos(); // update tableInfos based on region info in fs.
1413     checkHdfsIntegrity(false, false);
1414 
1415     // make sure ok.
1416     if (errors.getErrorList().size() != errs) {
1417       // While in error state, iterate until no more fixes possible
1418       while(true) {
1419         fixes = 0;
1420         suggestFixes(tablesInfo);
1421         errors.clear();
1422         loadHdfsRegionInfos(); // update tableInfos based on region info in fs.
1423         checkHdfsIntegrity(shouldFixHdfsHoles(), shouldFixHdfsOverlaps());
1424 
1425         int errCount = errors.getErrorList().size();
1426 
1427         if (fixes == 0) {
1428           if (errCount > 0) {
1429             return false; // failed to fix problems.
1430           } else {
1431             break; // no fixes and no problems? drop out and fix stuff!
1432           }
1433         }
1434       }
1435     }
1436 
1437     // we can rebuild, move old meta out of the way and start
1438     LOG.info("HDFS regioninfo's seems good.  Sidelining old hbase:meta");
1439     Path backupDir = sidelineOldMeta();
1440 
1441     LOG.info("Creating new hbase:meta");
1442     HRegion meta = createNewMeta();
1443 
1444     // populate meta
1445     List<Put> puts = generatePuts(tablesInfo);
1446     if (puts == null) {
1447       LOG.fatal("Problem encountered when creating new hbase:meta entries.  " +
1448         "You may need to restore the previously sidelined hbase:meta");
1449       return false;
1450     }
1451     meta.batchMutate(puts.toArray(new Put[puts.size()]));
1452     meta.close();
1453     if (meta.getWAL() != null) {
1454       meta.getWAL().close();
1455     }
1456     LOG.info("Success! hbase:meta table rebuilt.");
1457     LOG.info("Old hbase:meta is moved into " + backupDir);
1458     return true;
1459   }
1460 
1461   /**
1462    * Log an appropriate message about whether or not overlapping merges are computed in parallel.
1463    */
1464   private void logParallelMerge() {
1465     if (getConf().getBoolean("hbasefsck.overlap.merge.parallel", true)) {
1466       LOG.info("Handling overlap merges in parallel. set hbasefsck.overlap.merge.parallel to" +
1467           " false to run serially.");
1468     } else {
1469       LOG.info("Handling overlap merges serially.  set hbasefsck.overlap.merge.parallel to" +
1470           " true to run in parallel.");
1471     }
1472   }
1473 
1474   private SortedMap<TableName, TableInfo> checkHdfsIntegrity(boolean fixHoles,
1475       boolean fixOverlaps) throws IOException {
1476     LOG.info("Checking HBase region split map from HDFS data...");
1477     logParallelMerge();
1478     for (TableInfo tInfo : tablesInfo.values()) {
1479       TableIntegrityErrorHandler handler;
1480       if (fixHoles || fixOverlaps) {
1481         handler = tInfo.new HDFSIntegrityFixer(tInfo, errors, getConf(),
1482           fixHoles, fixOverlaps);
1483       } else {
1484         handler = tInfo.new IntegrityFixSuggester(tInfo, errors);
1485       }
1486       if (!tInfo.checkRegionChain(handler)) {
1487         // should dump info as well.
1488         errors.report("Found inconsistency in table " + tInfo.getName());
1489       }
1490     }
1491     return tablesInfo;
1492   }
1493 
1494   private Path getSidelineDir() throws IOException {
1495     if (sidelineDir == null) {
1496       Path hbaseDir = FSUtils.getRootDir(getConf());
1497       Path hbckDir = new Path(hbaseDir, HConstants.HBCK_SIDELINEDIR_NAME);
1498       sidelineDir = new Path(hbckDir, hbaseDir.getName() + "-"
1499           + startMillis);
1500     }
1501     return sidelineDir;
1502   }
1503 
1504   /**
1505    * Sideline a region dir (instead of deleting it)
1506    */
1507   Path sidelineRegionDir(FileSystem fs, HbckInfo hi) throws IOException {
1508     return sidelineRegionDir(fs, null, hi);
1509   }
1510 
1511   /**
1512    * Sideline a region dir (instead of deleting it)
1513    *
1514    * @param parentDir if specified, the region will be sidelined to
1515    * folder like .../parentDir/<table name>/<region name>. The purpose
1516    * is to group together similar regions sidelined, for example, those
1517    * regions should be bulk loaded back later on. If null, it is ignored.
1518    */
1519   Path sidelineRegionDir(FileSystem fs,
1520       String parentDir, HbckInfo hi) throws IOException {
1521     TableName tableName = hi.getTableName();
1522     Path regionDir = hi.getHdfsRegionDir();
1523 
1524     if (!fs.exists(regionDir)) {
1525       LOG.warn("No previous " + regionDir + " exists.  Continuing.");
1526       return null;
1527     }
1528 
1529     Path rootDir = getSidelineDir();
1530     if (parentDir != null) {
1531       rootDir = new Path(rootDir, parentDir);
1532     }
1533     Path sidelineTableDir= FSUtils.getTableDir(rootDir, tableName);
1534     Path sidelineRegionDir = new Path(sidelineTableDir, regionDir.getName());
1535     fs.mkdirs(sidelineRegionDir);
1536     boolean success = false;
1537     FileStatus[] cfs =  fs.listStatus(regionDir);
1538     if (cfs == null) {
1539       LOG.info("Region dir is empty: " + regionDir);
1540     } else {
1541       for (FileStatus cf : cfs) {
1542         Path src = cf.getPath();
1543         Path dst =  new Path(sidelineRegionDir, src.getName());
1544         if (fs.isFile(src)) {
1545           // simple file
1546           success = fs.rename(src, dst);
1547           if (!success) {
1548             String msg = "Unable to rename file " + src +  " to " + dst;
1549             LOG.error(msg);
1550             throw new IOException(msg);
1551           }
1552           continue;
1553         }
1554 
1555         // is a directory.
1556         fs.mkdirs(dst);
1557 
1558         LOG.info("Sidelining files from " + src + " into containing region " + dst);
1559         // FileSystem.rename is inconsistent with directories -- if the
1560         // dst (foo/a) exists and is a dir, and the src (foo/b) is a dir,
1561         // it moves the src into the dst dir resulting in (foo/a/b).  If
1562         // the dst does not exist, and the src a dir, src becomes dst. (foo/b)
1563         FileStatus[] hfiles = fs.listStatus(src);
1564         if (hfiles != null && hfiles.length > 0) {
1565           for (FileStatus hfile : hfiles) {
1566             success = fs.rename(hfile.getPath(), dst);
1567             if (!success) {
1568               String msg = "Unable to rename file " + src +  " to " + dst;
1569               LOG.error(msg);
1570               throw new IOException(msg);
1571             }
1572           }
1573         }
1574         LOG.debug("Sideline directory contents:");
1575         debugLsr(sidelineRegionDir);
1576       }
1577     }
1578 
1579     LOG.info("Removing old region dir: " + regionDir);
1580     success = fs.delete(regionDir, true);
1581     if (!success) {
1582       String msg = "Unable to delete dir " + regionDir;
1583       LOG.error(msg);
1584       throw new IOException(msg);
1585     }
1586     return sidelineRegionDir;
1587   }
1588 
1589   /**
1590    * Side line an entire table.
1591    */
1592   void sidelineTable(FileSystem fs, TableName tableName, Path hbaseDir,
1593       Path backupHbaseDir) throws IOException {
1594     Path tableDir = FSUtils.getTableDir(hbaseDir, tableName);
1595     if (fs.exists(tableDir)) {
1596       Path backupTableDir= FSUtils.getTableDir(backupHbaseDir, tableName);
1597       fs.mkdirs(backupTableDir.getParent());
1598       boolean success = fs.rename(tableDir, backupTableDir);
1599       if (!success) {
1600         throw new IOException("Failed to move  " + tableName + " from "
1601             +  tableDir + " to " + backupTableDir);
1602       }
1603     } else {
1604       LOG.info("No previous " + tableName +  " exists.  Continuing.");
1605     }
1606   }
1607 
1608   /**
1609    * @return Path to backup of original directory
1610    */
1611   Path sidelineOldMeta() throws IOException {
1612     // put current hbase:meta aside.
1613     Path hbaseDir = FSUtils.getRootDir(getConf());
1614     FileSystem fs = hbaseDir.getFileSystem(getConf());
1615     Path backupDir = getSidelineDir();
1616     fs.mkdirs(backupDir);
1617 
1618     try {
1619       sidelineTable(fs, TableName.META_TABLE_NAME, hbaseDir, backupDir);
1620     } catch (IOException e) {
1621         LOG.fatal("... failed to sideline meta. Currently in inconsistent state.  To restore "
1622             + "try to rename hbase:meta in " + backupDir.getName() + " to "
1623             + hbaseDir.getName() + ".", e);
1624       throw e; // throw original exception
1625     }
1626     return backupDir;
1627   }
1628 
1629   /**
1630    * Load the list of disabled tables in ZK into local set.
1631    * @throws ZooKeeperConnectionException
1632    * @throws IOException
1633    */
1634   private void loadTableStates()
1635   throws IOException {
1636     tableStates = MetaTableAccessor.getTableStates(connection);
1637   }
1638 
1639   /**
1640    * Check if the specified region's table is disabled.
1641    * @param tableName table to check status of
1642    */
1643   private boolean isTableDisabled(TableName tableName) {
1644     return tableStates.containsKey(tableName)
1645         && tableStates.get(tableName)
1646         .inStates(TableState.State.DISABLED, TableState.State.DISABLING);
1647   }
1648 
1649   /**
1650    * Scan HDFS for all regions, recording their information into
1651    * regionInfoMap
1652    */
1653   public void loadHdfsRegionDirs() throws IOException, InterruptedException {
1654     Path rootDir = FSUtils.getRootDir(getConf());
1655     FileSystem fs = rootDir.getFileSystem(getConf());
1656 
1657     // list all tables from HDFS
1658     List<FileStatus> tableDirs = Lists.newArrayList();
1659 
1660     boolean foundVersionFile = fs.exists(new Path(rootDir, HConstants.VERSION_FILE_NAME));
1661 
1662     List<Path> paths = FSUtils.getTableDirs(fs, rootDir);
1663     for (Path path : paths) {
1664       TableName tableName = FSUtils.getTableName(path);
1665        if ((!checkMetaOnly &&
1666            isTableIncluded(tableName)) ||
1667            tableName.equals(TableName.META_TABLE_NAME)) {
1668          tableDirs.add(fs.getFileStatus(path));
1669        }
1670     }
1671 
1672     // verify that version file exists
1673     if (!foundVersionFile) {
1674       errors.reportError(ERROR_CODE.NO_VERSION_FILE,
1675           "Version file does not exist in root dir " + rootDir);
1676       if (shouldFixVersionFile()) {
1677         LOG.info("Trying to create a new " + HConstants.VERSION_FILE_NAME
1678             + " file.");
1679         setShouldRerun();
1680         FSUtils.setVersion(fs, rootDir, getConf().getInt(
1681             HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000), getConf().getInt(
1682             HConstants.VERSION_FILE_WRITE_ATTEMPTS,
1683             HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS));
1684       }
1685     }
1686 
1687     // level 1:  <HBASE_DIR>/*
1688     List<WorkItemHdfsDir> dirs = new ArrayList<WorkItemHdfsDir>(tableDirs.size());
1689     List<Future<Void>> dirsFutures;
1690 
1691     for (FileStatus tableDir : tableDirs) {
1692       LOG.debug("Loading region dirs from " +tableDir.getPath());
1693       dirs.add(new WorkItemHdfsDir(this, fs, errors, tableDir));
1694     }
1695 
1696     // Invoke and wait for Callables to complete
1697     dirsFutures = executor.invokeAll(dirs);
1698 
1699     for(Future<Void> f: dirsFutures) {
1700       try {
1701         f.get();
1702       } catch(ExecutionException e) {
1703         LOG.warn("Could not load region dir " , e.getCause());
1704       }
1705     }
1706     errors.print("");
1707   }
1708 
1709   /**
1710    * Record the location of the hbase:meta region as found in ZooKeeper.
1711    */
1712   private boolean recordMetaRegion() throws IOException {
1713     RegionLocations rl = ((ClusterConnection)connection).locateRegion(TableName.META_TABLE_NAME,
1714         HConstants.EMPTY_START_ROW, false, false);
1715     if (rl == null) {
1716       errors.reportError(ERROR_CODE.NULL_META_REGION,
1717           "META region was not found in Zookeeper");
1718       return false;
1719     }
1720     for (HRegionLocation metaLocation : rl.getRegionLocations()) {
1721       // Check if Meta region is valid and existing
1722       if (metaLocation == null ) {
1723         errors.reportError(ERROR_CODE.NULL_META_REGION,
1724             "META region location is null");
1725         return false;
1726       }
1727       if (metaLocation.getRegionInfo() == null) {
1728         errors.reportError(ERROR_CODE.NULL_META_REGION,
1729             "META location regionInfo is null");
1730         return false;
1731       }
1732       if (metaLocation.getHostname() == null) {
1733         errors.reportError(ERROR_CODE.NULL_META_REGION,
1734             "META location hostName is null");
1735         return false;
1736       }
1737       ServerName sn = metaLocation.getServerName();
1738       MetaEntry m = new MetaEntry(metaLocation.getRegionInfo(), sn, EnvironmentEdgeManager.currentTime());
1739       HbckInfo hbckInfo = regionInfoMap.get(metaLocation.getRegionInfo().getEncodedName());
1740       if (hbckInfo == null) {
1741         regionInfoMap.put(metaLocation.getRegionInfo().getEncodedName(), new HbckInfo(m));
1742       } else {
1743         hbckInfo.metaEntry = m;
1744       }
1745     }
1746     return true;
1747   }
1748 
1749   private ZooKeeperWatcher createZooKeeperWatcher() throws IOException {
1750     return new ZooKeeperWatcher(getConf(), "hbase Fsck", new Abortable() {
1751       @Override
1752       public void abort(String why, Throwable e) {
1753         LOG.error(why, e);
1754         System.exit(1);
1755       }
1756 
1757       @Override
1758       public boolean isAborted() {
1759         return false;
1760       }
1761 
1762     });
1763   }
1764 
1765   private ServerName getMetaRegionServerName(int replicaId)
1766   throws IOException, KeeperException {
1767     ZooKeeperWatcher zkw = createZooKeeperWatcher();
1768     ServerName sn = null;
1769     try {
1770       sn = new MetaTableLocator().getMetaRegionLocation(zkw, replicaId);
1771     } finally {
1772       zkw.close();
1773     }
1774     return sn;
1775   }
1776 
1777   /**
1778    * Contacts each regionserver and fetches metadata about regions.
1779    * @param regionServerList - the list of region servers to connect to
1780    * @throws IOException if a remote or network exception occurs
1781    */
1782   void processRegionServers(Collection<ServerName> regionServerList)
1783     throws IOException, InterruptedException {
1784 
1785     List<WorkItemRegion> workItems = new ArrayList<WorkItemRegion>(regionServerList.size());
1786     List<Future<Void>> workFutures;
1787 
1788     // loop to contact each region server in parallel
1789     for (ServerName rsinfo: regionServerList) {
1790       workItems.add(new WorkItemRegion(this, rsinfo, errors, connection));
1791     }
1792 
1793     workFutures = executor.invokeAll(workItems);
1794 
1795     for(int i=0; i<workFutures.size(); i++) {
1796       WorkItemRegion item = workItems.get(i);
1797       Future<Void> f = workFutures.get(i);
1798       try {
1799         f.get();
1800       } catch(ExecutionException e) {
1801         LOG.warn("Could not process regionserver " + item.rsinfo.getHostAndPort(),
1802             e.getCause());
1803       }
1804     }
1805   }
1806 
1807   /**
1808    * Check consistency of all regions that have been found in previous phases.
1809    */
1810   private void checkAndFixConsistency()
1811   throws IOException, KeeperException, InterruptedException {
1812     // Divide the checks in two phases. One for default/primary replicas and another
1813     // for the non-primary ones. Keeps code cleaner this way.
1814 
1815     List<CheckRegionConsistencyWorkItem> workItems =
1816         new ArrayList<CheckRegionConsistencyWorkItem>(regionInfoMap.size());
1817     for (java.util.Map.Entry<String, HbckInfo> e: regionInfoMap.entrySet()) {
1818       if (e.getValue().getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
1819         workItems.add(new CheckRegionConsistencyWorkItem(e.getKey(), e.getValue()));
1820       }
1821     }
1822     checkRegionConsistencyConcurrently(workItems);
1823 
1824     boolean prevHdfsCheck = shouldCheckHdfs();
1825     setCheckHdfs(false); //replicas don't have any hdfs data
1826     // Run a pass over the replicas and fix any assignment issues that exist on the currently
1827     // deployed/undeployed replicas.
1828     List<CheckRegionConsistencyWorkItem> replicaWorkItems =
1829         new ArrayList<CheckRegionConsistencyWorkItem>(regionInfoMap.size());
1830     for (java.util.Map.Entry<String, HbckInfo> e: regionInfoMap.entrySet()) {
1831       if (e.getValue().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
1832         replicaWorkItems.add(new CheckRegionConsistencyWorkItem(e.getKey(), e.getValue()));
1833       }
1834     }
1835     checkRegionConsistencyConcurrently(replicaWorkItems);
1836     setCheckHdfs(prevHdfsCheck);
1837 
1838     // If some regions is skipped during checkRegionConsistencyConcurrently() phase, we might
1839     // not get accurate state of the hbase if continuing. The config here allows users to tune
1840     // the tolerance of number of skipped region.
1841     // TODO: evaluate the consequence to continue the hbck operation without config.
1842     int terminateThreshold =  getConf().getInt("hbase.hbck.skipped.regions.limit", 0);
1843     int numOfSkippedRegions = skippedRegions.size();
1844     if (numOfSkippedRegions > 0 && numOfSkippedRegions > terminateThreshold) {
1845       throw new IOException(numOfSkippedRegions
1846         + " region(s) could not be checked or repaired.  See logs for detail.");
1847     }
1848 
1849     if (shouldCheckHdfs()) {
1850       checkAndFixTableStates();
1851     }
1852   }
1853 
1854   /**
1855    * Check consistency of all regions using mulitple threads concurrently.
1856    */
1857   private void checkRegionConsistencyConcurrently(
1858     final List<CheckRegionConsistencyWorkItem> workItems)
1859     throws IOException, KeeperException, InterruptedException {
1860     if (workItems.isEmpty()) {
1861       return;  // nothing to check
1862     }
1863 
1864     List<Future<Void>> workFutures = executor.invokeAll(workItems);
1865     for(Future<Void> f: workFutures) {
1866       try {
1867         f.get();
1868       } catch(ExecutionException e1) {
1869         LOG.warn("Could not check region consistency " , e1.getCause());
1870         if (e1.getCause() instanceof IOException) {
1871           throw (IOException)e1.getCause();
1872         } else if (e1.getCause() instanceof KeeperException) {
1873           throw (KeeperException)e1.getCause();
1874         } else if (e1.getCause() instanceof InterruptedException) {
1875           throw (InterruptedException)e1.getCause();
1876         } else {
1877           throw new IOException(e1.getCause());
1878         }
1879       }
1880     }
1881   }
1882 
1883   class CheckRegionConsistencyWorkItem implements Callable<Void> {
1884     private final String key;
1885     private final HbckInfo hbi;
1886 
1887     CheckRegionConsistencyWorkItem(String key, HbckInfo hbi) {
1888       this.key = key;
1889       this.hbi = hbi;
1890     }
1891 
1892     @Override
1893     public synchronized Void call() throws Exception {
1894       try {
1895         checkRegionConsistency(key, hbi);
1896       } catch (Exception e) {
1897         // If the region is non-META region, skip this region and send warning/error message; if
1898         // the region is META region, we should not continue.
1899         LOG.warn("Unable to complete check or repair the region '" + hbi.getRegionNameAsString()
1900           + "'.", e);
1901         if (hbi.getHdfsHRI().isMetaRegion()) {
1902           throw e;
1903         }
1904         LOG.warn("Skip region '" + hbi.getRegionNameAsString() + "'");
1905         addSkippedRegion(hbi);
1906       }
1907       return null;
1908     }
1909   }
1910 
1911   private void addSkippedRegion(final HbckInfo hbi) {
1912     Set<String> skippedRegionNames = skippedRegions.get(hbi.getTableName());
1913     if (skippedRegionNames == null) {
1914       skippedRegionNames = new HashSet<String>();
1915     }
1916     skippedRegionNames.add(hbi.getRegionNameAsString());
1917     skippedRegions.put(hbi.getTableName(), skippedRegionNames);
1918   }
1919 
1920   /**
1921    * Check and fix table states, assumes full info available:
1922    * - tableInfos
1923    * - empty tables loaded
1924    */
1925   private void checkAndFixTableStates() throws IOException {
1926     // first check dangling states
1927     for (Entry<TableName, TableState> entry : tableStates.entrySet()) {
1928       TableName tableName = entry.getKey();
1929       TableState tableState = entry.getValue();
1930       TableInfo tableInfo = tablesInfo.get(tableName);
1931       if (isTableIncluded(tableName)
1932           && !tableName.isSystemTable()
1933           && tableInfo == null) {
1934         if (fixMeta) {
1935           MetaTableAccessor.deleteTableState(connection, tableName);
1936           TableState state = MetaTableAccessor.getTableState(connection, tableName);
1937           if (state != null) {
1938             errors.reportError(ERROR_CODE.ORPHAN_TABLE_STATE,
1939                 tableName + " unable to delete dangling table state " + tableState);
1940           }
1941         } else {
1942           errors.reportError(ERROR_CODE.ORPHAN_TABLE_STATE,
1943               tableName + " has dangling table state " + tableState);
1944         }
1945       }
1946     }
1947     // check that all tables have states
1948     for (TableName tableName : tablesInfo.keySet()) {
1949       if (isTableIncluded(tableName) && !tableStates.containsKey(tableName)) {
1950         if (fixMeta) {
1951           MetaTableAccessor.updateTableState(connection, tableName, TableState.State.ENABLED);
1952           TableState newState = MetaTableAccessor.getTableState(connection, tableName);
1953           if (newState == null) {
1954             errors.reportError(ERROR_CODE.NO_TABLE_STATE,
1955                 "Unable to change state for table " + tableName + " in meta ");
1956           }
1957         } else {
1958           errors.reportError(ERROR_CODE.NO_TABLE_STATE,
1959               tableName + " has no state in meta ");
1960         }
1961       }
1962     }
1963   }
1964 
1965   private void preCheckPermission() throws IOException, AccessDeniedException {
1966     if (shouldIgnorePreCheckPermission()) {
1967       return;
1968     }
1969 
1970     Path hbaseDir = FSUtils.getRootDir(getConf());
1971     FileSystem fs = hbaseDir.getFileSystem(getConf());
1972     UserProvider userProvider = UserProvider.instantiate(getConf());
1973     UserGroupInformation ugi = userProvider.getCurrent().getUGI();
1974     FileStatus[] files = fs.listStatus(hbaseDir);
1975     for (FileStatus file : files) {
1976       try {
1977         FSUtils.checkAccess(ugi, file, FsAction.WRITE);
1978       } catch (AccessDeniedException ace) {
1979         LOG.warn("Got AccessDeniedException when preCheckPermission ", ace);
1980         errors.reportError(ERROR_CODE.WRONG_USAGE, "Current user " + ugi.getUserName()
1981           + " does not have write perms to " + file.getPath()
1982           + ". Please rerun hbck as hdfs user " + file.getOwner());
1983         throw ace;
1984       }
1985     }
1986   }
1987 
1988   /**
1989    * Deletes region from meta table
1990    */
1991   private void deleteMetaRegion(HbckInfo hi) throws IOException {
1992     deleteMetaRegion(hi.metaEntry.getRegionName());
1993   }
1994 
1995   /**
1996    * Deletes region from meta table
1997    */
1998   private void deleteMetaRegion(byte[] metaKey) throws IOException {
1999     Delete d = new Delete(metaKey);
2000     meta.delete(d);
2001     LOG.info("Deleted " + Bytes.toString(metaKey) + " from META" );
2002   }
2003 
2004   /**
2005    * Reset the split parent region info in meta table
2006    */
2007   private void resetSplitParent(HbckInfo hi) throws IOException {
2008     RowMutations mutations = new RowMutations(hi.metaEntry.getRegionName());
2009     Delete d = new Delete(hi.metaEntry.getRegionName());
2010     d.deleteColumn(HConstants.CATALOG_FAMILY, HConstants.SPLITA_QUALIFIER);
2011     d.deleteColumn(HConstants.CATALOG_FAMILY, HConstants.SPLITB_QUALIFIER);
2012     mutations.add(d);
2013 
2014     HRegionInfo hri = new HRegionInfo(hi.metaEntry);
2015     hri.setOffline(false);
2016     hri.setSplit(false);
2017     Put p = MetaTableAccessor.makePutFromRegionInfo(hri);
2018     mutations.add(p);
2019 
2020     meta.mutateRow(mutations);
2021     LOG.info("Reset split parent " + hi.metaEntry.getRegionNameAsString() + " in META" );
2022   }
2023 
2024   /**
2025    * This backwards-compatibility wrapper for permanently offlining a region
2026    * that should not be alive.  If the region server does not support the
2027    * "offline" method, it will use the closest unassign method instead.  This
2028    * will basically work until one attempts to disable or delete the affected
2029    * table.  The problem has to do with in-memory only master state, so
2030    * restarting the HMaster or failing over to another should fix this.
2031    */
2032   private void offline(byte[] regionName) throws IOException {
2033     String regionString = Bytes.toStringBinary(regionName);
2034     if (!rsSupportsOffline) {
2035       LOG.warn("Using unassign region " + regionString
2036           + " instead of using offline method, you should"
2037           + " restart HMaster after these repairs");
2038       admin.unassign(regionName, true);
2039       return;
2040     }
2041 
2042     // first time we assume the rs's supports #offline.
2043     try {
2044       LOG.info("Offlining region " + regionString);
2045       admin.offline(regionName);
2046     } catch (IOException ioe) {
2047       String notFoundMsg = "java.lang.NoSuchMethodException: " +
2048         "org.apache.hadoop.hbase.master.HMaster.offline([B)";
2049       if (ioe.getMessage().contains(notFoundMsg)) {
2050         LOG.warn("Using unassign region " + regionString
2051             + " instead of using offline method, you should"
2052             + " restart HMaster after these repairs");
2053         rsSupportsOffline = false; // in the future just use unassign
2054         admin.unassign(regionName, true);
2055         return;
2056       }
2057       throw ioe;
2058     }
2059   }
2060 
2061   private void undeployRegions(HbckInfo hi) throws IOException, InterruptedException {
2062     undeployRegionsForHbi(hi);
2063     // undeploy replicas of the region (but only if the method is invoked for the primary)
2064     if (hi.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
2065       return;
2066     }
2067     int numReplicas = admin.getTableDescriptor(hi.getTableName()).getRegionReplication();
2068     for (int i = 1; i < numReplicas; i++) {
2069       if (hi.getPrimaryHRIForDeployedReplica() == null) continue;
2070       HRegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(
2071           hi.getPrimaryHRIForDeployedReplica(), i);
2072       HbckInfo h = regionInfoMap.get(hri.getEncodedName());
2073       if (h != null) {
2074         undeployRegionsForHbi(h);
2075         //set skip checks; we undeployed it, and we don't want to evaluate this anymore
2076         //in consistency checks
2077         h.setSkipChecks(true);
2078       }
2079     }
2080   }
2081 
2082   private void undeployRegionsForHbi(HbckInfo hi) throws IOException, InterruptedException {
2083     for (OnlineEntry rse : hi.deployedEntries) {
2084       LOG.debug("Undeploy region "  + rse.hri + " from " + rse.hsa);
2085       try {
2086         HBaseFsckRepair.closeRegionSilentlyAndWait(connection, rse.hsa, rse.hri);
2087         offline(rse.hri.getRegionName());
2088       } catch (IOException ioe) {
2089         LOG.warn("Got exception when attempting to offline region "
2090             + Bytes.toString(rse.hri.getRegionName()), ioe);
2091       }
2092     }
2093   }
2094 
2095   /**
2096    * Attempts to undeploy a region from a region server based in information in
2097    * META.  Any operations that modify the file system should make sure that
2098    * its corresponding region is not deployed to prevent data races.
2099    *
2100    * A separate call is required to update the master in-memory region state
2101    * kept in the AssignementManager.  Because disable uses this state instead of
2102    * that found in META, we can't seem to cleanly disable/delete tables that
2103    * have been hbck fixed.  When used on a version of HBase that does not have
2104    * the offline ipc call exposed on the master (<0.90.5, <0.92.0) a master
2105    * restart or failover may be required.
2106    */
2107   private void closeRegion(HbckInfo hi) throws IOException, InterruptedException {
2108     if (hi.metaEntry == null && hi.hdfsEntry == null) {
2109       undeployRegions(hi);
2110       return;
2111     }
2112 
2113     // get assignment info and hregioninfo from meta.
2114     Get get = new Get(hi.getRegionName());
2115     get.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
2116     get.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER);
2117     get.addColumn(HConstants.CATALOG_FAMILY, HConstants.STARTCODE_QUALIFIER);
2118     // also get the locations of the replicas to close if the primary region is being closed
2119     if (hi.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
2120       int numReplicas = admin.getTableDescriptor(hi.getTableName()).getRegionReplication();
2121       for (int i = 0; i < numReplicas; i++) {
2122         get.addColumn(HConstants.CATALOG_FAMILY, MetaTableAccessor.getServerColumn(i));
2123         get.addColumn(HConstants.CATALOG_FAMILY, MetaTableAccessor.getStartCodeColumn(i));
2124       }
2125     }
2126     Result r = meta.get(get);
2127     RegionLocations rl = MetaTableAccessor.getRegionLocations(r);
2128     if (rl == null) {
2129       LOG.warn("Unable to close region " + hi.getRegionNameAsString() +
2130           " since meta does not have handle to reach it");
2131       return;
2132     }
2133     for (HRegionLocation h : rl.getRegionLocations()) {
2134       ServerName serverName = h.getServerName();
2135       if (serverName == null) {
2136         errors.reportError("Unable to close region "
2137             + hi.getRegionNameAsString() +  " because meta does not "
2138             + "have handle to reach it.");
2139         continue;
2140       }
2141       HRegionInfo hri = h.getRegionInfo();
2142       if (hri == null) {
2143         LOG.warn("Unable to close region " + hi.getRegionNameAsString()
2144             + " because hbase:meta had invalid or missing "
2145             + HConstants.CATALOG_FAMILY_STR + ":"
2146             + Bytes.toString(HConstants.REGIONINFO_QUALIFIER)
2147             + " qualifier value.");
2148         continue;
2149       }
2150       // close the region -- close files and remove assignment
2151       HBaseFsckRepair.closeRegionSilentlyAndWait(connection, serverName, hri);
2152     }
2153   }
2154 
2155   private void tryAssignmentRepair(HbckInfo hbi, String msg) throws IOException,
2156     KeeperException, InterruptedException {
2157     // If we are trying to fix the errors
2158     if (shouldFixAssignments()) {
2159       errors.print(msg);
2160       undeployRegions(hbi);
2161       setShouldRerun();
2162       HRegionInfo hri = hbi.getHdfsHRI();
2163       if (hri == null) {
2164         hri = hbi.metaEntry;
2165       }
2166       HBaseFsckRepair.fixUnassigned(admin, hri);
2167       HBaseFsckRepair.waitUntilAssigned(admin, hri);
2168 
2169       // also assign replicas if needed (do it only when this call operates on a primary replica)
2170       if (hbi.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) return;
2171       int replicationCount = admin.getTableDescriptor(hri.getTable()).getRegionReplication();
2172       for (int i = 1; i < replicationCount; i++) {
2173         hri = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
2174         HbckInfo h = regionInfoMap.get(hri.getEncodedName());
2175         if (h != null) {
2176           undeployRegions(h);
2177           //set skip checks; we undeploy & deploy it; we don't want to evaluate this hbi anymore
2178           //in consistency checks
2179           h.setSkipChecks(true);
2180         }
2181         HBaseFsckRepair.fixUnassigned(admin, hri);
2182         HBaseFsckRepair.waitUntilAssigned(admin, hri);
2183       }
2184 
2185     }
2186   }
2187 
2188   /**
2189    * Check a single region for consistency and correct deployment.
2190    */
2191   private void checkRegionConsistency(final String key, final HbckInfo hbi)
2192   throws IOException, KeeperException, InterruptedException {
2193 
2194     if (hbi.isSkipChecks()) return;
2195     String descriptiveName = hbi.toString();
2196     boolean inMeta = hbi.metaEntry != null;
2197     // In case not checking HDFS, assume the region is on HDFS
2198     boolean inHdfs = !shouldCheckHdfs() || hbi.getHdfsRegionDir() != null;
2199     boolean hasMetaAssignment = inMeta && hbi.metaEntry.regionServer != null;
2200     boolean isDeployed = !hbi.deployedOn.isEmpty();
2201     boolean isMultiplyDeployed = hbi.deployedOn.size() > 1;
2202     boolean deploymentMatchesMeta =
2203       hasMetaAssignment && isDeployed && !isMultiplyDeployed &&
2204       hbi.metaEntry.regionServer.equals(hbi.deployedOn.get(0));
2205     boolean splitParent =
2206         inMeta && hbi.metaEntry.isSplit() && hbi.metaEntry.isOffline();
2207     boolean shouldBeDeployed = inMeta && !isTableDisabled(hbi.metaEntry.getTable());
2208     boolean recentlyModified = inHdfs &&
2209       hbi.getModTime() + timelag > EnvironmentEdgeManager.currentTime();
2210 
2211     // ========== First the healthy cases =============
2212     if (hbi.containsOnlyHdfsEdits()) {
2213       return;
2214     }
2215     if (inMeta && inHdfs && isDeployed && deploymentMatchesMeta && shouldBeDeployed) {
2216       return;
2217     } else if (inMeta && inHdfs && !shouldBeDeployed && !isDeployed) {
2218       LOG.info("Region " + descriptiveName + " is in META, and in a disabled " +
2219         "tabled that is not deployed");
2220       return;
2221     } else if (recentlyModified) {
2222       LOG.warn("Region " + descriptiveName + " was recently modified -- skipping");
2223       return;
2224     }
2225     // ========== Cases where the region is not in hbase:meta =============
2226     else if (!inMeta && !inHdfs && !isDeployed) {
2227       // We shouldn't have record of this region at all then!
2228       assert false : "Entry for region with no data";
2229     } else if (!inMeta && !inHdfs && isDeployed) {
2230       errors.reportError(ERROR_CODE.NOT_IN_META_HDFS, "Region "
2231           + descriptiveName + ", key=" + key + ", not on HDFS or in hbase:meta but " +
2232           "deployed on " + Joiner.on(", ").join(hbi.deployedOn));
2233       if (shouldFixAssignments()) {
2234         undeployRegions(hbi);
2235       }
2236 
2237     } else if (!inMeta && inHdfs && !isDeployed) {
2238       if (hbi.isMerged()) {
2239         // This region has already been merged, the remaining hdfs file will be
2240         // cleaned by CatalogJanitor later
2241         hbi.setSkipChecks(true);
2242         LOG.info("Region " + descriptiveName
2243             + " got merge recently, its file(s) will be cleaned by CatalogJanitor later");
2244         return;
2245       }
2246       errors.reportError(ERROR_CODE.NOT_IN_META_OR_DEPLOYED, "Region "
2247           + descriptiveName + " on HDFS, but not listed in hbase:meta " +
2248           "or deployed on any region server");
2249       // restore region consistency of an adopted orphan
2250       if (shouldFixMeta()) {
2251         if (!hbi.isHdfsRegioninfoPresent()) {
2252           LOG.error("Region " + hbi.getHdfsHRI() + " could have been repaired"
2253               +  " in table integrity repair phase if -fixHdfsOrphans was" +
2254               " used.");
2255           return;
2256         }
2257 
2258         HRegionInfo hri = hbi.getHdfsHRI();
2259         TableInfo tableInfo = tablesInfo.get(hri.getTable());
2260 
2261         for (HRegionInfo region : tableInfo.getRegionsFromMeta()) {
2262           if (Bytes.compareTo(region.getStartKey(), hri.getStartKey()) <= 0
2263               && (region.getEndKey().length == 0 || Bytes.compareTo(region.getEndKey(),
2264                 hri.getEndKey()) >= 0)
2265               && Bytes.compareTo(region.getStartKey(), hri.getEndKey()) <= 0) {
2266             if(region.isSplit() || region.isOffline()) continue;
2267             Path regionDir = hbi.getHdfsRegionDir();
2268             FileSystem fs = regionDir.getFileSystem(getConf());
2269             List<Path> familyDirs = FSUtils.getFamilyDirs(fs, regionDir);
2270             for (Path familyDir : familyDirs) {
2271               List<Path> referenceFilePaths = FSUtils.getReferenceFilePaths(fs, familyDir);
2272               for (Path referenceFilePath : referenceFilePaths) {
2273                 Path parentRegionDir =
2274                     StoreFileInfo.getReferredToFile(referenceFilePath).getParent().getParent();
2275                 if (parentRegionDir.toString().endsWith(region.getEncodedName())) {
2276                   LOG.warn(hri + " start and stop keys are in the range of " + region
2277                       + ". The region might not be cleaned up from hdfs when region " + region
2278                       + " split failed. Hence deleting from hdfs.");
2279                   HRegionFileSystem.deleteRegionFromFileSystem(getConf(), fs,
2280                     regionDir.getParent(), hri);
2281                   return;
2282                 }
2283               }
2284             }
2285           }
2286         }
2287         LOG.info("Patching hbase:meta with .regioninfo: " + hbi.getHdfsHRI());
2288         int numReplicas = admin.getTableDescriptor(hbi.getTableName()).getRegionReplication();
2289         HBaseFsckRepair.fixMetaHoleOnlineAndAddReplicas(getConf(), hbi.getHdfsHRI(),
2290             admin.getClusterStatus().getServers(), numReplicas);
2291 
2292         tryAssignmentRepair(hbi, "Trying to reassign region...");
2293       }
2294 
2295     } else if (!inMeta && inHdfs && isDeployed) {
2296       errors.reportError(ERROR_CODE.NOT_IN_META, "Region " + descriptiveName
2297           + " not in META, but deployed on " + Joiner.on(", ").join(hbi.deployedOn));
2298       debugLsr(hbi.getHdfsRegionDir());
2299       if (hbi.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
2300         // for replicas, this means that we should undeploy the region (we would have
2301         // gone over the primaries and fixed meta holes in first phase under
2302         // checkAndFixConsistency; we shouldn't get the condition !inMeta at
2303         // this stage unless unwanted replica)
2304         if (shouldFixAssignments()) {
2305           undeployRegionsForHbi(hbi);
2306         }
2307       }
2308       if (shouldFixMeta() && hbi.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
2309         if (!hbi.isHdfsRegioninfoPresent()) {
2310           LOG.error("This should have been repaired in table integrity repair phase");
2311           return;
2312         }
2313 
2314         LOG.info("Patching hbase:meta with with .regioninfo: " + hbi.getHdfsHRI());
2315         int numReplicas = admin.getTableDescriptor(hbi.getTableName()).getRegionReplication();
2316         HBaseFsckRepair.fixMetaHoleOnlineAndAddReplicas(getConf(), hbi.getHdfsHRI(),
2317             admin.getClusterStatus().getServers(), numReplicas);
2318         tryAssignmentRepair(hbi, "Trying to fix unassigned region...");
2319       }
2320 
2321     // ========== Cases where the region is in hbase:meta =============
2322     } else if (inMeta && inHdfs && !isDeployed && splitParent) {
2323       // check whether this is an actual error, or just transient state where parent
2324       // is not cleaned
2325       if (hbi.metaEntry.splitA != null && hbi.metaEntry.splitB != null) {
2326         // check that split daughters are there
2327         HbckInfo infoA = this.regionInfoMap.get(hbi.metaEntry.splitA.getEncodedName());
2328         HbckInfo infoB = this.regionInfoMap.get(hbi.metaEntry.splitB.getEncodedName());
2329         if (infoA != null && infoB != null) {
2330           // we already processed or will process daughters. Move on, nothing to see here.
2331           hbi.setSkipChecks(true);
2332           return;
2333         }
2334       }
2335       errors.reportError(ERROR_CODE.LINGERING_SPLIT_PARENT, "Region "
2336           + descriptiveName + " is a split parent in META, in HDFS, "
2337           + "and not deployed on any region server. This could be transient.");
2338       if (shouldFixSplitParents()) {
2339         setShouldRerun();
2340         resetSplitParent(hbi);
2341       }
2342     } else if (inMeta && !inHdfs && !isDeployed) {
2343       errors.reportError(ERROR_CODE.NOT_IN_HDFS_OR_DEPLOYED, "Region "
2344           + descriptiveName + " found in META, but not in HDFS "
2345           + "or deployed on any region server.");
2346       if (shouldFixMeta()) {
2347         deleteMetaRegion(hbi);
2348       }
2349     } else if (inMeta && !inHdfs && isDeployed) {
2350       errors.reportError(ERROR_CODE.NOT_IN_HDFS, "Region " + descriptiveName
2351           + " found in META, but not in HDFS, " +
2352           "and deployed on " + Joiner.on(", ").join(hbi.deployedOn));
2353       // We treat HDFS as ground truth.  Any information in meta is transient
2354       // and equivalent data can be regenerated.  So, lets unassign and remove
2355       // these problems from META.
2356       if (shouldFixAssignments()) {
2357         errors.print("Trying to fix unassigned region...");
2358         undeployRegions(hbi);
2359       }
2360       if (shouldFixMeta()) {
2361         // wait for it to complete
2362         deleteMetaRegion(hbi);
2363       }
2364     } else if (inMeta && inHdfs && !isDeployed && shouldBeDeployed) {
2365       errors.reportError(ERROR_CODE.NOT_DEPLOYED, "Region " + descriptiveName
2366           + " not deployed on any region server.");
2367       tryAssignmentRepair(hbi, "Trying to fix unassigned region...");
2368     } else if (inMeta && inHdfs && isDeployed && !shouldBeDeployed) {
2369       errors.reportError(ERROR_CODE.SHOULD_NOT_BE_DEPLOYED,
2370           "Region " + descriptiveName + " should not be deployed according " +
2371           "to META, but is deployed on " + Joiner.on(", ").join(hbi.deployedOn));
2372       if (shouldFixAssignments()) {
2373         errors.print("Trying to close the region " + descriptiveName);
2374         setShouldRerun();
2375         HBaseFsckRepair.fixMultiAssignment(connection, hbi.metaEntry, hbi.deployedOn);
2376       }
2377     } else if (inMeta && inHdfs && isMultiplyDeployed) {
2378       errors.reportError(ERROR_CODE.MULTI_DEPLOYED, "Region " + descriptiveName
2379           + " is listed in hbase:meta on region server " + hbi.metaEntry.regionServer
2380           + " but is multiply assigned to region servers " +
2381           Joiner.on(", ").join(hbi.deployedOn));
2382       // If we are trying to fix the errors
2383       if (shouldFixAssignments()) {
2384         errors.print("Trying to fix assignment error...");
2385         setShouldRerun();
2386         HBaseFsckRepair.fixMultiAssignment(connection, hbi.metaEntry, hbi.deployedOn);
2387       }
2388     } else if (inMeta && inHdfs && isDeployed && !deploymentMatchesMeta) {
2389       errors.reportError(ERROR_CODE.SERVER_DOES_NOT_MATCH_META, "Region "
2390           + descriptiveName + " listed in hbase:meta on region server " +
2391           hbi.metaEntry.regionServer + " but found on region server " +
2392           hbi.deployedOn.get(0));
2393       // If we are trying to fix the errors
2394       if (shouldFixAssignments()) {
2395         errors.print("Trying to fix assignment error...");
2396         setShouldRerun();
2397         HBaseFsckRepair.fixMultiAssignment(connection, hbi.metaEntry, hbi.deployedOn);
2398         HBaseFsckRepair.waitUntilAssigned(admin, hbi.getHdfsHRI());
2399       }
2400     } else {
2401       errors.reportError(ERROR_CODE.UNKNOWN, "Region " + descriptiveName +
2402           " is in an unforeseen state:" +
2403           " inMeta=" + inMeta +
2404           " inHdfs=" + inHdfs +
2405           " isDeployed=" + isDeployed +
2406           " isMultiplyDeployed=" + isMultiplyDeployed +
2407           " deploymentMatchesMeta=" + deploymentMatchesMeta +
2408           " shouldBeDeployed=" + shouldBeDeployed);
2409     }
2410   }
2411 
2412   /**
2413    * Checks tables integrity. Goes over all regions and scans the tables.
2414    * Collects all the pieces for each table and checks if there are missing,
2415    * repeated or overlapping ones.
2416    * @throws IOException
2417    */
2418   SortedMap<TableName, TableInfo> checkIntegrity() throws IOException {
2419     tablesInfo = new TreeMap<TableName,TableInfo> ();
2420     LOG.debug("There are " + regionInfoMap.size() + " region info entries");
2421     for (HbckInfo hbi : regionInfoMap.values()) {
2422       // Check only valid, working regions
2423       if (hbi.metaEntry == null) {
2424         // this assumes that consistency check has run loadMetaEntry
2425         Path p = hbi.getHdfsRegionDir();
2426         if (p == null) {
2427           errors.report("No regioninfo in Meta or HDFS. " + hbi);
2428         }
2429 
2430         // TODO test.
2431         continue;
2432       }
2433       if (hbi.metaEntry.regionServer == null) {
2434         errors.detail("Skipping region because no region server: " + hbi);
2435         continue;
2436       }
2437       if (hbi.metaEntry.isOffline()) {
2438         errors.detail("Skipping region because it is offline: " + hbi);
2439         continue;
2440       }
2441       if (hbi.containsOnlyHdfsEdits()) {
2442         errors.detail("Skipping region because it only contains edits" + hbi);
2443         continue;
2444       }
2445 
2446       // Missing regionDir or over-deployment is checked elsewhere. Include
2447       // these cases in modTInfo, so we can evaluate those regions as part of
2448       // the region chain in META
2449       //if (hbi.foundRegionDir == null) continue;
2450       //if (hbi.deployedOn.size() != 1) continue;
2451       if (hbi.deployedOn.size() == 0) continue;
2452 
2453       // We should be safe here
2454       TableName tableName = hbi.metaEntry.getTable();
2455       TableInfo modTInfo = tablesInfo.get(tableName);
2456       if (modTInfo == null) {
2457         modTInfo = new TableInfo(tableName);
2458       }
2459       for (ServerName server : hbi.deployedOn) {
2460         modTInfo.addServer(server);
2461       }
2462 
2463       if (!hbi.isSkipChecks()) {
2464         modTInfo.addRegionInfo(hbi);
2465       }
2466 
2467       tablesInfo.put(tableName, modTInfo);
2468     }
2469 
2470     loadTableInfosForTablesWithNoRegion();
2471 
2472     logParallelMerge();
2473     for (TableInfo tInfo : tablesInfo.values()) {
2474       TableIntegrityErrorHandler handler = tInfo.new IntegrityFixSuggester(tInfo, errors);
2475       if (!tInfo.checkRegionChain(handler)) {
2476         errors.report("Found inconsistency in table " + tInfo.getName());
2477       }
2478     }
2479     return tablesInfo;
2480   }
2481 
2482   /** Loads table info's for tables that may not have been included, since there are no
2483    * regions reported for the table, but table dir is there in hdfs
2484    */
2485   private void loadTableInfosForTablesWithNoRegion() throws IOException {
2486     Map<String, HTableDescriptor> allTables = new FSTableDescriptors(getConf()).getAll();
2487     for (HTableDescriptor htd : allTables.values()) {
2488       if (checkMetaOnly && !htd.isMetaTable()) {
2489         continue;
2490       }
2491 
2492       TableName tableName = htd.getTableName();
2493       if (isTableIncluded(tableName) && !tablesInfo.containsKey(tableName)) {
2494         TableInfo tableInfo = new TableInfo(tableName);
2495         tableInfo.htds.add(htd);
2496         tablesInfo.put(htd.getTableName(), tableInfo);
2497       }
2498     }
2499   }
2500 
2501   /**
2502    * Merge hdfs data by moving from contained HbckInfo into targetRegionDir.
2503    * @return number of file move fixes done to merge regions.
2504    */
2505   public int mergeRegionDirs(Path targetRegionDir, HbckInfo contained) throws IOException {
2506     int fileMoves = 0;
2507     String thread = Thread.currentThread().getName();
2508     LOG.debug("[" + thread + "] Contained region dir after close and pause");
2509     debugLsr(contained.getHdfsRegionDir());
2510 
2511     // rename the contained into the container.
2512     FileSystem fs = targetRegionDir.getFileSystem(getConf());
2513     FileStatus[] dirs = null;
2514     try {
2515       dirs = fs.listStatus(contained.getHdfsRegionDir());
2516     } catch (FileNotFoundException fnfe) {
2517       // region we are attempting to merge in is not present!  Since this is a merge, there is
2518       // no harm skipping this region if it does not exist.
2519       if (!fs.exists(contained.getHdfsRegionDir())) {
2520         LOG.warn("[" + thread + "] HDFS region dir " + contained.getHdfsRegionDir()
2521             + " is missing. Assuming already sidelined or moved.");
2522       } else {
2523         sidelineRegionDir(fs, contained);
2524       }
2525       return fileMoves;
2526     }
2527 
2528     if (dirs == null) {
2529       if (!fs.exists(contained.getHdfsRegionDir())) {
2530         LOG.warn("[" + thread + "] HDFS region dir " + contained.getHdfsRegionDir()
2531             + " already sidelined.");
2532       } else {
2533         sidelineRegionDir(fs, contained);
2534       }
2535       return fileMoves;
2536     }
2537 
2538     for (FileStatus cf : dirs) {
2539       Path src = cf.getPath();
2540       Path dst =  new Path(targetRegionDir, src.getName());
2541 
2542       if (src.getName().equals(HRegionFileSystem.REGION_INFO_FILE)) {
2543         // do not copy the old .regioninfo file.
2544         continue;
2545       }
2546 
2547       if (src.getName().equals(HConstants.HREGION_OLDLOGDIR_NAME)) {
2548         // do not copy the .oldlogs files
2549         continue;
2550       }
2551 
2552       LOG.info("[" + thread + "] Moving files from " + src + " into containing region " + dst);
2553       // FileSystem.rename is inconsistent with directories -- if the
2554       // dst (foo/a) exists and is a dir, and the src (foo/b) is a dir,
2555       // it moves the src into the dst dir resulting in (foo/a/b).  If
2556       // the dst does not exist, and the src a dir, src becomes dst. (foo/b)
2557       for (FileStatus hfile : fs.listStatus(src)) {
2558         boolean success = fs.rename(hfile.getPath(), dst);
2559         if (success) {
2560           fileMoves++;
2561         }
2562       }
2563       LOG.debug("[" + thread + "] Sideline directory contents:");
2564       debugLsr(targetRegionDir);
2565     }
2566 
2567     // if all success.
2568     sidelineRegionDir(fs, contained);
2569     LOG.info("[" + thread + "] Sidelined region dir "+ contained.getHdfsRegionDir() + " into " +
2570         getSidelineDir());
2571     debugLsr(contained.getHdfsRegionDir());
2572 
2573     return fileMoves;
2574   }
2575 
2576 
2577   static class WorkItemOverlapMerge implements Callable<Void> {
2578     private TableIntegrityErrorHandler handler;
2579     Collection<HbckInfo> overlapgroup;
2580 
2581     WorkItemOverlapMerge(Collection<HbckInfo> overlapgroup, TableIntegrityErrorHandler handler) {
2582       this.handler = handler;
2583       this.overlapgroup = overlapgroup;
2584     }
2585 
2586     @Override
2587     public Void call() throws Exception {
2588       handler.handleOverlapGroup(overlapgroup);
2589       return null;
2590     }
2591   };
2592 
2593 
2594   /**
2595    * Maintain information about a particular table.
2596    */
2597   public class TableInfo {
2598     TableName tableName;
2599     TreeSet <ServerName> deployedOn;
2600 
2601     // backwards regions
2602     final List<HbckInfo> backwards = new ArrayList<HbckInfo>();
2603 
2604     // sidelined big overlapped regions
2605     final Map<Path, HbckInfo> sidelinedRegions = new HashMap<Path, HbckInfo>();
2606 
2607     // region split calculator
2608     final RegionSplitCalculator<HbckInfo> sc = new RegionSplitCalculator<HbckInfo>(cmp);
2609 
2610     // Histogram of different HTableDescriptors found.  Ideally there is only one!
2611     final Set<HTableDescriptor> htds = new HashSet<HTableDescriptor>();
2612 
2613     // key = start split, values = set of splits in problem group
2614     final Multimap<byte[], HbckInfo> overlapGroups =
2615       TreeMultimap.create(RegionSplitCalculator.BYTES_COMPARATOR, cmp);
2616 
2617     // list of regions derived from meta entries.
2618     private ImmutableList<HRegionInfo> regionsFromMeta = null;
2619 
2620     TableInfo(TableName name) {
2621       this.tableName = name;
2622       deployedOn = new TreeSet <ServerName>();
2623     }
2624 
2625     /**
2626      * @return descriptor common to all regions.  null if are none or multiple!
2627      */
2628     private HTableDescriptor getHTD() {
2629       if (htds.size() == 1) {
2630         return (HTableDescriptor)htds.toArray()[0];
2631       } else {
2632         LOG.error("None/Multiple table descriptors found for table '"
2633           + tableName + "' regions: " + htds);
2634       }
2635       return null;
2636     }
2637 
2638     public void addRegionInfo(HbckInfo hir) {
2639       if (Bytes.equals(hir.getEndKey(), HConstants.EMPTY_END_ROW)) {
2640         // end key is absolute end key, just add it.
2641         // ignore replicas other than primary for these checks
2642         if (hir.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) sc.add(hir);
2643         return;
2644       }
2645 
2646       // if not the absolute end key, check for cycle
2647       if (Bytes.compareTo(hir.getStartKey(), hir.getEndKey()) > 0) {
2648         errors.reportError(
2649             ERROR_CODE.REGION_CYCLE,
2650             String.format("The endkey for this region comes before the "
2651                 + "startkey, startkey=%s, endkey=%s",
2652                 Bytes.toStringBinary(hir.getStartKey()),
2653                 Bytes.toStringBinary(hir.getEndKey())), this, hir);
2654         backwards.add(hir);
2655         return;
2656       }
2657 
2658       // main case, add to split calculator
2659       // ignore replicas other than primary for these checks
2660       if (hir.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) sc.add(hir);
2661     }
2662 
2663     public void addServer(ServerName server) {
2664       this.deployedOn.add(server);
2665     }
2666 
2667     public TableName getName() {
2668       return tableName;
2669     }
2670 
2671     public int getNumRegions() {
2672       return sc.getStarts().size() + backwards.size();
2673     }
2674 
2675     public synchronized ImmutableList<HRegionInfo> getRegionsFromMeta() {
2676       // lazy loaded, synchronized to ensure a single load
2677       if (regionsFromMeta == null) {
2678         List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
2679         for (HbckInfo h : HBaseFsck.this.regionInfoMap.values()) {
2680           if (tableName.equals(h.getTableName())) {
2681             if (h.metaEntry != null) {
2682               regions.add((HRegionInfo) h.metaEntry);
2683             }
2684           }
2685         }
2686         regionsFromMeta = Ordering.natural().immutableSortedCopy(regions);
2687       }
2688       
2689       return regionsFromMeta;
2690     }
2691     
2692     private class IntegrityFixSuggester extends TableIntegrityErrorHandlerImpl {
2693       ErrorReporter errors;
2694 
2695       IntegrityFixSuggester(TableInfo ti, ErrorReporter errors) {
2696         this.errors = errors;
2697         setTableInfo(ti);
2698       }
2699 
2700       @Override
2701       public void handleRegionStartKeyNotEmpty(HbckInfo hi) throws IOException{
2702         errors.reportError(ERROR_CODE.FIRST_REGION_STARTKEY_NOT_EMPTY,
2703             "First region should start with an empty key.  You need to "
2704             + " create a new region and regioninfo in HDFS to plug the hole.",
2705             getTableInfo(), hi);
2706       }
2707 
2708       @Override
2709       public void handleRegionEndKeyNotEmpty(byte[] curEndKey) throws IOException {
2710         errors.reportError(ERROR_CODE.LAST_REGION_ENDKEY_NOT_EMPTY,
2711             "Last region should end with an empty key. You need to "
2712                 + "create a new region and regioninfo in HDFS to plug the hole.", getTableInfo());
2713       }
2714 
2715       @Override
2716       public void handleDegenerateRegion(HbckInfo hi) throws IOException{
2717         errors.reportError(ERROR_CODE.DEGENERATE_REGION,
2718             "Region has the same start and end key.", getTableInfo(), hi);
2719       }
2720 
2721       @Override
2722       public void handleDuplicateStartKeys(HbckInfo r1, HbckInfo r2) throws IOException{
2723         byte[] key = r1.getStartKey();
2724         // dup start key
2725         errors.reportError(ERROR_CODE.DUPE_STARTKEYS,
2726             "Multiple regions have the same startkey: "
2727             + Bytes.toStringBinary(key), getTableInfo(), r1);
2728         errors.reportError(ERROR_CODE.DUPE_STARTKEYS,
2729             "Multiple regions have the same startkey: "
2730             + Bytes.toStringBinary(key), getTableInfo(), r2);
2731       }
2732 
2733       @Override
2734       public void handleOverlapInRegionChain(HbckInfo hi1, HbckInfo hi2) throws IOException{
2735         errors.reportError(ERROR_CODE.OVERLAP_IN_REGION_CHAIN,
2736             "There is an overlap in the region chain.",
2737             getTableInfo(), hi1, hi2);
2738       }
2739 
2740       @Override
2741       public void handleHoleInRegionChain(byte[] holeStart, byte[] holeStop) throws IOException{
2742         errors.reportError(
2743             ERROR_CODE.HOLE_IN_REGION_CHAIN,
2744             "There is a hole in the region chain between "
2745                 + Bytes.toStringBinary(holeStart) + " and "
2746                 + Bytes.toStringBinary(holeStop)
2747                 + ".  You need to create a new .regioninfo and region "
2748                 + "dir in hdfs to plug the hole.");
2749       }
2750     };
2751 
2752     /**
2753      * This handler fixes integrity errors from hdfs information.  There are
2754      * basically three classes of integrity problems 1) holes, 2) overlaps, and
2755      * 3) invalid regions.
2756      *
2757      * This class overrides methods that fix holes and the overlap group case.
2758      * Individual cases of particular overlaps are handled by the general
2759      * overlap group merge repair case.
2760      *
2761      * If hbase is online, this forces regions offline before doing merge
2762      * operations.
2763      */
2764     private class HDFSIntegrityFixer extends IntegrityFixSuggester {
2765       Configuration conf;
2766 
2767       boolean fixOverlaps = true;
2768 
2769       HDFSIntegrityFixer(TableInfo ti, ErrorReporter errors, Configuration conf,
2770           boolean fixHoles, boolean fixOverlaps) {
2771         super(ti, errors);
2772         this.conf = conf;
2773         this.fixOverlaps = fixOverlaps;
2774         // TODO properly use fixHoles
2775       }
2776 
2777       /**
2778        * This is a special case hole -- when the first region of a table is
2779        * missing from META, HBase doesn't acknowledge the existance of the
2780        * table.
2781        */
2782       @Override
2783       public void handleRegionStartKeyNotEmpty(HbckInfo next) throws IOException {
2784         errors.reportError(ERROR_CODE.FIRST_REGION_STARTKEY_NOT_EMPTY,
2785             "First region should start with an empty key.  Creating a new " +
2786             "region and regioninfo in HDFS to plug the hole.",
2787             getTableInfo(), next);
2788         HTableDescriptor htd = getTableInfo().getHTD();
2789         // from special EMPTY_START_ROW to next region's startKey
2790         HRegionInfo newRegion = new HRegionInfo(htd.getTableName(),
2791             HConstants.EMPTY_START_ROW, next.getStartKey());
2792 
2793         // TODO test
2794         HRegion region = HBaseFsckRepair.createHDFSRegionDir(conf, newRegion, htd);
2795         LOG.info("Table region start key was not empty.  Created new empty region: "
2796             + newRegion + " " +region);
2797         fixes++;
2798       }
2799 
2800       @Override
2801       public void handleRegionEndKeyNotEmpty(byte[] curEndKey) throws IOException {
2802         errors.reportError(ERROR_CODE.LAST_REGION_ENDKEY_NOT_EMPTY,
2803             "Last region should end with an empty key.  Creating a new "
2804                 + "region and regioninfo in HDFS to plug the hole.", getTableInfo());
2805         HTableDescriptor htd = getTableInfo().getHTD();
2806         // from curEndKey to EMPTY_START_ROW
2807         HRegionInfo newRegion = new HRegionInfo(htd.getTableName(), curEndKey,
2808             HConstants.EMPTY_START_ROW);
2809 
2810         HRegion region = HBaseFsckRepair.createHDFSRegionDir(conf, newRegion, htd);
2811         LOG.info("Table region end key was not empty.  Created new empty region: " + newRegion
2812             + " " + region);
2813         fixes++;
2814       }
2815 
2816       /**
2817        * There is a hole in the hdfs regions that violates the table integrity
2818        * rules.  Create a new empty region that patches the hole.
2819        */
2820       @Override
2821       public void handleHoleInRegionChain(byte[] holeStartKey, byte[] holeStopKey) throws IOException {
2822         errors.reportError(
2823             ERROR_CODE.HOLE_IN_REGION_CHAIN,
2824             "There is a hole in the region chain between "
2825                 + Bytes.toStringBinary(holeStartKey) + " and "
2826                 + Bytes.toStringBinary(holeStopKey)
2827                 + ".  Creating a new regioninfo and region "
2828                 + "dir in hdfs to plug the hole.");
2829         HTableDescriptor htd = getTableInfo().getHTD();
2830         HRegionInfo newRegion = new HRegionInfo(htd.getTableName(), holeStartKey, holeStopKey);
2831         HRegion region = HBaseFsckRepair.createHDFSRegionDir(conf, newRegion, htd);
2832         LOG.info("Plugged hole by creating new empty region: "+ newRegion + " " +region);
2833         fixes++;
2834       }
2835 
2836       /**
2837        * This takes set of overlapping regions and merges them into a single
2838        * region.  This covers cases like degenerate regions, shared start key,
2839        * general overlaps, duplicate ranges, and partial overlapping regions.
2840        *
2841        * Cases:
2842        * - Clean regions that overlap
2843        * - Only .oldlogs regions (can't find start/stop range, or figure out)
2844        *
2845        * This is basically threadsafe, except for the fixer increment in mergeOverlaps.
2846        */
2847       @Override
2848       public void handleOverlapGroup(Collection<HbckInfo> overlap)
2849           throws IOException {
2850         Preconditions.checkNotNull(overlap);
2851         Preconditions.checkArgument(overlap.size() >0);
2852 
2853         if (!this.fixOverlaps) {
2854           LOG.warn("Not attempting to repair overlaps.");
2855           return;
2856         }
2857 
2858         if (overlap.size() > maxMerge) {
2859           LOG.warn("Overlap group has " + overlap.size() + " overlapping " +
2860             "regions which is greater than " + maxMerge + ", the max number of regions to merge");
2861           if (sidelineBigOverlaps) {
2862             // we only sideline big overlapped groups that exceeds the max number of regions to merge
2863             sidelineBigOverlaps(overlap);
2864           }
2865           return;
2866         }
2867 
2868         mergeOverlaps(overlap);
2869       }
2870 
2871       void mergeOverlaps(Collection<HbckInfo> overlap)
2872           throws IOException {
2873         String thread = Thread.currentThread().getName();
2874         LOG.info("== [" + thread + "] Merging regions into one region: "
2875           + Joiner.on(",").join(overlap));
2876         // get the min / max range and close all concerned regions
2877         Pair<byte[], byte[]> range = null;
2878         for (HbckInfo hi : overlap) {
2879           if (range == null) {
2880             range = new Pair<byte[], byte[]>(hi.getStartKey(), hi.getEndKey());
2881           } else {
2882             if (RegionSplitCalculator.BYTES_COMPARATOR
2883                 .compare(hi.getStartKey(), range.getFirst()) < 0) {
2884               range.setFirst(hi.getStartKey());
2885             }
2886             if (RegionSplitCalculator.BYTES_COMPARATOR
2887                 .compare(hi.getEndKey(), range.getSecond()) > 0) {
2888               range.setSecond(hi.getEndKey());
2889             }
2890           }
2891           // need to close files so delete can happen.
2892           LOG.debug("[" + thread + "] Closing region before moving data around: " +  hi);
2893           LOG.debug("[" + thread + "] Contained region dir before close");
2894           debugLsr(hi.getHdfsRegionDir());
2895           try {
2896             LOG.info("[" + thread + "] Closing region: " + hi);
2897             closeRegion(hi);
2898           } catch (IOException ioe) {
2899             LOG.warn("[" + thread + "] Was unable to close region " + hi
2900               + ".  Just continuing... ", ioe);
2901           } catch (InterruptedException e) {
2902             LOG.warn("[" + thread + "] Was unable to close region " + hi
2903               + ".  Just continuing... ", e);
2904           }
2905 
2906           try {
2907             LOG.info("[" + thread + "] Offlining region: " + hi);
2908             offline(hi.getRegionName());
2909           } catch (IOException ioe) {
2910             LOG.warn("[" + thread + "] Unable to offline region from master: " + hi
2911               + ".  Just continuing... ", ioe);
2912           }
2913         }
2914 
2915         // create new empty container region.
2916         HTableDescriptor htd = getTableInfo().getHTD();
2917         // from start key to end Key
2918         HRegionInfo newRegion = new HRegionInfo(htd.getTableName(), range.getFirst(),
2919             range.getSecond());
2920         HRegion region = HBaseFsckRepair.createHDFSRegionDir(conf, newRegion, htd);
2921         LOG.info("[" + thread + "] Created new empty container region: " +
2922             newRegion + " to contain regions: " + Joiner.on(",").join(overlap));
2923         debugLsr(region.getRegionFileSystem().getRegionDir());
2924 
2925         // all target regions are closed, should be able to safely cleanup.
2926         boolean didFix= false;
2927         Path target = region.getRegionFileSystem().getRegionDir();
2928         for (HbckInfo contained : overlap) {
2929           LOG.info("[" + thread + "] Merging " + contained  + " into " + target );
2930           int merges = mergeRegionDirs(target, contained);
2931           if (merges > 0) {
2932             didFix = true;
2933           }
2934         }
2935         if (didFix) {
2936           fixes++;
2937         }
2938       }
2939 
2940       /**
2941        * Sideline some regions in a big overlap group so that it
2942        * will have fewer regions, and it is easier to merge them later on.
2943        *
2944        * @param bigOverlap the overlapped group with regions more than maxMerge
2945        * @throws IOException
2946        */
2947       void sidelineBigOverlaps(
2948           Collection<HbckInfo> bigOverlap) throws IOException {
2949         int overlapsToSideline = bigOverlap.size() - maxMerge;
2950         if (overlapsToSideline > maxOverlapsToSideline) {
2951           overlapsToSideline = maxOverlapsToSideline;
2952         }
2953         List<HbckInfo> regionsToSideline =
2954           RegionSplitCalculator.findBigRanges(bigOverlap, overlapsToSideline);
2955         FileSystem fs = FileSystem.get(conf);
2956         for (HbckInfo regionToSideline: regionsToSideline) {
2957           try {
2958             LOG.info("Closing region: " + regionToSideline);
2959             closeRegion(regionToSideline);
2960           } catch (IOException ioe) {
2961             LOG.warn("Was unable to close region " + regionToSideline
2962               + ".  Just continuing... ", ioe);
2963           } catch (InterruptedException e) {
2964             LOG.warn("Was unable to close region " + regionToSideline
2965               + ".  Just continuing... ", e);
2966           }
2967 
2968           try {
2969             LOG.info("Offlining region: " + regionToSideline);
2970             offline(regionToSideline.getRegionName());
2971           } catch (IOException ioe) {
2972             LOG.warn("Unable to offline region from master: " + regionToSideline
2973               + ".  Just continuing... ", ioe);
2974           }
2975 
2976           LOG.info("Before sideline big overlapped region: " + regionToSideline.toString());
2977           Path sidelineRegionDir = sidelineRegionDir(fs, TO_BE_LOADED, regionToSideline);
2978           if (sidelineRegionDir != null) {
2979             sidelinedRegions.put(sidelineRegionDir, regionToSideline);
2980             LOG.info("After sidelined big overlapped region: "
2981               + regionToSideline.getRegionNameAsString()
2982               + " to " + sidelineRegionDir.toString());
2983             fixes++;
2984           }
2985         }
2986       }
2987     }
2988 
2989     /**
2990      * Check the region chain (from META) of this table.  We are looking for
2991      * holes, overlaps, and cycles.
2992      * @return false if there are errors
2993      * @throws IOException
2994      */
2995     public boolean checkRegionChain(TableIntegrityErrorHandler handler) throws IOException {
2996       // When table is disabled no need to check for the region chain. Some of the regions
2997       // accidently if deployed, this below code might report some issues like missing start
2998       // or end regions or region hole in chain and may try to fix which is unwanted.
2999       if (isTableDisabled(this.tableName)) {
3000         return true;
3001       }
3002       int originalErrorsCount = errors.getErrorList().size();
3003       Multimap<byte[], HbckInfo> regions = sc.calcCoverage();
3004       SortedSet<byte[]> splits = sc.getSplits();
3005 
3006       byte[] prevKey = null;
3007       byte[] problemKey = null;
3008 
3009       if (splits.size() == 0) {
3010         // no region for this table
3011         handler.handleHoleInRegionChain(HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW);
3012       }
3013 
3014       for (byte[] key : splits) {
3015         Collection<HbckInfo> ranges = regions.get(key);
3016         if (prevKey == null && !Bytes.equals(key, HConstants.EMPTY_BYTE_ARRAY)) {
3017           for (HbckInfo rng : ranges) {
3018             handler.handleRegionStartKeyNotEmpty(rng);
3019           }
3020         }
3021 
3022         // check for degenerate ranges
3023         for (HbckInfo rng : ranges) {
3024           // special endkey case converts '' to null
3025           byte[] endKey = rng.getEndKey();
3026           endKey = (endKey.length == 0) ? null : endKey;
3027           if (Bytes.equals(rng.getStartKey(),endKey)) {
3028             handler.handleDegenerateRegion(rng);
3029           }
3030         }
3031 
3032         if (ranges.size() == 1) {
3033           // this split key is ok -- no overlap, not a hole.
3034           if (problemKey != null) {
3035             LOG.warn("reached end of problem group: " + Bytes.toStringBinary(key));
3036           }
3037           problemKey = null; // fell through, no more problem.
3038         } else if (ranges.size() > 1) {
3039           // set the new problem key group name, if already have problem key, just
3040           // keep using it.
3041           if (problemKey == null) {
3042             // only for overlap regions.
3043             LOG.warn("Naming new problem group: " + Bytes.toStringBinary(key));
3044             problemKey = key;
3045           }
3046           overlapGroups.putAll(problemKey, ranges);
3047 
3048           // record errors
3049           ArrayList<HbckInfo> subRange = new ArrayList<HbckInfo>(ranges);
3050           //  this dumb and n^2 but this shouldn't happen often
3051           for (HbckInfo r1 : ranges) {
3052             if (r1.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) continue;
3053             subRange.remove(r1);
3054             for (HbckInfo r2 : subRange) {
3055               if (r2.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) continue;
3056               if (Bytes.compareTo(r1.getStartKey(), r2.getStartKey())==0) {
3057                 handler.handleDuplicateStartKeys(r1,r2);
3058               } else {
3059                 // overlap
3060                 handler.handleOverlapInRegionChain(r1, r2);
3061               }
3062             }
3063           }
3064 
3065         } else if (ranges.size() == 0) {
3066           if (problemKey != null) {
3067             LOG.warn("reached end of problem group: " + Bytes.toStringBinary(key));
3068           }
3069           problemKey = null;
3070 
3071           byte[] holeStopKey = sc.getSplits().higher(key);
3072           // if higher key is null we reached the top.
3073           if (holeStopKey != null) {
3074             // hole
3075             handler.handleHoleInRegionChain(key, holeStopKey);
3076           }
3077         }
3078         prevKey = key;
3079       }
3080 
3081       // When the last region of a table is proper and having an empty end key, 'prevKey'
3082       // will be null.
3083       if (prevKey != null) {
3084         handler.handleRegionEndKeyNotEmpty(prevKey);
3085       }
3086 
3087       // TODO fold this into the TableIntegrityHandler
3088       if (getConf().getBoolean("hbasefsck.overlap.merge.parallel", true)) {
3089         boolean ok = handleOverlapsParallel(handler, prevKey);
3090         if (!ok) {
3091           return false;
3092         }
3093       } else {
3094         for (Collection<HbckInfo> overlap : overlapGroups.asMap().values()) {
3095           handler.handleOverlapGroup(overlap);
3096         }
3097       }
3098 
3099       if (details) {
3100         // do full region split map dump
3101         errors.print("---- Table '"  +  this.tableName
3102             + "': region split map");
3103         dump(splits, regions);
3104         errors.print("---- Table '"  +  this.tableName
3105             + "': overlap groups");
3106         dumpOverlapProblems(overlapGroups);
3107         errors.print("There are " + overlapGroups.keySet().size()
3108             + " overlap groups with " + overlapGroups.size()
3109             + " overlapping regions");
3110       }
3111       if (!sidelinedRegions.isEmpty()) {
3112         LOG.warn("Sidelined big overlapped regions, please bulk load them!");
3113         errors.print("---- Table '"  +  this.tableName
3114             + "': sidelined big overlapped regions");
3115         dumpSidelinedRegions(sidelinedRegions);
3116       }
3117       return errors.getErrorList().size() == originalErrorsCount;
3118     }
3119 
3120     private boolean handleOverlapsParallel(TableIntegrityErrorHandler handler, byte[] prevKey)
3121         throws IOException {
3122       // we parallelize overlap handler for the case we have lots of groups to fix.  We can
3123       // safely assume each group is independent.
3124       List<WorkItemOverlapMerge> merges = new ArrayList<WorkItemOverlapMerge>(overlapGroups.size());
3125       List<Future<Void>> rets;
3126       for (Collection<HbckInfo> overlap : overlapGroups.asMap().values()) {
3127         //
3128         merges.add(new WorkItemOverlapMerge(overlap, handler));
3129       }
3130       try {
3131         rets = executor.invokeAll(merges);
3132       } catch (InterruptedException e) {
3133         LOG.error("Overlap merges were interrupted", e);
3134         return false;
3135       }
3136       for(int i=0; i<merges.size(); i++) {
3137         WorkItemOverlapMerge work = merges.get(i);
3138         Future<Void> f = rets.get(i);
3139         try {
3140           f.get();
3141         } catch(ExecutionException e) {
3142           LOG.warn("Failed to merge overlap group" + work, e.getCause());
3143         } catch (InterruptedException e) {
3144           LOG.error("Waiting for overlap merges was interrupted", e);
3145           return false;
3146         }
3147       }
3148       return true;
3149     }
3150 
3151     /**
3152      * This dumps data in a visually reasonable way for visual debugging
3153      *
3154      * @param splits
3155      * @param regions
3156      */
3157     void dump(SortedSet<byte[]> splits, Multimap<byte[], HbckInfo> regions) {
3158       // we display this way because the last end key should be displayed as well.
3159       StringBuilder sb = new StringBuilder();
3160       for (byte[] k : splits) {
3161         sb.setLength(0); // clear out existing buffer, if any.
3162         sb.append(Bytes.toStringBinary(k) + ":\t");
3163         for (HbckInfo r : regions.get(k)) {
3164           sb.append("[ "+ r.toString() + ", "
3165               + Bytes.toStringBinary(r.getEndKey())+ "]\t");
3166         }
3167         errors.print(sb.toString());
3168       }
3169     }
3170   }
3171 
3172   public void dumpOverlapProblems(Multimap<byte[], HbckInfo> regions) {
3173     // we display this way because the last end key should be displayed as
3174     // well.
3175     for (byte[] k : regions.keySet()) {
3176       errors.print(Bytes.toStringBinary(k) + ":");
3177       for (HbckInfo r : regions.get(k)) {
3178         errors.print("[ " + r.toString() + ", "
3179             + Bytes.toStringBinary(r.getEndKey()) + "]");
3180       }
3181       errors.print("----");
3182     }
3183   }
3184 
3185   public void dumpSidelinedRegions(Map<Path, HbckInfo> regions) {
3186     for (Map.Entry<Path, HbckInfo> entry: regions.entrySet()) {
3187       TableName tableName = entry.getValue().getTableName();
3188       Path path = entry.getKey();
3189       errors.print("This sidelined region dir should be bulk loaded: "
3190         + path.toString());
3191       errors.print("Bulk load command looks like: "
3192         + "hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles "
3193         + path.toUri().getPath() + " "+ tableName);
3194     }
3195   }
3196 
3197   public Multimap<byte[], HbckInfo> getOverlapGroups(
3198       TableName table) {
3199     TableInfo ti = tablesInfo.get(table);
3200     return ti.overlapGroups;
3201   }
3202 
3203   /**
3204    * Return a list of user-space table names whose metadata have not been
3205    * modified in the last few milliseconds specified by timelag
3206    * if any of the REGIONINFO_QUALIFIER, SERVER_QUALIFIER, STARTCODE_QUALIFIER,
3207    * SPLITA_QUALIFIER, SPLITB_QUALIFIER have not changed in the last
3208    * milliseconds specified by timelag, then the table is a candidate to be returned.
3209    * @return tables that have not been modified recently
3210    * @throws IOException if an error is encountered
3211    */
3212   HTableDescriptor[] getTables(AtomicInteger numSkipped) {
3213     List<TableName> tableNames = new ArrayList<TableName>();
3214     long now = EnvironmentEdgeManager.currentTime();
3215 
3216     for (HbckInfo hbi : regionInfoMap.values()) {
3217       MetaEntry info = hbi.metaEntry;
3218 
3219       // if the start key is zero, then we have found the first region of a table.
3220       // pick only those tables that were not modified in the last few milliseconds.
3221       if (info != null && info.getStartKey().length == 0 && !info.isMetaRegion()) {
3222         if (info.modTime + timelag < now) {
3223           tableNames.add(info.getTable());
3224         } else {
3225           numSkipped.incrementAndGet(); // one more in-flux table
3226         }
3227       }
3228     }
3229     return getHTableDescriptors(tableNames);
3230   }
3231 
3232   HTableDescriptor[] getHTableDescriptors(List<TableName> tableNames) {
3233     HTableDescriptor[] htd = new HTableDescriptor[0];
3234       LOG.info("getHTableDescriptors == tableNames => " + tableNames);
3235     try (Connection conn = ConnectionFactory.createConnection(getConf());
3236         Admin admin = conn.getAdmin()) {
3237       htd = admin.getTableDescriptorsByTableName(tableNames);
3238     } catch (IOException e) {
3239       LOG.debug("Exception getting table descriptors", e);
3240     }
3241     return htd;
3242   }
3243 
3244   /**
3245    * Gets the entry in regionInfo corresponding to the the given encoded
3246    * region name. If the region has not been seen yet, a new entry is added
3247    * and returned.
3248    */
3249   private synchronized HbckInfo getOrCreateInfo(String name) {
3250     HbckInfo hbi = regionInfoMap.get(name);
3251     if (hbi == null) {
3252       hbi = new HbckInfo(null);
3253       regionInfoMap.put(name, hbi);
3254     }
3255     return hbi;
3256   }
3257 
3258   private void checkAndFixTableLocks() throws IOException {
3259     TableLockChecker checker = new TableLockChecker(createZooKeeperWatcher(), errors);
3260     checker.checkTableLocks();
3261 
3262     if (this.fixTableLocks) {
3263       checker.fixExpiredTableLocks();
3264     }
3265   }
3266 
3267   /**
3268     * Check values in regionInfo for hbase:meta
3269     * Check if zero or more than one regions with hbase:meta are found.
3270     * If there are inconsistencies (i.e. zero or more than one regions
3271     * pretend to be holding the hbase:meta) try to fix that and report an error.
3272     * @throws IOException from HBaseFsckRepair functions
3273     * @throws KeeperException
3274     * @throws InterruptedException
3275     */
3276   boolean checkMetaRegion() throws IOException, KeeperException, InterruptedException {
3277     Map<Integer, HbckInfo> metaRegions = new HashMap<Integer, HbckInfo>();
3278     for (HbckInfo value : regionInfoMap.values()) {
3279       if (value.metaEntry != null && value.metaEntry.isMetaRegion()) {
3280         metaRegions.put(value.getReplicaId(), value);
3281       }
3282     }
3283     int metaReplication = admin.getTableDescriptor(TableName.META_TABLE_NAME)
3284         .getRegionReplication();
3285     boolean noProblem = true;
3286     // There will be always entries in regionInfoMap corresponding to hbase:meta & its replicas
3287     // Check the deployed servers. It should be exactly one server for each replica.
3288     for (int i = 0; i < metaReplication; i++) {
3289       HbckInfo metaHbckInfo = metaRegions.remove(i);
3290       List<ServerName> servers = new ArrayList<ServerName>();
3291       if (metaHbckInfo != null) {
3292         servers = metaHbckInfo.deployedOn;
3293       }
3294       if (servers.size() != 1) {
3295         noProblem = false;
3296         if (servers.size() == 0) {
3297           assignMetaReplica(i);
3298         } else if (servers.size() > 1) {
3299           errors
3300           .reportError(ERROR_CODE.MULTI_META_REGION, "hbase:meta, replicaId " +
3301                        metaHbckInfo.getReplicaId() + " is found on more than one region.");
3302           if (shouldFixAssignments()) {
3303             errors.print("Trying to fix a problem with hbase:meta, replicaId " +
3304                          metaHbckInfo.getReplicaId() +"..");
3305             setShouldRerun();
3306             // try fix it (treat is a dupe assignment)
3307             HBaseFsckRepair.fixMultiAssignment(connection, metaHbckInfo.metaEntry, servers);
3308           }
3309         }
3310       }
3311     }
3312     // unassign whatever is remaining in metaRegions. They are excess replicas.
3313     for (Map.Entry<Integer, HbckInfo> entry : metaRegions.entrySet()) {
3314       noProblem = false;
3315       errors.reportError(ERROR_CODE.SHOULD_NOT_BE_DEPLOYED,
3316           "hbase:meta replicas are deployed in excess. Configured " + metaReplication +
3317           ", deployed " + metaRegions.size());
3318       if (shouldFixAssignments()) {
3319         errors.print("Trying to undeploy excess replica, replicaId: " + entry.getKey() +
3320             " of hbase:meta..");
3321         setShouldRerun();
3322         unassignMetaReplica(entry.getValue());
3323       }
3324     }
3325     // if noProblem is false, rerun hbck with hopefully fixed META
3326     // if noProblem is true, no errors, so continue normally
3327     return noProblem;
3328   }
3329 
3330   private void unassignMetaReplica(HbckInfo hi) throws IOException, InterruptedException,
3331   KeeperException {
3332     undeployRegions(hi);
3333     ZooKeeperWatcher zkw = createZooKeeperWatcher();
3334     try {
3335       ZKUtil.deleteNode(zkw, zkw.getZNodeForReplica(hi.metaEntry.getReplicaId()));
3336     } finally {
3337       zkw.close();
3338     }
3339   }
3340 
3341   private void assignMetaReplica(int replicaId)
3342       throws IOException, KeeperException, InterruptedException {
3343     errors.reportError(ERROR_CODE.NO_META_REGION, "hbase:meta, replicaId " +
3344         replicaId +" is not found on any region.");
3345     if (shouldFixAssignments()) {
3346       errors.print("Trying to fix a problem with hbase:meta..");
3347       setShouldRerun();
3348       // try to fix it (treat it as unassigned region)
3349       HRegionInfo h = RegionReplicaUtil.getRegionInfoForReplica(
3350           HRegionInfo.FIRST_META_REGIONINFO, replicaId);
3351       HBaseFsckRepair.fixUnassigned(admin, h);
3352       HBaseFsckRepair.waitUntilAssigned(admin, h);
3353     }
3354   }
3355 
3356   /**
3357    * Scan hbase:meta, adding all regions found to the regionInfo map.
3358    * @throws IOException if an error is encountered
3359    */
3360   boolean loadMetaEntries() throws IOException {
3361     MetaTableAccessor.Visitor visitor = new MetaTableAccessor.Visitor() {
3362       int countRecord = 1;
3363 
3364       // comparator to sort KeyValues with latest modtime
3365       final Comparator<Cell> comp = new Comparator<Cell>() {
3366         @Override
3367         public int compare(Cell k1, Cell k2) {
3368           return (int)(k1.getTimestamp() - k2.getTimestamp());
3369         }
3370       };
3371 
3372       @Override
3373       public boolean visit(Result result) throws IOException {
3374         try {
3375 
3376           // record the latest modification of this META record
3377           long ts =  Collections.max(result.listCells(), comp).getTimestamp();
3378           RegionLocations rl = MetaTableAccessor.getRegionLocations(result);
3379           if (rl == null) {
3380             emptyRegionInfoQualifiers.add(result);
3381             errors.reportError(ERROR_CODE.EMPTY_META_CELL,
3382               "Empty REGIONINFO_QUALIFIER found in hbase:meta");
3383             return true;
3384           }
3385           ServerName sn = null;
3386           if (rl.getRegionLocation(HRegionInfo.DEFAULT_REPLICA_ID) == null ||
3387               rl.getRegionLocation(HRegionInfo.DEFAULT_REPLICA_ID).getRegionInfo() == null) {
3388             emptyRegionInfoQualifiers.add(result);
3389             errors.reportError(ERROR_CODE.EMPTY_META_CELL,
3390               "Empty REGIONINFO_QUALIFIER found in hbase:meta");
3391             return true;
3392           }
3393           HRegionInfo hri = rl.getRegionLocation(HRegionInfo.DEFAULT_REPLICA_ID).getRegionInfo();
3394           if (!(isTableIncluded(hri.getTable())
3395               || hri.isMetaRegion())) {
3396             return true;
3397           }
3398           PairOfSameType<HRegionInfo> daughters = HRegionInfo.getDaughterRegions(result);
3399           for (HRegionLocation h : rl.getRegionLocations()) {
3400             if (h == null || h.getRegionInfo() == null) {
3401               continue;
3402             }
3403             sn = h.getServerName();
3404             hri = h.getRegionInfo();
3405 
3406             MetaEntry m = null;
3407             if (hri.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
3408               m = new MetaEntry(hri, sn, ts, daughters.getFirst(), daughters.getSecond());
3409             } else {
3410               m = new MetaEntry(hri, sn, ts, null, null);
3411             }
3412             HbckInfo previous = regionInfoMap.get(hri.getEncodedName());
3413             if (previous == null) {
3414               regionInfoMap.put(hri.getEncodedName(), new HbckInfo(m));
3415             } else if (previous.metaEntry == null) {
3416               previous.metaEntry = m;
3417             } else {
3418               throw new IOException("Two entries in hbase:meta are same " + previous);
3419             }
3420           }
3421           PairOfSameType<HRegionInfo> mergeRegions = HRegionInfo.getMergeRegions(result);
3422           for (HRegionInfo mergeRegion : new HRegionInfo[] {
3423               mergeRegions.getFirst(), mergeRegions.getSecond() }) {
3424             if (mergeRegion != null) {
3425               // This region is already been merged
3426               HbckInfo hbInfo = getOrCreateInfo(mergeRegion.getEncodedName());
3427               hbInfo.setMerged(true);
3428             }
3429           }
3430 
3431           // show proof of progress to the user, once for every 100 records.
3432           if (countRecord % 100 == 0) {
3433             errors.progress();
3434           }
3435           countRecord++;
3436           return true;
3437         } catch (RuntimeException e) {
3438           LOG.error("Result=" + result);
3439           throw e;
3440         }
3441       }
3442     };
3443     if (!checkMetaOnly) {
3444       // Scan hbase:meta to pick up user regions
3445       MetaTableAccessor.fullScanRegions(connection, visitor);
3446     }
3447 
3448     errors.print("");
3449     return true;
3450   }
3451 
3452   /**
3453    * Stores the regioninfo entries scanned from META
3454    */
3455   static class MetaEntry extends HRegionInfo {
3456     ServerName regionServer;   // server hosting this region
3457     long modTime;          // timestamp of most recent modification metadata
3458     HRegionInfo splitA, splitB; //split daughters
3459 
3460     public MetaEntry(HRegionInfo rinfo, ServerName regionServer, long modTime) {
3461       this(rinfo, regionServer, modTime, null, null);
3462     }
3463 
3464     public MetaEntry(HRegionInfo rinfo, ServerName regionServer, long modTime,
3465         HRegionInfo splitA, HRegionInfo splitB) {
3466       super(rinfo);
3467       this.regionServer = regionServer;
3468       this.modTime = modTime;
3469       this.splitA = splitA;
3470       this.splitB = splitB;
3471     }
3472 
3473     @Override
3474     public boolean equals(Object o) {
3475       boolean superEq = super.equals(o);
3476       if (!superEq) {
3477         return superEq;
3478       }
3479 
3480       MetaEntry me = (MetaEntry) o;
3481       if (!regionServer.equals(me.regionServer)) {
3482         return false;
3483       }
3484       return (modTime == me.modTime);
3485     }
3486 
3487     @Override
3488     public int hashCode() {
3489       int hash = Arrays.hashCode(getRegionName());
3490       hash ^= getRegionId();
3491       hash ^= Arrays.hashCode(getStartKey());
3492       hash ^= Arrays.hashCode(getEndKey());
3493       hash ^= Boolean.valueOf(isOffline()).hashCode();
3494       hash ^= getTable().hashCode();
3495       if (regionServer != null) {
3496         hash ^= regionServer.hashCode();
3497       }
3498       hash ^= modTime;
3499       return hash;
3500     }
3501   }
3502 
3503   /**
3504    * Stores the regioninfo entries from HDFS
3505    */
3506   static class HdfsEntry {
3507     HRegionInfo hri;
3508     Path hdfsRegionDir = null;
3509     long hdfsRegionDirModTime  = 0;
3510     boolean hdfsRegioninfoFilePresent = false;
3511     boolean hdfsOnlyEdits = false;
3512   }
3513 
3514   /**
3515    * Stores the regioninfo retrieved from Online region servers.
3516    */
3517   static class OnlineEntry {
3518     HRegionInfo hri;
3519     ServerName hsa;
3520 
3521     @Override
3522     public String toString() {
3523       return hsa.toString() + ";" + hri.getRegionNameAsString();
3524     }
3525   }
3526 
3527   /**
3528    * Maintain information about a particular region.  It gathers information
3529    * from three places -- HDFS, META, and region servers.
3530    */
3531   public static class HbckInfo implements KeyRange {
3532     private MetaEntry metaEntry = null; // info in META
3533     private HdfsEntry hdfsEntry = null; // info in HDFS
3534     private List<OnlineEntry> deployedEntries = Lists.newArrayList(); // on Region Server
3535     private List<ServerName> deployedOn = Lists.newArrayList(); // info on RS's
3536     private boolean skipChecks = false; // whether to skip further checks to this region info.
3537     private boolean isMerged = false;// whether this region has already been merged into another one
3538     private int deployedReplicaId = HRegionInfo.DEFAULT_REPLICA_ID;
3539     private HRegionInfo primaryHRIForDeployedReplica = null;
3540 
3541     HbckInfo(MetaEntry metaEntry) {
3542       this.metaEntry = metaEntry;
3543     }
3544 
3545     public int getReplicaId() {
3546       if (metaEntry != null) return metaEntry.getReplicaId();
3547       return deployedReplicaId;
3548     }
3549 
3550     public synchronized void addServer(HRegionInfo hri, ServerName server) {
3551       OnlineEntry rse = new OnlineEntry() ;
3552       rse.hri = hri;
3553       rse.hsa = server;
3554       this.deployedEntries.add(rse);
3555       this.deployedOn.add(server);
3556       // save the replicaId that we see deployed in the cluster
3557       this.deployedReplicaId = hri.getReplicaId();
3558       this.primaryHRIForDeployedReplica =
3559           RegionReplicaUtil.getRegionInfoForDefaultReplica(hri);
3560     }
3561 
3562     @Override
3563     public synchronized String toString() {
3564       StringBuilder sb = new StringBuilder();
3565       sb.append("{ meta => ");
3566       sb.append((metaEntry != null)? metaEntry.getRegionNameAsString() : "null");
3567       sb.append( ", hdfs => " + getHdfsRegionDir());
3568       sb.append( ", deployed => " + Joiner.on(", ").join(deployedEntries));
3569       sb.append( ", replicaId => " + getReplicaId());
3570       sb.append(" }");
3571       return sb.toString();
3572     }
3573 
3574     @Override
3575     public byte[] getStartKey() {
3576       if (this.metaEntry != null) {
3577         return this.metaEntry.getStartKey();
3578       } else if (this.hdfsEntry != null) {
3579         return this.hdfsEntry.hri.getStartKey();
3580       } else {
3581         LOG.error("Entry " + this + " has no meta or hdfs region start key.");
3582         return null;
3583       }
3584     }
3585 
3586     @Override
3587     public byte[] getEndKey() {
3588       if (this.metaEntry != null) {
3589         return this.metaEntry.getEndKey();
3590       } else if (this.hdfsEntry != null) {
3591         return this.hdfsEntry.hri.getEndKey();
3592       } else {
3593         LOG.error("Entry " + this + " has no meta or hdfs region start key.");
3594         return null;
3595       }
3596     }
3597 
3598     public TableName getTableName() {
3599       if (this.metaEntry != null) {
3600         return this.metaEntry.getTable();
3601       } else if (this.hdfsEntry != null) {
3602         // we are only guaranteed to have a path and not an HRI for hdfsEntry,
3603         // so we get the name from the Path
3604         Path tableDir = this.hdfsEntry.hdfsRegionDir.getParent();
3605         return FSUtils.getTableName(tableDir);
3606       } else {
3607         // return the info from the first online/deployed hri
3608         for (OnlineEntry e : deployedEntries) {
3609           return e.hri.getTable();
3610         }
3611         return null;
3612       }
3613     }
3614 
3615     public String getRegionNameAsString() {
3616       if (metaEntry != null) {
3617         return metaEntry.getRegionNameAsString();
3618       } else if (hdfsEntry != null) {
3619         if (hdfsEntry.hri != null) {
3620           return hdfsEntry.hri.getRegionNameAsString();
3621         }
3622       } else {
3623         // return the info from the first online/deployed hri
3624         for (OnlineEntry e : deployedEntries) {
3625           return e.hri.getRegionNameAsString();
3626         }
3627       }
3628       return null;
3629     }
3630 
3631     public byte[] getRegionName() {
3632       if (metaEntry != null) {
3633         return metaEntry.getRegionName();
3634       } else if (hdfsEntry != null) {
3635         return hdfsEntry.hri.getRegionName();
3636       } else {
3637         // return the info from the first online/deployed hri
3638         for (OnlineEntry e : deployedEntries) {
3639           return e.hri.getRegionName();
3640         }
3641         return null;
3642       }
3643     }
3644 
3645     public HRegionInfo getPrimaryHRIForDeployedReplica() {
3646       return primaryHRIForDeployedReplica;
3647     }
3648 
3649     Path getHdfsRegionDir() {
3650       if (hdfsEntry == null) {
3651         return null;
3652       }
3653       return hdfsEntry.hdfsRegionDir;
3654     }
3655 
3656     boolean containsOnlyHdfsEdits() {
3657       if (hdfsEntry == null) {
3658         return false;
3659       }
3660       return hdfsEntry.hdfsOnlyEdits;
3661     }
3662 
3663     boolean isHdfsRegioninfoPresent() {
3664       if (hdfsEntry == null) {
3665         return false;
3666       }
3667       return hdfsEntry.hdfsRegioninfoFilePresent;
3668     }
3669 
3670     long getModTime() {
3671       if (hdfsEntry == null) {
3672         return 0;
3673       }
3674       return hdfsEntry.hdfsRegionDirModTime;
3675     }
3676 
3677     HRegionInfo getHdfsHRI() {
3678       if (hdfsEntry == null) {
3679         return null;
3680       }
3681       return hdfsEntry.hri;
3682     }
3683 
3684     public void setSkipChecks(boolean skipChecks) {
3685       this.skipChecks = skipChecks;
3686     }
3687 
3688     public boolean isSkipChecks() {
3689       return skipChecks;
3690     }
3691 
3692     public void setMerged(boolean isMerged) {
3693       this.isMerged = isMerged;
3694     }
3695 
3696     public boolean isMerged() {
3697       return this.isMerged;
3698     }
3699   }
3700 
3701   final static Comparator<HbckInfo> cmp = new Comparator<HbckInfo>() {
3702     @Override
3703     public int compare(HbckInfo l, HbckInfo r) {
3704       if (l == r) {
3705         // same instance
3706         return 0;
3707       }
3708 
3709       int tableCompare = l.getTableName().compareTo(r.getTableName());
3710       if (tableCompare != 0) {
3711         return tableCompare;
3712       }
3713 
3714       int startComparison = RegionSplitCalculator.BYTES_COMPARATOR.compare(
3715           l.getStartKey(), r.getStartKey());
3716       if (startComparison != 0) {
3717         return startComparison;
3718       }
3719 
3720       // Special case for absolute endkey
3721       byte[] endKey = r.getEndKey();
3722       endKey = (endKey.length == 0) ? null : endKey;
3723       byte[] endKey2 = l.getEndKey();
3724       endKey2 = (endKey2.length == 0) ? null : endKey2;
3725       int endComparison = RegionSplitCalculator.BYTES_COMPARATOR.compare(
3726           endKey2,  endKey);
3727 
3728       if (endComparison != 0) {
3729         return endComparison;
3730       }
3731 
3732       // use regionId as tiebreaker.
3733       // Null is considered after all possible values so make it bigger.
3734       if (l.hdfsEntry == null && r.hdfsEntry == null) {
3735         return 0;
3736       }
3737       if (l.hdfsEntry == null && r.hdfsEntry != null) {
3738         return 1;
3739       }
3740       // l.hdfsEntry must not be null
3741       if (r.hdfsEntry == null) {
3742         return -1;
3743       }
3744       // both l.hdfsEntry and r.hdfsEntry must not be null.
3745       return (int) (l.hdfsEntry.hri.getRegionId()- r.hdfsEntry.hri.getRegionId());
3746     }
3747   };
3748 
3749   /**
3750    * Prints summary of all tables found on the system.
3751    */
3752   private void printTableSummary(SortedMap<TableName, TableInfo> tablesInfo) {
3753     StringBuilder sb = new StringBuilder();
3754     int numOfSkippedRegions;
3755     errors.print("Summary:");
3756     for (TableInfo tInfo : tablesInfo.values()) {
3757       numOfSkippedRegions = (skippedRegions.containsKey(tInfo.getName())) ?
3758           skippedRegions.get(tInfo.getName()).size() : 0;
3759 
3760       if (errors.tableHasErrors(tInfo)) {
3761         errors.print("Table " + tInfo.getName() + " is inconsistent.");
3762       } else if (numOfSkippedRegions > 0){
3763         errors.print("Table " + tInfo.getName() + " is okay (with "
3764           + numOfSkippedRegions + " skipped regions).");
3765       }
3766       else {
3767         errors.print("Table " + tInfo.getName() + " is okay.");
3768       }
3769       errors.print("    Number of regions: " + tInfo.getNumRegions());
3770       if (numOfSkippedRegions > 0) {
3771         Set<String> skippedRegionStrings = skippedRegions.get(tInfo.getName());
3772         System.out.println("    Number of skipped regions: " + numOfSkippedRegions);
3773         System.out.println("      List of skipped regions:");
3774         for(String sr : skippedRegionStrings) {
3775           System.out.println("        " + sr);
3776         }
3777       }
3778       sb.setLength(0); // clear out existing buffer, if any.
3779       sb.append("    Deployed on: ");
3780       for (ServerName server : tInfo.deployedOn) {
3781         sb.append(" " + server.toString());
3782       }
3783       errors.print(sb.toString());
3784     }
3785   }
3786 
3787   static ErrorReporter getErrorReporter(
3788       final Configuration conf) throws ClassNotFoundException {
3789     Class<? extends ErrorReporter> reporter = conf.getClass("hbasefsck.errorreporter", PrintingErrorReporter.class, ErrorReporter.class);
3790     return ReflectionUtils.newInstance(reporter, conf);
3791   }
3792 
3793   public interface ErrorReporter {
3794     enum ERROR_CODE {
3795       UNKNOWN, NO_META_REGION, NULL_META_REGION, NO_VERSION_FILE, NOT_IN_META_HDFS, NOT_IN_META,
3796       NOT_IN_META_OR_DEPLOYED, NOT_IN_HDFS_OR_DEPLOYED, NOT_IN_HDFS, SERVER_DOES_NOT_MATCH_META,
3797       NOT_DEPLOYED,
3798       MULTI_DEPLOYED, SHOULD_NOT_BE_DEPLOYED, MULTI_META_REGION, RS_CONNECT_FAILURE,
3799       FIRST_REGION_STARTKEY_NOT_EMPTY, LAST_REGION_ENDKEY_NOT_EMPTY, DUPE_STARTKEYS,
3800       HOLE_IN_REGION_CHAIN, OVERLAP_IN_REGION_CHAIN, REGION_CYCLE, DEGENERATE_REGION,
3801       ORPHAN_HDFS_REGION, LINGERING_SPLIT_PARENT, NO_TABLEINFO_FILE, LINGERING_REFERENCE_HFILE,
3802       WRONG_USAGE, EMPTY_META_CELL, EXPIRED_TABLE_LOCK, BOUNDARIES_ERROR, ORPHAN_TABLE_STATE,
3803       NO_TABLE_STATE
3804     }
3805     void clear();
3806     void report(String message);
3807     void reportError(String message);
3808     void reportError(ERROR_CODE errorCode, String message);
3809     void reportError(ERROR_CODE errorCode, String message, TableInfo table);
3810     void reportError(ERROR_CODE errorCode, String message, TableInfo table, HbckInfo info);
3811     void reportError(
3812       ERROR_CODE errorCode,
3813       String message,
3814       TableInfo table,
3815       HbckInfo info1,
3816       HbckInfo info2
3817     );
3818     int summarize();
3819     void detail(String details);
3820     ArrayList<ERROR_CODE> getErrorList();
3821     void progress();
3822     void print(String message);
3823     void resetErrors();
3824     boolean tableHasErrors(TableInfo table);
3825   }
3826 
3827   static class PrintingErrorReporter implements ErrorReporter {
3828     public int errorCount = 0;
3829     private int showProgress;
3830     // How frequently calls to progress() will create output
3831     private static final int progressThreshold = 100;
3832 
3833     Set<TableInfo> errorTables = new HashSet<TableInfo>();
3834 
3835     // for use by unit tests to verify which errors were discovered
3836     private ArrayList<ERROR_CODE> errorList = new ArrayList<ERROR_CODE>();
3837 
3838     @Override
3839     public void clear() {
3840       errorTables.clear();
3841       errorList.clear();
3842       errorCount = 0;
3843     }
3844 
3845     @Override
3846     public synchronized void reportError(ERROR_CODE errorCode, String message) {
3847       if (errorCode == ERROR_CODE.WRONG_USAGE) {
3848         System.err.println(message);
3849         return;
3850       }
3851 
3852       errorList.add(errorCode);
3853       if (!summary) {
3854         System.out.println("ERROR: " + message);
3855       }
3856       errorCount++;
3857       showProgress = 0;
3858     }
3859 
3860     @Override
3861     public synchronized void reportError(ERROR_CODE errorCode, String message, TableInfo table) {
3862       errorTables.add(table);
3863       reportError(errorCode, message);
3864     }
3865 
3866     @Override
3867     public synchronized void reportError(ERROR_CODE errorCode, String message, TableInfo table,
3868                                          HbckInfo info) {
3869       errorTables.add(table);
3870       String reference = "(region " + info.getRegionNameAsString() + ")";
3871       reportError(errorCode, reference + " " + message);
3872     }
3873 
3874     @Override
3875     public synchronized void reportError(ERROR_CODE errorCode, String message, TableInfo table,
3876                                          HbckInfo info1, HbckInfo info2) {
3877       errorTables.add(table);
3878       String reference = "(regions " + info1.getRegionNameAsString()
3879           + " and " + info2.getRegionNameAsString() + ")";
3880       reportError(errorCode, reference + " " + message);
3881     }
3882 
3883     @Override
3884     public synchronized void reportError(String message) {
3885       reportError(ERROR_CODE.UNKNOWN, message);
3886     }
3887 
3888     /**
3889      * Report error information, but do not increment the error count.  Intended for cases
3890      * where the actual error would have been reported previously.
3891      * @param message
3892      */
3893     @Override
3894     public synchronized void report(String message) {
3895       if (! summary) {
3896         System.out.println("ERROR: " + message);
3897       }
3898       showProgress = 0;
3899     }
3900 
3901     @Override
3902     public synchronized int summarize() {
3903       System.out.println(Integer.toString(errorCount) +
3904                          " inconsistencies detected.");
3905       if (errorCount == 0) {
3906         System.out.println("Status: OK");
3907         return 0;
3908       } else {
3909         System.out.println("Status: INCONSISTENT");
3910         return -1;
3911       }
3912     }
3913 
3914     @Override
3915     public ArrayList<ERROR_CODE> getErrorList() {
3916       return errorList;
3917     }
3918 
3919     @Override
3920     public synchronized void print(String message) {
3921       if (!summary) {
3922         System.out.println(message);
3923       }
3924     }
3925 
3926     @Override
3927     public boolean tableHasErrors(TableInfo table) {
3928       return errorTables.contains(table);
3929     }
3930 
3931     @Override
3932     public void resetErrors() {
3933       errorCount = 0;
3934     }
3935 
3936     @Override
3937     public synchronized void detail(String message) {
3938       if (details) {
3939         System.out.println(message);
3940       }
3941       showProgress = 0;
3942     }
3943 
3944     @Override
3945     public synchronized void progress() {
3946       if (showProgress++ == progressThreshold) {
3947         if (!summary) {
3948           System.out.print(".");
3949         }
3950         showProgress = 0;
3951       }
3952     }
3953   }
3954 
3955   /**
3956    * Contact a region server and get all information from it
3957    */
3958   static class WorkItemRegion implements Callable<Void> {
3959     private HBaseFsck hbck;
3960     private ServerName rsinfo;
3961     private ErrorReporter errors;
3962     private HConnection connection;
3963 
3964     WorkItemRegion(HBaseFsck hbck, ServerName info,
3965                    ErrorReporter errors, HConnection connection) {
3966       this.hbck = hbck;
3967       this.rsinfo = info;
3968       this.errors = errors;
3969       this.connection = connection;
3970     }
3971 
3972     @Override
3973     public synchronized Void call() throws IOException {
3974       errors.progress();
3975       try {
3976         BlockingInterface server = connection.getAdmin(rsinfo);
3977 
3978         // list all online regions from this region server
3979         List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(server);
3980         regions = filterRegions(regions);
3981 
3982         if (details) {
3983           errors.detail("RegionServer: " + rsinfo.getServerName() +
3984                            " number of regions: " + regions.size());
3985           for (HRegionInfo rinfo: regions) {
3986             errors.detail("  " + rinfo.getRegionNameAsString() +
3987                              " id: " + rinfo.getRegionId() +
3988                              " encoded_name: " + rinfo.getEncodedName() +
3989                              " start: " + Bytes.toStringBinary(rinfo.getStartKey()) +
3990                              " end: " + Bytes.toStringBinary(rinfo.getEndKey()));
3991           }
3992         }
3993 
3994         // check to see if the existence of this region matches the region in META
3995         for (HRegionInfo r:regions) {
3996           HbckInfo hbi = hbck.getOrCreateInfo(r.getEncodedName());
3997           hbi.addServer(r, rsinfo);
3998         }
3999       } catch (IOException e) {          // unable to connect to the region server.
4000         errors.reportError(ERROR_CODE.RS_CONNECT_FAILURE, "RegionServer: " + rsinfo.getServerName() +
4001           " Unable to fetch region information. " + e);
4002         throw e;
4003       }
4004       return null;
4005     }
4006 
4007     private List<HRegionInfo> filterRegions(List<HRegionInfo> regions) {
4008       List<HRegionInfo> ret = Lists.newArrayList();
4009       for (HRegionInfo hri : regions) {
4010         if (hri.isMetaTable() || (!hbck.checkMetaOnly
4011             && hbck.isTableIncluded(hri.getTable()))) {
4012           ret.add(hri);
4013         }
4014       }
4015       return ret;
4016     }
4017   }
4018 
4019   /**
4020    * Contact hdfs and get all information about specified table directory into
4021    * regioninfo list.
4022    */
4023   static class WorkItemHdfsDir implements Callable<Void> {
4024     private HBaseFsck hbck;
4025     private FileStatus tableDir;
4026     private ErrorReporter errors;
4027     private FileSystem fs;
4028 
4029     WorkItemHdfsDir(HBaseFsck hbck, FileSystem fs, ErrorReporter errors,
4030                     FileStatus status) {
4031       this.hbck = hbck;
4032       this.fs = fs;
4033       this.tableDir = status;
4034       this.errors = errors;
4035     }
4036 
4037     @Override
4038     public synchronized Void call() throws IOException {
4039       try {
4040         // level 2: <HBASE_DIR>/<table>/*
4041         FileStatus[] regionDirs = fs.listStatus(tableDir.getPath());
4042         for (FileStatus regionDir : regionDirs) {
4043           errors.progress();
4044           String encodedName = regionDir.getPath().getName();
4045           // ignore directories that aren't hexadecimal
4046           if (!encodedName.toLowerCase().matches("[0-9a-f]+")) {
4047             continue;
4048           }
4049 
4050           LOG.debug("Loading region info from hdfs:"+ regionDir.getPath());
4051           HbckInfo hbi = hbck.getOrCreateInfo(encodedName);
4052           HdfsEntry he = new HdfsEntry();
4053           synchronized (hbi) {
4054             if (hbi.getHdfsRegionDir() != null) {
4055               errors.print("Directory " + encodedName + " duplicate??" +
4056                            hbi.getHdfsRegionDir());
4057             }
4058 
4059             he.hdfsRegionDir = regionDir.getPath();
4060             he.hdfsRegionDirModTime = regionDir.getModificationTime();
4061             Path regioninfoFile = new Path(he.hdfsRegionDir, HRegionFileSystem.REGION_INFO_FILE);
4062             he.hdfsRegioninfoFilePresent = fs.exists(regioninfoFile);
4063             // we add to orphan list when we attempt to read .regioninfo
4064 
4065             // Set a flag if this region contains only edits
4066             // This is special case if a region is left after split
4067             he.hdfsOnlyEdits = true;
4068             FileStatus[] subDirs = fs.listStatus(regionDir.getPath());
4069             Path ePath = WALSplitter.getRegionDirRecoveredEditsDir(regionDir.getPath());
4070             for (FileStatus subDir : subDirs) {
4071               errors.progress();
4072               String sdName = subDir.getPath().getName();
4073               if (!sdName.startsWith(".") && !sdName.equals(ePath.getName())) {
4074                 he.hdfsOnlyEdits = false;
4075                 break;
4076               }
4077             }
4078             hbi.hdfsEntry = he;
4079           }
4080         }
4081       } catch (IOException e) {
4082         // unable to connect to the region server.
4083         errors.reportError(ERROR_CODE.RS_CONNECT_FAILURE, "Table Directory: "
4084             + tableDir.getPath().getName()
4085             + " Unable to fetch region information. " + e);
4086         throw e;
4087       }
4088       return null;
4089     }
4090   }
4091 
4092   /**
4093    * Contact hdfs and get all information about specified table directory into
4094    * regioninfo list.
4095    */
4096   static class WorkItemHdfsRegionInfo implements Callable<Void> {
4097     private HbckInfo hbi;
4098     private HBaseFsck hbck;
4099     private ErrorReporter errors;
4100 
4101     WorkItemHdfsRegionInfo(HbckInfo hbi, HBaseFsck hbck, ErrorReporter errors) {
4102       this.hbi = hbi;
4103       this.hbck = hbck;
4104       this.errors = errors;
4105     }
4106 
4107     @Override
4108     public synchronized Void call() throws IOException {
4109       // only load entries that haven't been loaded yet.
4110       if (hbi.getHdfsHRI() == null) {
4111         try {
4112           errors.progress();
4113           hbck.loadHdfsRegioninfo(hbi);
4114         } catch (IOException ioe) {
4115           String msg = "Orphan region in HDFS: Unable to load .regioninfo from table "
4116               + hbi.getTableName() + " in hdfs dir "
4117               + hbi.getHdfsRegionDir()
4118               + "!  It may be an invalid format or version file.  Treating as "
4119               + "an orphaned regiondir.";
4120           errors.reportError(ERROR_CODE.ORPHAN_HDFS_REGION, msg);
4121           try {
4122             hbck.debugLsr(hbi.getHdfsRegionDir());
4123           } catch (IOException ioe2) {
4124             LOG.error("Unable to read directory " + hbi.getHdfsRegionDir(), ioe2);
4125             throw ioe2;
4126           }
4127           hbck.orphanHdfsDirs.add(hbi);
4128           throw ioe;
4129         }
4130       }
4131       return null;
4132     }
4133   };
4134 
4135   /**
4136    * Display the full report from fsck. This displays all live and dead region
4137    * servers, and all known regions.
4138    */
4139   public static void setDisplayFullReport() {
4140     details = true;
4141   }
4142 
4143   /**
4144    * Set summary mode.
4145    * Print only summary of the tables and status (OK or INCONSISTENT)
4146    */
4147   static void setSummary() {
4148     summary = true;
4149   }
4150 
4151   /**
4152    * Set hbase:meta check mode.
4153    * Print only info about hbase:meta table deployment/state
4154    */
4155   void setCheckMetaOnly() {
4156     checkMetaOnly = true;
4157   }
4158 
4159   /**
4160    * Set region boundaries check mode.
4161    */
4162   void setRegionBoundariesCheck() {
4163     checkRegionBoundaries = true;
4164   }
4165 
4166   /**
4167    * Set table locks fix mode.
4168    * Delete table locks held for a long time
4169    */
4170   public void setFixTableLocks(boolean shouldFix) {
4171     fixTableLocks = shouldFix;
4172     fixAny |= shouldFix;
4173   }
4174 
4175   /**
4176    * Check if we should rerun fsck again. This checks if we've tried to
4177    * fix something and we should rerun fsck tool again.
4178    * Display the full report from fsck. This displays all live and dead
4179    * region servers, and all known regions.
4180    */
4181   void setShouldRerun() {
4182     rerun = true;
4183   }
4184 
4185   boolean shouldRerun() {
4186     return rerun;
4187   }
4188 
4189   /**
4190    * Fix inconsistencies found by fsck. This should try to fix errors (if any)
4191    * found by fsck utility.
4192    */
4193   public void setFixAssignments(boolean shouldFix) {
4194     fixAssignments = shouldFix;
4195     fixAny |= shouldFix;
4196   }
4197 
4198   boolean shouldFixAssignments() {
4199     return fixAssignments;
4200   }
4201 
4202   public void setFixMeta(boolean shouldFix) {
4203     fixMeta = shouldFix;
4204     fixAny |= shouldFix;
4205   }
4206 
4207   boolean shouldFixMeta() {
4208     return fixMeta;
4209   }
4210 
4211   public void setFixEmptyMetaCells(boolean shouldFix) {
4212     fixEmptyMetaCells = shouldFix;
4213     fixAny |= shouldFix;
4214   }
4215 
4216   boolean shouldFixEmptyMetaCells() {
4217     return fixEmptyMetaCells;
4218   }
4219 
4220   public void setCheckHdfs(boolean checking) {
4221     checkHdfs = checking;
4222   }
4223 
4224   boolean shouldCheckHdfs() {
4225     return checkHdfs;
4226   }
4227 
4228   public void setFixHdfsHoles(boolean shouldFix) {
4229     fixHdfsHoles = shouldFix;
4230     fixAny |= shouldFix;
4231   }
4232 
4233   boolean shouldFixHdfsHoles() {
4234     return fixHdfsHoles;
4235   }
4236 
4237   public void setFixTableOrphans(boolean shouldFix) {
4238     fixTableOrphans = shouldFix;
4239     fixAny |= shouldFix;
4240   }
4241 
4242   boolean shouldFixTableOrphans() {
4243     return fixTableOrphans;
4244   }
4245 
4246   public void setFixHdfsOverlaps(boolean shouldFix) {
4247     fixHdfsOverlaps = shouldFix;
4248     fixAny |= shouldFix;
4249   }
4250 
4251   boolean shouldFixHdfsOverlaps() {
4252     return fixHdfsOverlaps;
4253   }
4254 
4255   public void setFixHdfsOrphans(boolean shouldFix) {
4256     fixHdfsOrphans = shouldFix;
4257     fixAny |= shouldFix;
4258   }
4259 
4260   boolean shouldFixHdfsOrphans() {
4261     return fixHdfsOrphans;
4262   }
4263 
4264   public void setFixVersionFile(boolean shouldFix) {
4265     fixVersionFile = shouldFix;
4266     fixAny |= shouldFix;
4267   }
4268 
4269   public boolean shouldFixVersionFile() {
4270     return fixVersionFile;
4271   }
4272 
4273   public void setSidelineBigOverlaps(boolean sbo) {
4274     this.sidelineBigOverlaps = sbo;
4275   }
4276 
4277   public boolean shouldSidelineBigOverlaps() {
4278     return sidelineBigOverlaps;
4279   }
4280 
4281   public void setFixSplitParents(boolean shouldFix) {
4282     fixSplitParents = shouldFix;
4283     fixAny |= shouldFix;
4284   }
4285 
4286   boolean shouldFixSplitParents() {
4287     return fixSplitParents;
4288   }
4289 
4290   public void setFixReferenceFiles(boolean shouldFix) {
4291     fixReferenceFiles = shouldFix;
4292     fixAny |= shouldFix;
4293   }
4294 
4295   boolean shouldFixReferenceFiles() {
4296     return fixReferenceFiles;
4297   }
4298 
4299   public boolean shouldIgnorePreCheckPermission() {
4300     return !fixAny || ignorePreCheckPermission;
4301   }
4302 
4303   public void setIgnorePreCheckPermission(boolean ignorePreCheckPermission) {
4304     this.ignorePreCheckPermission = ignorePreCheckPermission;
4305   }
4306 
4307   /**
4308    * @param mm maximum number of regions to merge into a single region.
4309    */
4310   public void setMaxMerge(int mm) {
4311     this.maxMerge = mm;
4312   }
4313 
4314   public int getMaxMerge() {
4315     return maxMerge;
4316   }
4317 
4318   public void setMaxOverlapsToSideline(int mo) {
4319     this.maxOverlapsToSideline = mo;
4320   }
4321 
4322   public int getMaxOverlapsToSideline() {
4323     return maxOverlapsToSideline;
4324   }
4325 
4326   /**
4327    * Only check/fix tables specified by the list,
4328    * Empty list means all tables are included.
4329    */
4330   boolean isTableIncluded(TableName table) {
4331     return (tablesIncluded.size() == 0) || tablesIncluded.contains(table);
4332   }
4333 
4334   public void includeTable(TableName table) {
4335     tablesIncluded.add(table);
4336   }
4337 
4338   Set<TableName> getIncludedTables() {
4339     return new HashSet<TableName>(tablesIncluded);
4340   }
4341 
4342   /**
4343    * We are interested in only those tables that have not changed their state in
4344    * hbase:meta during the last few seconds specified by hbase.admin.fsck.timelag
4345    * @param seconds - the time in seconds
4346    */
4347   public void setTimeLag(long seconds) {
4348     timelag = seconds * 1000; // convert to milliseconds
4349   }
4350 
4351   /**
4352    *
4353    * @param sidelineDir - HDFS path to sideline data
4354    */
4355   public void setSidelineDir(String sidelineDir) {
4356     this.sidelineDir = new Path(sidelineDir);
4357   }
4358 
4359   protected HFileCorruptionChecker createHFileCorruptionChecker(boolean sidelineCorruptHFiles) throws IOException {
4360     return new HFileCorruptionChecker(getConf(), executor, sidelineCorruptHFiles);
4361   }
4362 
4363   public HFileCorruptionChecker getHFilecorruptionChecker() {
4364     return hfcc;
4365   }
4366 
4367   public void setHFileCorruptionChecker(HFileCorruptionChecker hfcc) {
4368     this.hfcc = hfcc;
4369   }
4370 
4371   public void setRetCode(int code) {
4372     this.retcode = code;
4373   }
4374 
4375   public int getRetCode() {
4376     return retcode;
4377   }
4378 
4379   protected HBaseFsck printUsageAndExit() {
4380     StringWriter sw = new StringWriter(2048);
4381     PrintWriter out = new PrintWriter(sw);
4382     out.println("Usage: fsck [opts] {only tables}");
4383     out.println(" where [opts] are:");
4384     out.println("   -help Display help options (this)");
4385     out.println("   -details Display full report of all regions.");
4386     out.println("   -timelag <timeInSeconds>  Process only regions that " +
4387                        " have not experienced any metadata updates in the last " +
4388                        " <timeInSeconds> seconds.");
4389     out.println("   -sleepBeforeRerun <timeInSeconds> Sleep this many seconds" +
4390         " before checking if the fix worked if run with -fix");
4391     out.println("   -summary Print only summary of the tables and status.");
4392     out.println("   -metaonly Only check the state of the hbase:meta table.");
4393     out.println("   -sidelineDir <hdfs://> HDFS path to backup existing meta.");
4394     out.println("   -boundaries Verify that regions boundaries are the same between META and store files.");
4395 
4396     out.println("");
4397     out.println("  Metadata Repair options: (expert features, use with caution!)");
4398     out.println("   -fix              Try to fix region assignments.  This is for backwards compatiblity");
4399     out.println("   -fixAssignments   Try to fix region assignments.  Replaces the old -fix");
4400     out.println("   -fixMeta          Try to fix meta problems.  This assumes HDFS region info is good.");
4401     out.println("   -noHdfsChecking   Don't load/check region info from HDFS."
4402         + " Assumes hbase:meta region info is good. Won't check/fix any HDFS issue, e.g. hole, orphan, or overlap");
4403     out.println("   -fixHdfsHoles     Try to fix region holes in hdfs.");
4404     out.println("   -fixHdfsOrphans   Try to fix region dirs with no .regioninfo file in hdfs");
4405     out.println("   -fixTableOrphans  Try to fix table dirs with no .tableinfo file in hdfs (online mode only)");
4406     out.println("   -fixHdfsOverlaps  Try to fix region overlaps in hdfs.");
4407     out.println("   -fixVersionFile   Try to fix missing hbase.version file in hdfs.");
4408     out.println("   -maxMerge <n>     When fixing region overlaps, allow at most <n> regions to merge. (n=" + DEFAULT_MAX_MERGE +" by default)");
4409     out.println("   -sidelineBigOverlaps  When fixing region overlaps, allow to sideline big overlaps");
4410     out.println("   -maxOverlapsToSideline <n>  When fixing region overlaps, allow at most <n> regions to sideline per group. (n=" + DEFAULT_OVERLAPS_TO_SIDELINE +" by default)");
4411     out.println("   -fixSplitParents  Try to force offline split parents to be online.");
4412     out.println("   -ignorePreCheckPermission  ignore filesystem permission pre-check");
4413     out.println("   -fixReferenceFiles  Try to offline lingering reference store files");
4414     out.println("   -fixEmptyMetaCells  Try to fix hbase:meta entries not referencing any region"
4415         + " (empty REGIONINFO_QUALIFIER rows)");
4416 
4417     out.println("");
4418     out.println("  Datafile Repair options: (expert features, use with caution!)");
4419     out.println("   -checkCorruptHFiles     Check all Hfiles by opening them to make sure they are valid");
4420     out.println("   -sidelineCorruptHFiles  Quarantine corrupted HFiles.  implies -checkCorruptHFiles");
4421 
4422     out.println("");
4423     out.println("  Metadata Repair shortcuts");
4424     out.println("   -repair           Shortcut for -fixAssignments -fixMeta -fixHdfsHoles " +
4425         "-fixHdfsOrphans -fixHdfsOverlaps -fixVersionFile -sidelineBigOverlaps -fixReferenceFiles -fixTableLocks");
4426     out.println("   -repairHoles      Shortcut for -fixAssignments -fixMeta -fixHdfsHoles");
4427 
4428     out.println("");
4429     out.println("  Table lock options");
4430     out.println("   -fixTableLocks    Deletes table locks held for a long time (hbase.table.lock.expire.ms, 10min by default)");
4431 
4432     out.flush();
4433     errors.reportError(ERROR_CODE.WRONG_USAGE, sw.toString());
4434 
4435     setRetCode(-2);
4436     return this;
4437   }
4438 
4439   /**
4440    * Main program
4441    *
4442    * @param args
4443    * @throws Exception
4444    */
4445   public static void main(String[] args) throws Exception {
4446     // create a fsck object
4447     Configuration conf = HBaseConfiguration.create();
4448     Path hbasedir = FSUtils.getRootDir(conf);
4449     URI defaultFs = hbasedir.getFileSystem(conf).getUri();
4450     FSUtils.setFsDefault(conf, new Path(defaultFs));
4451     int ret = ToolRunner.run(new HBaseFsckTool(conf), args);
4452     System.exit(ret);
4453   }
4454 
4455   /**
4456    * This is a Tool wrapper that gathers -Dxxx=yyy configuration settings from the command line.
4457    */
4458   static class HBaseFsckTool extends Configured implements Tool {
4459     HBaseFsckTool(Configuration conf) { super(conf); }
4460     @Override
4461     public int run(String[] args) throws Exception {
4462       HBaseFsck hbck = new HBaseFsck(getConf());
4463       hbck.exec(hbck.executor, args);
4464       hbck.close();
4465       return hbck.getRetCode();
4466     }
4467   };
4468 
4469 
4470   public HBaseFsck exec(ExecutorService exec, String[] args) throws KeeperException, IOException,
4471     ServiceException, InterruptedException {
4472     long sleepBeforeRerun = DEFAULT_SLEEP_BEFORE_RERUN;
4473 
4474     boolean checkCorruptHFiles = false;
4475     boolean sidelineCorruptHFiles = false;
4476 
4477     // Process command-line args.
4478     for (int i = 0; i < args.length; i++) {
4479       String cmd = args[i];
4480       if (cmd.equals("-help") || cmd.equals("-h")) {
4481         return printUsageAndExit();
4482       } else if (cmd.equals("-details")) {
4483         setDisplayFullReport();
4484       } else if (cmd.equals("-timelag")) {
4485         if (i == args.length - 1) {
4486           errors.reportError(ERROR_CODE.WRONG_USAGE, "HBaseFsck: -timelag needs a value.");
4487           return printUsageAndExit();
4488         }
4489         try {
4490           long timelag = Long.parseLong(args[i+1]);
4491           setTimeLag(timelag);
4492         } catch (NumberFormatException e) {
4493           errors.reportError(ERROR_CODE.WRONG_USAGE, "-timelag needs a numeric value.");
4494           return printUsageAndExit();
4495         }
4496         i++;
4497       } else if (cmd.equals("-sleepBeforeRerun")) {
4498         if (i == args.length - 1) {
4499           errors.reportError(ERROR_CODE.WRONG_USAGE,
4500             "HBaseFsck: -sleepBeforeRerun needs a value.");
4501           return printUsageAndExit();
4502         }
4503         try {
4504           sleepBeforeRerun = Long.parseLong(args[i+1]);
4505         } catch (NumberFormatException e) {
4506           errors.reportError(ERROR_CODE.WRONG_USAGE, "-sleepBeforeRerun needs a numeric value.");
4507           return printUsageAndExit();
4508         }
4509         i++;
4510       } else if (cmd.equals("-sidelineDir")) {
4511         if (i == args.length - 1) {
4512           errors.reportError(ERROR_CODE.WRONG_USAGE, "HBaseFsck: -sidelineDir needs a value.");
4513           return printUsageAndExit();
4514         }
4515         i++;
4516         setSidelineDir(args[i]);
4517       } else if (cmd.equals("-fix")) {
4518         errors.reportError(ERROR_CODE.WRONG_USAGE,
4519           "This option is deprecated, please use  -fixAssignments instead.");
4520         setFixAssignments(true);
4521       } else if (cmd.equals("-fixAssignments")) {
4522         setFixAssignments(true);
4523       } else if (cmd.equals("-fixMeta")) {
4524         setFixMeta(true);
4525       } else if (cmd.equals("-noHdfsChecking")) {
4526         setCheckHdfs(false);
4527       } else if (cmd.equals("-fixHdfsHoles")) {
4528         setFixHdfsHoles(true);
4529       } else if (cmd.equals("-fixHdfsOrphans")) {
4530         setFixHdfsOrphans(true);
4531       } else if (cmd.equals("-fixTableOrphans")) {
4532         setFixTableOrphans(true);
4533       } else if (cmd.equals("-fixHdfsOverlaps")) {
4534         setFixHdfsOverlaps(true);
4535       } else if (cmd.equals("-fixVersionFile")) {
4536         setFixVersionFile(true);
4537       } else if (cmd.equals("-sidelineBigOverlaps")) {
4538         setSidelineBigOverlaps(true);
4539       } else if (cmd.equals("-fixSplitParents")) {
4540         setFixSplitParents(true);
4541       } else if (cmd.equals("-ignorePreCheckPermission")) {
4542         setIgnorePreCheckPermission(true);
4543       } else if (cmd.equals("-checkCorruptHFiles")) {
4544         checkCorruptHFiles = true;
4545       } else if (cmd.equals("-sidelineCorruptHFiles")) {
4546         sidelineCorruptHFiles = true;
4547       } else if (cmd.equals("-fixReferenceFiles")) {
4548         setFixReferenceFiles(true);
4549       } else if (cmd.equals("-fixEmptyMetaCells")) {
4550         setFixEmptyMetaCells(true);
4551       } else if (cmd.equals("-repair")) {
4552         // this attempts to merge overlapping hdfs regions, needs testing
4553         // under load
4554         setFixHdfsHoles(true);
4555         setFixHdfsOrphans(true);
4556         setFixMeta(true);
4557         setFixAssignments(true);
4558         setFixHdfsOverlaps(true);
4559         setFixVersionFile(true);
4560         setSidelineBigOverlaps(true);
4561         setFixSplitParents(false);
4562         setCheckHdfs(true);
4563         setFixReferenceFiles(true);
4564         setFixTableLocks(true);
4565       } else if (cmd.equals("-repairHoles")) {
4566         // this will make all missing hdfs regions available but may lose data
4567         setFixHdfsHoles(true);
4568         setFixHdfsOrphans(false);
4569         setFixMeta(true);
4570         setFixAssignments(true);
4571         setFixHdfsOverlaps(false);
4572         setSidelineBigOverlaps(false);
4573         setFixSplitParents(false);
4574         setCheckHdfs(true);
4575       } else if (cmd.equals("-maxOverlapsToSideline")) {
4576         if (i == args.length - 1) {
4577           errors.reportError(ERROR_CODE.WRONG_USAGE,
4578             "-maxOverlapsToSideline needs a numeric value argument.");
4579           return printUsageAndExit();
4580         }
4581         try {
4582           int maxOverlapsToSideline = Integer.parseInt(args[i+1]);
4583           setMaxOverlapsToSideline(maxOverlapsToSideline);
4584         } catch (NumberFormatException e) {
4585           errors.reportError(ERROR_CODE.WRONG_USAGE,
4586             "-maxOverlapsToSideline needs a numeric value argument.");
4587           return printUsageAndExit();
4588         }
4589         i++;
4590       } else if (cmd.equals("-maxMerge")) {
4591         if (i == args.length - 1) {
4592           errors.reportError(ERROR_CODE.WRONG_USAGE,
4593             "-maxMerge needs a numeric value argument.");
4594           return printUsageAndExit();
4595         }
4596         try {
4597           int maxMerge = Integer.parseInt(args[i+1]);
4598           setMaxMerge(maxMerge);
4599         } catch (NumberFormatException e) {
4600           errors.reportError(ERROR_CODE.WRONG_USAGE,
4601             "-maxMerge needs a numeric value argument.");
4602           return printUsageAndExit();
4603         }
4604         i++;
4605       } else if (cmd.equals("-summary")) {
4606         setSummary();
4607       } else if (cmd.equals("-metaonly")) {
4608         setCheckMetaOnly();
4609       } else if (cmd.equals("-boundaries")) {
4610         setRegionBoundariesCheck();
4611       } else if (cmd.equals("-fixTableLocks")) {
4612         setFixTableLocks(true);
4613       } else if (cmd.startsWith("-")) {
4614         errors.reportError(ERROR_CODE.WRONG_USAGE, "Unrecognized option:" + cmd);
4615         return printUsageAndExit();
4616       } else {
4617         includeTable(TableName.valueOf(cmd));
4618         errors.print("Allow checking/fixes for table: " + cmd);
4619       }
4620     }
4621 
4622     errors.print("HBaseFsck command line options: " + StringUtils.join(args, " "));
4623 
4624     // pre-check current user has FS write permission or not
4625     try {
4626       preCheckPermission();
4627     } catch (AccessDeniedException ace) {
4628       Runtime.getRuntime().exit(-1);
4629     } catch (IOException ioe) {
4630       Runtime.getRuntime().exit(-1);
4631     }
4632 
4633     // do the real work of hbck
4634     connect();
4635 
4636     try {
4637       // if corrupt file mode is on, first fix them since they may be opened later
4638       if (checkCorruptHFiles || sidelineCorruptHFiles) {
4639         LOG.info("Checking all hfiles for corruption");
4640         HFileCorruptionChecker hfcc = createHFileCorruptionChecker(sidelineCorruptHFiles);
4641         setHFileCorruptionChecker(hfcc); // so we can get result
4642         Collection<TableName> tables = getIncludedTables();
4643         Collection<Path> tableDirs = new ArrayList<Path>();
4644         Path rootdir = FSUtils.getRootDir(getConf());
4645         if (tables.size() > 0) {
4646           for (TableName t : tables) {
4647             tableDirs.add(FSUtils.getTableDir(rootdir, t));
4648           }
4649         } else {
4650           tableDirs = FSUtils.getTableDirs(FSUtils.getCurrentFileSystem(getConf()), rootdir);
4651         }
4652         hfcc.checkTables(tableDirs);
4653         hfcc.report(errors);
4654       }
4655 
4656       // check and fix table integrity, region consistency.
4657       int code = onlineHbck();
4658       setRetCode(code);
4659       // If we have changed the HBase state it is better to run hbck again
4660       // to see if we haven't broken something else in the process.
4661       // We run it only once more because otherwise we can easily fall into
4662       // an infinite loop.
4663       if (shouldRerun()) {
4664         try {
4665           LOG.info("Sleeping " + sleepBeforeRerun + "ms before re-checking after fix...");
4666           Thread.sleep(sleepBeforeRerun);
4667         } catch (InterruptedException ie) {
4668           LOG.warn("Interrupted while sleeping");
4669           return this;
4670         }
4671         // Just report
4672         setFixAssignments(false);
4673         setFixMeta(false);
4674         setFixHdfsHoles(false);
4675         setFixHdfsOverlaps(false);
4676         setFixVersionFile(false);
4677         setFixTableOrphans(false);
4678         errors.resetErrors();
4679         code = onlineHbck();
4680         setRetCode(code);
4681       }
4682     } finally {
4683       IOUtils.closeQuietly(this);
4684     }
4685     return this;
4686   }
4687 
4688   /**
4689    * ls -r for debugging purposes
4690    */
4691   void debugLsr(Path p) throws IOException {
4692     debugLsr(getConf(), p, errors);
4693   }
4694 
4695   /**
4696    * ls -r for debugging purposes
4697    */
4698   public static void debugLsr(Configuration conf,
4699       Path p) throws IOException {
4700     debugLsr(conf, p, new PrintingErrorReporter());
4701   }
4702 
4703   /**
4704    * ls -r for debugging purposes
4705    */
4706   public static void debugLsr(Configuration conf,
4707       Path p, ErrorReporter errors) throws IOException {
4708     if (!LOG.isDebugEnabled() || p == null) {
4709       return;
4710     }
4711     FileSystem fs = p.getFileSystem(conf);
4712 
4713     if (!fs.exists(p)) {
4714       // nothing
4715       return;
4716     }
4717     errors.print(p.toString());
4718 
4719     if (fs.isFile(p)) {
4720       return;
4721     }
4722 
4723     if (fs.getFileStatus(p).isDirectory()) {
4724       FileStatus[] fss= fs.listStatus(p);
4725       for (FileStatus status : fss) {
4726         debugLsr(conf, status.getPath(), errors);
4727       }
4728     }
4729   }
4730 }