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