View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.master;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.util.ArrayList;
24  import java.util.HashSet;
25  import java.util.List;
26  import java.util.Set;
27  import java.util.concurrent.locks.Lock;
28  import java.util.concurrent.locks.ReentrantLock;
29  
30  import org.apache.commons.logging.Log;
31  import org.apache.commons.logging.LogFactory;
32  import org.apache.hadoop.classification.InterfaceAudience;
33  import org.apache.hadoop.conf.Configuration;
34  import org.apache.hadoop.fs.FileStatus;
35  import org.apache.hadoop.fs.FileSystem;
36  import org.apache.hadoop.fs.Path;
37  import org.apache.hadoop.fs.PathFilter;
38  import org.apache.hadoop.hbase.ClusterId;
39  import org.apache.hadoop.hbase.HColumnDescriptor;
40  import org.apache.hadoop.hbase.HConstants;
41  import org.apache.hadoop.hbase.HRegionInfo;
42  import org.apache.hadoop.hbase.HTableDescriptor;
43  import org.apache.hadoop.hbase.InvalidFamilyOperationException;
44  import org.apache.hadoop.hbase.Server;
45  import org.apache.hadoop.hbase.ServerName;
46  import org.apache.hadoop.hbase.TableName;
47  import org.apache.hadoop.hbase.backup.HFileArchiver;
48  import org.apache.hadoop.hbase.exceptions.DeserializationException;
49  import org.apache.hadoop.hbase.fs.HFileSystem;
50  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
51  import org.apache.hadoop.hbase.regionserver.HRegion;
52  import org.apache.hadoop.hbase.regionserver.wal.HLog;
53  import org.apache.hadoop.hbase.regionserver.wal.HLogUtil;
54  import org.apache.hadoop.hbase.util.Bytes;
55  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
56  import org.apache.hadoop.hbase.util.FSTableDescriptors;
57  import org.apache.hadoop.hbase.util.FSUtils;
58  import org.apache.hadoop.ipc.RemoteException;
59  import org.apache.zookeeper.KeeperException;
60  
61  /**
62   * This class abstracts a bunch of operations the HMaster needs to interact with
63   * the underlying file system, including splitting log files, checking file
64   * system status, etc.
65   */
66  @InterfaceAudience.Private
67  public class MasterFileSystem {
68    private static final Log LOG = LogFactory.getLog(MasterFileSystem.class.getName());
69    // HBase configuration
70    Configuration conf;
71    // master status
72    Server master;
73    // metrics for master
74    private final MetricsMasterFileSystem metricsMasterFilesystem = new MetricsMasterFileSystem();
75    // Persisted unique cluster ID
76    private ClusterId clusterId;
77    // Keep around for convenience.
78    private final FileSystem fs;
79    // Is the fileystem ok?
80    private volatile boolean fsOk = true;
81    // The Path to the old logs dir
82    private final Path oldLogDir;
83    // root hbase directory on the FS
84    private final Path rootdir;
85    // hbase temp directory used for table construction and deletion
86    private final Path tempdir;
87    // create the split log lock
88    final Lock splitLogLock = new ReentrantLock();
89    final boolean distributedLogReplay;
90    final SplitLogManager splitLogManager;
91    private final MasterServices services;
92  
93    final static PathFilter META_FILTER = new PathFilter() {
94      public boolean accept(Path p) {
95        return HLogUtil.isMetaFile(p);
96      }
97    };
98  
99    final static PathFilter NON_META_FILTER = new PathFilter() {
100     public boolean accept(Path p) {
101       return !HLogUtil.isMetaFile(p);
102     }
103   };
104 
105   public MasterFileSystem(Server master, MasterServices services)
106   throws IOException {
107     this.conf = master.getConfiguration();
108     this.master = master;
109     this.services = services;
110     // Set filesystem to be that of this.rootdir else we get complaints about
111     // mismatched filesystems if hbase.rootdir is hdfs and fs.defaultFS is
112     // default localfs.  Presumption is that rootdir is fully-qualified before
113     // we get to here with appropriate fs scheme.
114     this.rootdir = FSUtils.getRootDir(conf);
115     this.tempdir = new Path(this.rootdir, HConstants.HBASE_TEMP_DIRECTORY);
116     // Cover both bases, the old way of setting default fs and the new.
117     // We're supposed to run on 0.20 and 0.21 anyways.
118     this.fs = this.rootdir.getFileSystem(conf);
119     FSUtils.setFsDefault(conf, new Path(this.fs.getUri()));
120     // make sure the fs has the same conf
121     fs.setConf(conf);
122     // setup the filesystem variable
123     // set up the archived logs path
124     this.oldLogDir = createInitialFileSystemLayout();
125     HFileSystem.addLocationsOrderInterceptor(conf);
126     try {
127       this.splitLogManager = new SplitLogManager(master.getZooKeeper(),
128         master.getConfiguration(), master, services,
129         master.getServerName());
130     } catch (KeeperException e) {
131       throw new IOException(e);
132     }
133     this.distributedLogReplay = (this.splitLogManager.getRecoveryMode() == RecoveryMode.LOG_REPLAY);
134   }
135 
136   /**
137    * Create initial layout in filesystem.
138    * <ol>
139    * <li>Check if the meta region exists and is readable, if not create it.
140    * Create hbase.version and the hbase:meta directory if not one.
141    * </li>
142    * <li>Create a log archive directory for RS to put archived logs</li>
143    * </ol>
144    * Idempotent.
145    */
146   private Path createInitialFileSystemLayout() throws IOException {
147     // check if the root directory exists
148     checkRootDir(this.rootdir, conf, this.fs);
149 
150     // check if temp directory exists and clean it
151     checkTempDir(this.tempdir, conf, this.fs);
152 
153     Path oldLogDir = new Path(this.rootdir, HConstants.HREGION_OLDLOGDIR_NAME);
154 
155     // Make sure the region servers can archive their old logs
156     if(!this.fs.exists(oldLogDir)) {
157       this.fs.mkdirs(oldLogDir);
158     }
159 
160     return oldLogDir;
161   }
162 
163   public FileSystem getFileSystem() {
164     return this.fs;
165   }
166 
167   /**
168    * Get the directory where old logs go
169    * @return the dir
170    */
171   public Path getOldLogDir() {
172     return this.oldLogDir;
173   }
174 
175   /**
176    * Checks to see if the file system is still accessible.
177    * If not, sets closed
178    * @return false if file system is not available
179    */
180   public boolean checkFileSystem() {
181     if (this.fsOk) {
182       try {
183         FSUtils.checkFileSystemAvailable(this.fs);
184         FSUtils.checkDfsSafeMode(this.conf);
185       } catch (IOException e) {
186         master.abort("Shutting down HBase cluster: file system not available", e);
187         this.fsOk = false;
188       }
189     }
190     return this.fsOk;
191   }
192 
193   /**
194    * @return HBase root dir.
195    */
196   public Path getRootDir() {
197     return this.rootdir;
198   }
199 
200   /**
201    * @return HBase temp dir.
202    */
203   public Path getTempDir() {
204     return this.tempdir;
205   }
206 
207   /**
208    * @return The unique identifier generated for this cluster
209    */
210   public ClusterId getClusterId() {
211     return clusterId;
212   }
213 
214   /**
215    * Inspect the log directory to find dead servers which need recovery work
216    * @return A set of ServerNames which aren't running but still have WAL files left in file system
217    */
218   Set<ServerName> getFailedServersFromLogFolders() {
219     boolean retrySplitting = !conf.getBoolean("hbase.hlog.split.skip.errors",
220       HLog.SPLIT_SKIP_ERRORS_DEFAULT);
221 
222     Set<ServerName> serverNames = new HashSet<ServerName>();
223     Path logsDirPath = new Path(this.rootdir, HConstants.HREGION_LOGDIR_NAME);
224 
225     do {
226       if (master.isStopped()) {
227         LOG.warn("Master stopped while trying to get failed servers.");
228         break;
229       }
230       try {
231         if (!this.fs.exists(logsDirPath)) return serverNames;
232         FileStatus[] logFolders = FSUtils.listStatus(this.fs, logsDirPath, null);
233         // Get online servers after getting log folders to avoid log folder deletion of newly
234         // checked in region servers . see HBASE-5916
235         Set<ServerName> onlineServers = ((HMaster) master).getServerManager().getOnlineServers()
236             .keySet();
237 
238         if (logFolders == null || logFolders.length == 0) {
239           LOG.debug("No log files to split, proceeding...");
240           return serverNames;
241         }
242         for (FileStatus status : logFolders) {
243           String sn = status.getPath().getName();
244           // truncate splitting suffix if present (for ServerName parsing)
245           if (sn.endsWith(HLog.SPLITTING_EXT)) {
246             sn = sn.substring(0, sn.length() - HLog.SPLITTING_EXT.length());
247           }
248           ServerName serverName = ServerName.parseServerName(sn);
249           if (!onlineServers.contains(serverName)) {
250             LOG.info("Log folder " + status.getPath() + " doesn't belong "
251                 + "to a known region server, splitting");
252             serverNames.add(serverName);
253           } else {
254             LOG.info("Log folder " + status.getPath() + " belongs to an existing region server");
255           }
256         }
257         retrySplitting = false;
258       } catch (IOException ioe) {
259         LOG.warn("Failed getting failed servers to be recovered.", ioe);
260         if (!checkFileSystem()) {
261           LOG.warn("Bad Filesystem, exiting");
262           Runtime.getRuntime().halt(1);
263         }
264         try {
265           if (retrySplitting) {
266             Thread.sleep(conf.getInt("hbase.hlog.split.failure.retry.interval", 30 * 1000));
267           }
268         } catch (InterruptedException e) {
269           LOG.warn("Interrupted, aborting since cannot return w/o splitting");
270           Thread.currentThread().interrupt();
271           retrySplitting = false;
272           Runtime.getRuntime().halt(1);
273         }
274       }
275     } while (retrySplitting);
276 
277     return serverNames;
278   }
279 
280   public void splitLog(final ServerName serverName) throws IOException {
281     Set<ServerName> serverNames = new HashSet<ServerName>();
282     serverNames.add(serverName);
283     splitLog(serverNames);
284   }
285 
286   /**
287    * Specialized method to handle the splitting for meta HLog
288    * @param serverName
289    * @throws IOException
290    */
291   public void splitMetaLog(final ServerName serverName) throws IOException {
292     Set<ServerName> serverNames = new HashSet<ServerName>();
293     serverNames.add(serverName);
294     splitMetaLog(serverNames);
295   }
296 
297   /**
298    * Specialized method to handle the splitting for meta HLog
299    * @param serverNames
300    * @throws IOException
301    */
302   public void splitMetaLog(final Set<ServerName> serverNames) throws IOException {
303     splitLog(serverNames, META_FILTER);
304   }
305 
306   private List<Path> getLogDirs(final Set<ServerName> serverNames) throws IOException {
307     List<Path> logDirs = new ArrayList<Path>();
308     boolean needReleaseLock = false;
309     if (!this.services.isInitialized()) {
310       // during master initialization, we could have multiple places splitting a same wal
311       this.splitLogLock.lock();
312       needReleaseLock = true;
313     }
314     try {
315       for (ServerName serverName : serverNames) {
316         Path logDir = new Path(this.rootdir, HLogUtil.getHLogDirectoryName(serverName.toString()));
317         Path splitDir = logDir.suffix(HLog.SPLITTING_EXT);
318         // Rename the directory so a rogue RS doesn't create more HLogs
319         if (fs.exists(logDir)) {
320           if (!this.fs.rename(logDir, splitDir)) {
321             throw new IOException("Failed fs.rename for log split: " + logDir);
322           }
323           logDir = splitDir;
324           LOG.debug("Renamed region directory: " + splitDir);
325         } else if (!fs.exists(splitDir)) {
326           LOG.info("Log dir for server " + serverName + " does not exist");
327           continue;
328         }
329         logDirs.add(splitDir);
330       }
331     } finally {
332       if (needReleaseLock) {
333         this.splitLogLock.unlock();
334       }
335     }
336     return logDirs;
337   }
338 
339   /**
340    * Mark regions in recovering state when distributedLogReplay are set true
341    * @param serverName Failed region server whose wals to be replayed
342    * @param regions Set of regions to be recovered
343    * @throws IOException
344    */
345   public void prepareLogReplay(ServerName serverName, Set<HRegionInfo> regions) throws IOException {
346     if (!this.distributedLogReplay) {
347       return;
348     }
349     // mark regions in recovering state
350     if (regions == null || regions.isEmpty()) {
351       return;
352     }
353     try {
354       this.splitLogManager.markRegionsRecoveringInZK(serverName, regions);
355     } catch (KeeperException e) {
356       throw new IOException(e);
357     }
358   }
359 
360   public void splitLog(final Set<ServerName> serverNames) throws IOException {
361     splitLog(serverNames, NON_META_FILTER);
362   }
363 
364   /**
365    * Wrapper function on {@link SplitLogManager#removeStaleRecoveringRegionsFromZK(Set)}
366    * @param failedServers
367    * @throws KeeperException
368    */
369   void removeStaleRecoveringRegionsFromZK(final Set<ServerName> failedServers)
370       throws KeeperException, InterruptedIOException {
371     this.splitLogManager.removeStaleRecoveringRegionsFromZK(failedServers);
372   }
373 
374   /**
375    * This method is the base split method that splits HLog files matching a filter. Callers should
376    * pass the appropriate filter for meta and non-meta HLogs.
377    * @param serverNames
378    * @param filter
379    * @throws IOException
380    */
381   public void splitLog(final Set<ServerName> serverNames, PathFilter filter) throws IOException {
382     long splitTime = 0, splitLogSize = 0;
383     List<Path> logDirs = getLogDirs(serverNames);
384 
385     splitLogManager.handleDeadWorkers(serverNames);
386     splitTime = EnvironmentEdgeManager.currentTimeMillis();
387     splitLogSize = splitLogManager.splitLogDistributed(serverNames, logDirs, filter);
388     splitTime = EnvironmentEdgeManager.currentTimeMillis() - splitTime;
389 
390     if (this.metricsMasterFilesystem != null) {
391       if (filter == META_FILTER) {
392         this.metricsMasterFilesystem.addMetaWALSplit(splitTime, splitLogSize);
393       } else {
394         this.metricsMasterFilesystem.addSplit(splitTime, splitLogSize);
395       }
396     }
397   }
398 
399   /**
400    * Get the rootdir.  Make sure its wholesome and exists before returning.
401    * @param rd
402    * @param c
403    * @param fs
404    * @return hbase.rootdir (after checks for existence and bootstrapping if
405    * needed populating the directory with necessary bootup files).
406    * @throws IOException
407    */
408   @SuppressWarnings("deprecation")
409   private Path checkRootDir(final Path rd, final Configuration c,
410     final FileSystem fs)
411   throws IOException {
412     // If FS is in safe mode wait till out of it.
413     FSUtils.waitOnSafeMode(c, c.getInt(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000));
414     // Filesystem is good. Go ahead and check for hbase.rootdir.
415     try {
416       if (!fs.exists(rd)) {
417         fs.mkdirs(rd);
418         // DFS leaves safe mode with 0 DNs when there are 0 blocks.
419         // We used to handle this by checking the current DN count and waiting until
420         // it is nonzero. With security, the check for datanode count doesn't work --
421         // it is a privileged op. So instead we adopt the strategy of the jobtracker
422         // and simply retry file creation during bootstrap indefinitely. As soon as
423         // there is one datanode it will succeed. Permission problems should have
424         // already been caught by mkdirs above.
425         FSUtils.setVersion(fs, rd, c.getInt(HConstants.THREAD_WAKE_FREQUENCY,
426           10 * 1000), c.getInt(HConstants.VERSION_FILE_WRITE_ATTEMPTS,
427             HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS));
428       } else {
429         if (!fs.isDirectory(rd)) {
430           throw new IllegalArgumentException(rd.toString() + " is not a directory");
431         }
432         // as above
433         FSUtils.checkVersion(fs, rd, true, c.getInt(HConstants.THREAD_WAKE_FREQUENCY,
434           10 * 1000), c.getInt(HConstants.VERSION_FILE_WRITE_ATTEMPTS,
435             HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS));
436       }
437     } catch (DeserializationException de) {
438       LOG.fatal("Please fix invalid configuration for " + HConstants.HBASE_DIR, de);
439       IOException ioe = new IOException();
440       ioe.initCause(de);
441       throw ioe;
442     } catch (IllegalArgumentException iae) {
443       LOG.fatal("Please fix invalid configuration for "
444         + HConstants.HBASE_DIR + " " + rd.toString(), iae);
445       throw iae;
446     }
447     // Make sure cluster ID exists
448     if (!FSUtils.checkClusterIdExists(fs, rd, c.getInt(
449         HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000))) {
450       FSUtils.setClusterId(fs, rd, new ClusterId(), c.getInt(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000));
451     }
452     clusterId = FSUtils.getClusterId(fs, rd);
453 
454     // Make sure the meta region directory exists!
455     if (!FSUtils.metaRegionExists(fs, rd)) {
456       bootstrap(rd, c);
457     } else {
458       // Migrate table descriptor files if necessary
459       org.apache.hadoop.hbase.util.FSTableDescriptorMigrationToSubdir
460         .migrateFSTableDescriptorsIfNecessary(fs, rd);
461     }
462       
463     // Create tableinfo-s for hbase:meta if not already there.
464     new FSTableDescriptors(fs, rd).createTableDescriptor(HTableDescriptor.META_TABLEDESC);
465 
466     return rd;
467   }
468 
469   /**
470    * Make sure the hbase temp directory exists and is empty.
471    * NOTE that this method is only executed once just after the master becomes the active one.
472    */
473   private void checkTempDir(final Path tmpdir, final Configuration c, final FileSystem fs)
474       throws IOException {
475     // If the temp directory exists, clear the content (left over, from the previous run)
476     if (fs.exists(tmpdir)) {
477       // Archive table in temp, maybe left over from failed deletion,
478       // if not the cleaner will take care of them.
479       for (Path tabledir: FSUtils.getTableDirs(fs, tmpdir)) {
480         for (Path regiondir: FSUtils.getRegionDirs(fs, tabledir)) {
481           HFileArchiver.archiveRegion(fs, this.rootdir, tabledir, regiondir);
482         }
483       }
484       if (!fs.delete(tmpdir, true)) {
485         throw new IOException("Unable to clean the temp directory: " + tmpdir);
486       }
487     }
488 
489     // Create the temp directory
490     if (!fs.mkdirs(tmpdir)) {
491       throw new IOException("HBase temp directory '" + tmpdir + "' creation failure.");
492     }
493   }
494 
495   private static void bootstrap(final Path rd, final Configuration c)
496   throws IOException {
497     LOG.info("BOOTSTRAP: creating hbase:meta region");
498     try {
499       // Bootstrapping, make sure blockcache is off.  Else, one will be
500       // created here in bootstrap and it'll need to be cleaned up.  Better to
501       // not make it in first place.  Turn off block caching for bootstrap.
502       // Enable after.
503       HRegionInfo metaHRI = new HRegionInfo(HRegionInfo.FIRST_META_REGIONINFO);
504       setInfoFamilyCachingForMeta(false);
505       HRegion meta = HRegion.createHRegion(metaHRI, rd, c,
506           HTableDescriptor.META_TABLEDESC);
507       setInfoFamilyCachingForMeta(true);
508       HRegion.closeHRegion(meta);
509     } catch (IOException e) {
510         e = e instanceof RemoteException ?
511                 ((RemoteException)e).unwrapRemoteException() : e;
512       LOG.error("bootstrap", e);
513       throw e;
514     }
515   }
516 
517   /**
518    * Enable in memory caching for hbase:meta
519    */
520   public static void setInfoFamilyCachingForMeta(final boolean b) {
521     for (HColumnDescriptor hcd:
522         HTableDescriptor.META_TABLEDESC.getColumnFamilies()) {
523       if (Bytes.equals(hcd.getName(), HConstants.CATALOG_FAMILY)) {
524         hcd.setBlockCacheEnabled(b);
525         hcd.setInMemory(b);
526       }
527     }
528   }
529 
530 
531   public void deleteRegion(HRegionInfo region) throws IOException {
532     HFileArchiver.archiveRegion(conf, fs, region);
533   }
534 
535   public void deleteTable(TableName tableName) throws IOException {
536     fs.delete(FSUtils.getTableDir(rootdir, tableName), true);
537   }
538 
539   /**
540    * Move the specified table to the hbase temp directory
541    * @param tableName Table name to move
542    * @return The temp location of the table moved
543    * @throws IOException in case of file-system failure
544    */
545   public Path moveTableToTemp(TableName tableName) throws IOException {
546     Path srcPath = FSUtils.getTableDir(rootdir, tableName);
547     Path tempPath = FSUtils.getTableDir(this.tempdir, tableName);
548 
549     // Ensure temp exists
550     if (!fs.exists(tempPath.getParent()) && !fs.mkdirs(tempPath.getParent())) {
551       throw new IOException("HBase temp directory '" + tempPath.getParent() + "' creation failure.");
552     }
553 
554     if (!fs.rename(srcPath, tempPath)) {
555       throw new IOException("Unable to move '" + srcPath + "' to temp '" + tempPath + "'");
556     }
557 
558     return tempPath;
559   }
560 
561   public void updateRegionInfo(HRegionInfo region) {
562     // TODO implement this.  i think this is currently broken in trunk i don't
563     //      see this getting updated.
564     //      @see HRegion.checkRegioninfoOnFilesystem()
565   }
566 
567   public void deleteFamilyFromFS(HRegionInfo region, byte[] familyName)
568       throws IOException {
569     // archive family store files
570     Path tableDir = FSUtils.getTableDir(rootdir, region.getTable());
571     HFileArchiver.archiveFamily(fs, conf, region, tableDir, familyName);
572 
573     // delete the family folder
574     Path familyDir = new Path(tableDir,
575       new Path(region.getEncodedName(), Bytes.toString(familyName)));
576     if (fs.delete(familyDir, true) == false) {
577       throw new IOException("Could not delete family "
578           + Bytes.toString(familyName) + " from FileSystem for region "
579           + region.getRegionNameAsString() + "(" + region.getEncodedName()
580           + ")");
581     }
582   }
583 
584   public void stop() {
585     if (splitLogManager != null) {
586       this.splitLogManager.stop();
587     }
588   }
589 
590   /**
591    * Delete column of a table
592    * @param tableName
593    * @param familyName
594    * @return Modified HTableDescriptor with requested column deleted.
595    * @throws IOException
596    */
597   public HTableDescriptor deleteColumn(TableName tableName, byte[] familyName)
598       throws IOException {
599     LOG.info("DeleteColumn. Table = " + tableName
600         + " family = " + Bytes.toString(familyName));
601     HTableDescriptor htd = this.services.getTableDescriptors().get(tableName);
602     htd.removeFamily(familyName);
603     this.services.getTableDescriptors().add(htd);
604     return htd;
605   }
606 
607   /**
608    * Modify Column of a table
609    * @param tableName
610    * @param hcd HColumnDesciptor
611    * @return Modified HTableDescriptor with the column modified.
612    * @throws IOException
613    */
614   public HTableDescriptor modifyColumn(TableName tableName, HColumnDescriptor hcd)
615       throws IOException {
616     LOG.info("AddModifyColumn. Table = " + tableName
617         + " HCD = " + hcd.toString());
618 
619     HTableDescriptor htd = this.services.getTableDescriptors().get(tableName);
620     byte [] familyName = hcd.getName();
621     if(!htd.hasFamily(familyName)) {
622       throw new InvalidFamilyOperationException("Family '" +
623         Bytes.toString(familyName) + "' doesn't exists so cannot be modified");
624     }
625     htd.addFamily(hcd);
626     this.services.getTableDescriptors().add(htd);
627     return htd;
628   }
629 
630   /**
631    * Add column to a table
632    * @param tableName
633    * @param hcd
634    * @return Modified HTableDescriptor with new column added.
635    * @throws IOException
636    */
637   public HTableDescriptor addColumn(TableName tableName, HColumnDescriptor hcd)
638       throws IOException {
639     LOG.info("AddColumn. Table = " + tableName + " HCD = " +
640       hcd.toString());
641     HTableDescriptor htd = this.services.getTableDescriptors().get(tableName);
642     if (htd == null) {
643       throw new InvalidFamilyOperationException("Family '" +
644         hcd.getNameAsString() + "' cannot be modified as HTD is null");
645     }
646     htd.addFamily(hcd);
647     this.services.getTableDescriptors().add(htd);
648     return htd;
649   }
650 
651   /**
652    * The function is used in SSH to set recovery mode based on configuration after all outstanding
653    * log split tasks drained.
654    * @throws KeeperException
655    * @throws InterruptedIOException
656    */
657   public void setLogRecoveryMode() throws IOException {
658     try {
659       this.splitLogManager.setRecoveryMode(false);
660     } catch (KeeperException e) {
661       throw new IOException(e);
662     }
663   }
664 
665   public RecoveryMode getLogRecoveryMode() {
666     return this.splitLogManager.getRecoveryMode();
667   }
668 }