View Javadoc

1   /**
2    * Copyright 2010 The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  package org.apache.hadoop.hbase.master;
21  
22  import java.io.IOException;
23  import java.util.ArrayList;
24  import java.util.HashSet;
25  import java.util.List;
26  import java.util.Set;
27  import java.util.UUID;
28  import java.util.concurrent.locks.Lock;
29  import java.util.concurrent.locks.ReentrantLock;
30  
31  import org.apache.commons.logging.Log;
32  import org.apache.commons.logging.LogFactory;
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.HBaseFileSystem;
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.RemoteExceptionHandler;
45  import org.apache.hadoop.hbase.Server;
46  import org.apache.hadoop.hbase.ServerName;
47  import org.apache.hadoop.hbase.backup.HFileArchiver;
48  import org.apache.hadoop.hbase.master.metrics.MasterMetrics;
49  import org.apache.hadoop.hbase.regionserver.HRegion;
50  import org.apache.hadoop.hbase.regionserver.wal.HLog;
51  import org.apache.hadoop.hbase.regionserver.wal.HLogSplitter;
52  import org.apache.hadoop.hbase.regionserver.wal.OrphanHLogAfterSplitException;
53  import org.apache.hadoop.hbase.util.Bytes;
54  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
55  import org.apache.hadoop.hbase.util.FSTableDescriptors;
56  import org.apache.hadoop.hbase.util.FSUtils;
57  
58  /**
59   * This class abstracts a bunch of operations the HMaster needs to interact with
60   * the underlying file system, including splitting log files, checking file
61   * system status, etc.
62   */
63  public class MasterFileSystem {
64    private static final Log LOG = LogFactory.getLog(MasterFileSystem.class.getName());
65    // HBase configuration
66    Configuration conf;
67    // master status
68    Server master;
69    // metrics for master
70    MasterMetrics metrics;
71    // Persisted unique cluster ID
72    private String clusterId;
73    // Keep around for convenience.
74    private final FileSystem fs;
75    // Is the fileystem ok?
76    private volatile boolean fsOk = true;
77    // The Path to the old logs dir
78    private final Path oldLogDir;
79    // root hbase directory on the FS
80    private final Path rootdir;
81    // hbase temp directory used for table construction and deletion
82    private final Path tempdir;
83    // create the split log lock
84    final Lock splitLogLock = new ReentrantLock();
85    final boolean distributedLogSplitting;
86    final SplitLogManager splitLogManager;
87    private final MasterServices services;
88  
89    private final static PathFilter META_FILTER = new PathFilter() {
90      public boolean accept(Path p) {
91        return HLog.isMetaFile(p);
92      }
93    };
94  
95    private final static PathFilter NON_META_FILTER = new PathFilter() {
96      public boolean accept(Path p) {
97        return !HLog.isMetaFile(p);
98      }
99    };
100 
101   public MasterFileSystem(Server master, MasterServices services,
102       MasterMetrics metrics, boolean masterRecovery)
103   throws IOException {
104     this.conf = master.getConfiguration();
105     this.master = master;
106     this.services = services;
107     this.metrics = metrics;
108     // Set filesystem to be that of this.rootdir else we get complaints about
109     // mismatched filesystems if hbase.rootdir is hdfs and fs.defaultFS is
110     // default localfs.  Presumption is that rootdir is fully-qualified before
111     // we get to here with appropriate fs scheme.
112     this.rootdir = FSUtils.getRootDir(conf);
113     this.tempdir = new Path(this.rootdir, HConstants.HBASE_TEMP_DIRECTORY);
114     // Cover both bases, the old way of setting default fs and the new.
115     // We're supposed to run on 0.20 and 0.21 anyways.
116     this.fs = this.rootdir.getFileSystem(conf);
117     String fsUri = this.fs.getUri().toString();
118     conf.set("fs.default.name", fsUri);
119     conf.set("fs.defaultFS", fsUri);
120     // make sure the fs has the same conf
121     fs.setConf(conf);
122     this.distributedLogSplitting =
123       conf.getBoolean("hbase.master.distributed.log.splitting", true);
124     if (this.distributedLogSplitting) {
125       this.splitLogManager = new SplitLogManager(master.getZooKeeper(),
126           master.getConfiguration(), master, this.services, master.getServerName().toString());
127       this.splitLogManager.finishInitialization(masterRecovery);
128     } else {
129       this.splitLogManager = null;
130     }
131     // setup the filesystem variable
132     // set up the archived logs path
133     this.oldLogDir = createInitialFileSystemLayout();
134   }
135 
136   /**
137    * Create initial layout in filesystem.
138    * <ol>
139    * <li>Check if the root region exists and is readable, if not create it.
140    * Create hbase.version and the -ROOT- 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       HBaseFileSystem.makeDirOnFileSystem(fs, 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 String getClusterId() {
211     return clusterId;
212   }
213 
214   /**
215    * Inspect the log directory to find dead servers which need log splitting
216    */
217   Set<ServerName> getFailedServersFromLogFolders() {
218     boolean retrySplitting = !conf.getBoolean("hbase.hlog.split.skip.errors",
219       HLog.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 =
235             ((HMaster) master).getServerManager().getOnlineServers().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           String sn = status.getPath().getName();
243           // truncate splitting suffix if present (for ServerName parsing)
244           if (sn.endsWith(HLog.SPLITTING_EXT)) {
245             sn = sn.substring(0, sn.length() - HLog.SPLITTING_EXT.length());
246           }
247           ServerName serverName = ServerName.parseServerName(sn);
248           if (!onlineServers.contains(serverName)) {
249             LOG.info("Log folder " + status.getPath() + " doesn't belong "
250                 + "to a known region server, splitting");
251             serverNames.add(serverName);
252           } else {
253             LOG.info("Log folder " + status.getPath() + " belongs to an existing region server");
254           }
255         }
256         retrySplitting = false;
257       } catch (IOException ioe) {
258         LOG.warn("Failed getting failed servers to be recovered.", ioe);
259         if (!checkFileSystem()) {
260           LOG.warn("Bad Filesystem, exiting");
261           Runtime.getRuntime().halt(1);
262         }
263         try {
264           if (retrySplitting) {
265             Thread.sleep(conf.getInt("hbase.hlog.split.failure.retry.interval", 30 * 1000));
266           }
267         } catch (InterruptedException e) {
268           LOG.warn("Interrupted, aborting since cannot return w/o splitting");
269           Thread.currentThread().interrupt();
270           retrySplitting = false;
271           Runtime.getRuntime().halt(1);
272         }
273       }
274     } while (retrySplitting);
275 
276     return serverNames;
277   }
278 
279   public void splitLog(final ServerName serverName) throws IOException {
280     List<ServerName> serverNames = new ArrayList<ServerName>();
281     serverNames.add(serverName);
282     splitLog(serverNames);
283   }
284 
285   public void splitAllLogs(final ServerName serverName) throws IOException {
286     List<ServerName> serverNames = new ArrayList<ServerName>();
287     serverNames.add(serverName);
288     splitAllLogs(serverNames);
289   }
290 
291   /**
292    * Specialized method to handle the splitting for meta HLog
293    * @param serverName
294    * @throws IOException
295    */
296   public void splitMetaLog(final ServerName serverName) throws IOException {
297     long splitTime = 0, splitLogSize = 0;
298     List<ServerName> serverNames = new ArrayList<ServerName>();
299     serverNames.add(serverName);
300     List<Path> logDirs = getLogDirs(serverNames);
301     if (logDirs.isEmpty()) {
302       LOG.info("No meta logs to split");
303       return;
304     }
305     splitLogManager.handleDeadWorkers(serverNames);
306     splitTime = EnvironmentEdgeManager.currentTimeMillis();
307     splitLogSize = splitLogManager.splitLogDistributed(logDirs, META_FILTER);
308     splitTime = EnvironmentEdgeManager.currentTimeMillis() - splitTime;
309     if (this.metrics != null) {
310       this.metrics.addSplit(splitTime, splitLogSize);
311     }
312   }
313 
314   private List<Path> getLogDirs(final List<ServerName> serverNames) throws IOException {
315     List<Path> logDirs = new ArrayList<Path>();
316     for(ServerName serverName: serverNames){
317       Path logDir = new Path(this.rootdir,
318         HLog.getHLogDirectoryName(serverName.toString()));
319       Path splitDir = logDir.suffix(HLog.SPLITTING_EXT);
320       // rename the directory so a rogue RS doesn't create more HLogs
321       if (fs.exists(logDir)) {
322         if (!HBaseFileSystem.renameDirForFileSystem(fs, logDir, splitDir)) {
323           throw new IOException("Failed fs.rename for log split: " + logDir);
324         }
325         logDir = splitDir;
326         LOG.debug("Renamed region directory: " + splitDir);
327       } else if (!fs.exists(splitDir)) {
328         LOG.info("Log dir for server " + serverName + " does not exist");
329         continue;
330       }
331       logDirs.add(splitDir);
332     }
333     return logDirs;
334   }
335 
336   public void splitLog(final List<ServerName> serverNames) throws IOException {
337     splitLog(serverNames, NON_META_FILTER);
338   }
339 
340   public void splitAllLogs(final List<ServerName> serverNames) throws IOException {
341     splitLog(serverNames, null); //no filter
342   }
343 
344   /**
345    * This method is the base split method that splits HLog files matching a filter.
346    * Callers should pass the appropriate filter for meta and non-meta HLogs.
347    * @param serverNames
348    * @param filter
349    * @throws IOException
350    */
351   public void splitLog(final List<ServerName> serverNames, PathFilter filter) throws IOException {
352     long splitTime = 0, splitLogSize = 0;
353     List<Path> logDirs = getLogDirs(serverNames);
354 
355     if (logDirs.isEmpty()) {
356       LOG.info("No logs to split");
357       return;
358     }
359 
360     boolean lockAcquired = false;
361     if (distributedLogSplitting) {
362       try {
363         if (!this.services.isServerShutdownHandlerEnabled()) {
364           // process one log splitting task at one time before SSH is enabled.
365           // because ROOT SSH and HMaster#assignMeta could both log split a same server
366           this.splitLogLock.lock();
367           lockAcquired = true;
368         }
369         splitLogManager.handleDeadWorkers(serverNames);
370         splitTime = EnvironmentEdgeManager.currentTimeMillis();
371         splitLogSize = splitLogManager.splitLogDistributed(logDirs, filter);
372         splitTime = EnvironmentEdgeManager.currentTimeMillis() - splitTime;
373       } finally {
374         if (lockAcquired) {
375           this.splitLogLock.unlock();
376         }
377       }
378     } else {
379       for(Path logDir: logDirs){
380         // splitLogLock ensures that dead region servers' logs are processed
381         // one at a time
382         this.splitLogLock.lock();
383         try {
384           HLogSplitter splitter = HLogSplitter.createLogSplitter(
385             conf, rootdir, logDir, oldLogDir, this.fs);
386           try {
387             // If FS is in safe mode, just wait till out of it.
388             FSUtils.waitOnSafeMode(conf, conf.getInt(HConstants.THREAD_WAKE_FREQUENCY, 1000));
389             splitter.splitLog();
390           } catch (OrphanHLogAfterSplitException e) {
391             LOG.warn("Retrying splitting because of:", e);
392             //An HLogSplitter instance can only be used once.  Get new instance.
393             splitter = HLogSplitter.createLogSplitter(conf, rootdir, logDir,
394               oldLogDir, this.fs);
395             splitter.splitLog();
396           }
397           splitTime = splitter.getTime();
398           splitLogSize = splitter.getSize();
399         } finally {
400           this.splitLogLock.unlock();
401         }
402       }
403     }
404 
405     if (this.metrics != null) {
406       this.metrics.addSplit(splitTime, splitLogSize);
407     }
408   }
409 
410   /**
411    * Get the rootdir.  Make sure its wholesome and exists before returning.
412    * @param rd
413    * @param conf
414    * @param fs
415    * @return hbase.rootdir (after checks for existence and bootstrapping if
416    * needed populating the directory with necessary bootup files).
417    * @throws IOException
418    */
419   private Path checkRootDir(final Path rd, final Configuration c,
420     final FileSystem fs)
421   throws IOException {
422     // If FS is in safe mode wait till out of it.
423     FSUtils.waitOnSafeMode(c, c.getInt(HConstants.THREAD_WAKE_FREQUENCY,
424         10 * 1000));
425     // Filesystem is good. Go ahead and check for hbase.rootdir.
426     try {
427       if (!fs.exists(rd)) {
428         HBaseFileSystem.makeDirOnFileSystem(fs, rd);
429         // DFS leaves safe mode with 0 DNs when there are 0 blocks.
430         // We used to handle this by checking the current DN count and waiting until
431         // it is nonzero. With security, the check for datanode count doesn't work --
432         // it is a privileged op. So instead we adopt the strategy of the jobtracker
433         // and simply retry file creation during bootstrap indefinitely. As soon as
434         // there is one datanode it will succeed. Permission problems should have
435         // already been caught by mkdirs above.
436         FSUtils.setVersion(fs, rd, c.getInt(HConstants.THREAD_WAKE_FREQUENCY,
437           10 * 1000), c.getInt(HConstants.VERSION_FILE_WRITE_ATTEMPTS,
438         		  HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS));
439       } else {
440         if (!fs.isDirectory(rd)) {
441           throw new IllegalArgumentException(rd.toString() + " is not a directory");
442         }
443         // as above
444         FSUtils.checkVersion(fs, rd, true, c.getInt(HConstants.THREAD_WAKE_FREQUENCY,
445           10 * 1000), c.getInt(HConstants.VERSION_FILE_WRITE_ATTEMPTS,
446         		  HConstants.DEFAULT_VERSION_FILE_WRITE_ATTEMPTS));
447       }
448     } catch (IllegalArgumentException iae) {
449       LOG.fatal("Please fix invalid configuration for "
450         + HConstants.HBASE_DIR + " " + rd.toString(), iae);
451       throw iae;
452     }
453     // Make sure cluster ID exists
454     if (!FSUtils.checkClusterIdExists(fs, rd, c.getInt(
455         HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000))) {
456       FSUtils.setClusterId(fs, rd, UUID.randomUUID().toString(), c.getInt(
457           HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000));
458     }
459     clusterId = FSUtils.getClusterId(fs, rd);
460 
461     // Make sure the root region directory exists!
462     if (!FSUtils.rootRegionExists(fs, rd)) {
463       bootstrap(rd, c);
464     }
465     createRootTableInfo(rd);
466     return rd;
467   }
468 
469   private void createRootTableInfo(Path rd) throws IOException {
470     // Create ROOT tableInfo if required.
471     if (!FSTableDescriptors.isTableInfoExists(fs, rd,
472         Bytes.toString(HRegionInfo.ROOT_REGIONINFO.getTableName()))) {
473       FSTableDescriptors.createTableDescriptor(HTableDescriptor.ROOT_TABLEDESC, this.conf);
474     }
475   }
476 
477   /**
478    * Make sure the hbase temp directory exists and is empty.
479    * NOTE that this method is only executed once just after the master becomes the active one.
480    */
481   private void checkTempDir(final Path tmpdir, final Configuration c, final FileSystem fs)
482       throws IOException {
483     // If the temp directory exists, clear the content (left over, from the previous run)
484     if (fs.exists(tmpdir)) {
485       // Archive table in temp, maybe left over from failed deletion,
486       // if not the cleaner will take care of them.
487       for (Path tabledir: FSUtils.getTableDirs(fs, tmpdir)) {
488         for (Path regiondir: FSUtils.getRegionDirs(fs, tabledir)) {
489           HFileArchiver.archiveRegion(fs, this.rootdir, tabledir, regiondir);
490         }
491       }
492       if (!HBaseFileSystem.deleteDirFromFileSystem(fs, tmpdir)) {
493         throw new IOException("Unable to clean the temp directory: " + tmpdir);
494       }
495     }
496 
497     // Create the temp directory
498     if (!HBaseFileSystem.makeDirOnFileSystem(fs, tmpdir)) {
499       throw new IOException("HBase temp directory '" + tmpdir + "' creation failure.");
500     }
501   }
502 
503   private static void bootstrap(final Path rd, final Configuration c)
504   throws IOException {
505     LOG.info("BOOTSTRAP: creating ROOT and first META regions");
506     try {
507       // Bootstrapping, make sure blockcache is off.  Else, one will be
508       // created here in bootstap and it'll need to be cleaned up.  Better to
509       // not make it in first place.  Turn off block caching for bootstrap.
510       // Enable after.
511       HRegionInfo rootHRI = new HRegionInfo(HRegionInfo.ROOT_REGIONINFO);
512       setInfoFamilyCachingForRoot(false);
513       HRegionInfo metaHRI = new HRegionInfo(HRegionInfo.FIRST_META_REGIONINFO);
514       setInfoFamilyCachingForMeta(false);
515       HRegion root = HRegion.createHRegion(rootHRI, rd, c,
516           HTableDescriptor.ROOT_TABLEDESC);
517       HRegion meta = HRegion.createHRegion(metaHRI, rd, c,
518           HTableDescriptor.META_TABLEDESC);
519       setInfoFamilyCachingForRoot(true);
520       setInfoFamilyCachingForMeta(true);
521       // Add first region from the META table to the ROOT region.
522       HRegion.addRegionToMETA(root, meta);
523       root.close();
524       root.getLog().closeAndDelete();
525       meta.close();
526       meta.getLog().closeAndDelete();
527     } catch (IOException e) {
528       e = RemoteExceptionHandler.checkIOException(e);
529       LOG.error("bootstrap", e);
530       throw e;
531     }
532   }
533 
534   /**
535    * Enable in-memory caching for -ROOT-
536    */
537   public static void setInfoFamilyCachingForRoot(final boolean b) {
538     for (HColumnDescriptor hcd:
539         HTableDescriptor.ROOT_TABLEDESC.getColumnFamilies()) {
540        if (Bytes.equals(hcd.getName(), HConstants.CATALOG_FAMILY)) {
541          hcd.setBlockCacheEnabled(b);
542          hcd.setInMemory(b);
543      }
544     }
545   }
546 
547   /**
548    * Enable in memory caching for .META.
549    */
550   public static void setInfoFamilyCachingForMeta(final boolean b) {
551     for (HColumnDescriptor hcd:
552         HTableDescriptor.META_TABLEDESC.getColumnFamilies()) {
553       if (Bytes.equals(hcd.getName(), HConstants.CATALOG_FAMILY)) {
554         hcd.setBlockCacheEnabled(b);
555         hcd.setInMemory(b);
556       }
557     }
558   }
559 
560 
561   public void deleteRegion(HRegionInfo region) throws IOException {
562     HFileArchiver.archiveRegion(conf, fs, region);
563   }
564 
565   public void deleteTable(byte[] tableName) throws IOException {
566     HBaseFileSystem.deleteDirFromFileSystem(fs, new Path(rootdir, Bytes.toString(tableName)));
567   }
568 
569   /**
570    * Move the specified file/directory to the hbase temp directory.
571    * @param path The path of the file/directory to move
572    * @return The temp location of the file/directory moved
573    * @throws IOException in case of file-system failure
574    */
575   public Path moveToTemp(final Path path) throws IOException {
576     Path tempPath = new Path(this.tempdir, path.getName());
577 
578     // Ensure temp exists
579     if (!fs.exists(tempdir) && !HBaseFileSystem.makeDirOnFileSystem(fs, tempdir)) {
580       throw new IOException("HBase temp directory '" + tempdir + "' creation failure.");
581     }
582 
583     if (!HBaseFileSystem.renameDirForFileSystem(fs, path, tempPath)) {
584       throw new IOException("Unable to move '" + path + "' to temp '" + tempPath + "'");
585     }
586 
587     return tempPath;
588   }
589 
590   /**
591    * Move the specified table to the hbase temp directory
592    * @param tableName Table name to move
593    * @return The temp location of the table moved
594    * @throws IOException in case of file-system failure
595    */
596   public Path moveTableToTemp(byte[] tableName) throws IOException {
597     return moveToTemp(HTableDescriptor.getTableDir(this.rootdir, tableName));
598   }
599 
600   public void updateRegionInfo(HRegionInfo region) {
601     // TODO implement this.  i think this is currently broken in trunk i don't
602     //      see this getting updated.
603     //      @see HRegion.checkRegioninfoOnFilesystem()
604   }
605 
606   public void deleteFamilyFromFS(HRegionInfo region, byte[] familyName)
607       throws IOException {
608     // archive family store files
609     Path tableDir = new Path(rootdir, region.getTableNameAsString());
610     HFileArchiver.archiveFamily(fs, conf, region, tableDir, familyName);
611 
612     // delete the family folder
613     Path familyDir = new Path(tableDir,
614       new Path(region.getEncodedName(), Bytes.toString(familyName)));
615     if (!HBaseFileSystem.deleteDirFromFileSystem(fs, familyDir)) {
616       throw new IOException("Could not delete family "
617           + Bytes.toString(familyName) + " from FileSystem for region "
618           + region.getRegionNameAsString() + "(" + region.getEncodedName()
619           + ")");
620     }
621   }
622 
623   public void stop() {
624     if (splitLogManager != null) {
625       this.splitLogManager.stop();
626     }
627   }
628 
629   /**
630    * Create new HTableDescriptor in HDFS.
631    *
632    * @param htableDescriptor
633    */
634   public void createTableDescriptor(HTableDescriptor htableDescriptor)
635       throws IOException {
636     FSTableDescriptors.createTableDescriptor(htableDescriptor, conf);
637   }
638 
639   /**
640    * Delete column of a table
641    * @param tableName
642    * @param familyName
643    * @return Modified HTableDescriptor with requested column deleted.
644    * @throws IOException
645    */
646   public HTableDescriptor deleteColumn(byte[] tableName, byte[] familyName)
647       throws IOException {
648     LOG.info("DeleteColumn. Table = " + Bytes.toString(tableName)
649         + " family = " + Bytes.toString(familyName));
650     HTableDescriptor htd = this.services.getTableDescriptors().get(tableName);
651     htd.removeFamily(familyName);
652     this.services.getTableDescriptors().add(htd);
653     return htd;
654   }
655 
656   /**
657    * Modify Column of a table
658    * @param tableName
659    * @param hcd HColumnDesciptor
660    * @return Modified HTableDescriptor with the column modified.
661    * @throws IOException
662    */
663   public HTableDescriptor modifyColumn(byte[] tableName, HColumnDescriptor hcd)
664       throws IOException {
665     LOG.info("AddModifyColumn. Table = " + Bytes.toString(tableName)
666         + " HCD = " + hcd.toString());
667 
668     HTableDescriptor htd = this.services.getTableDescriptors().get(tableName);
669     byte [] familyName = hcd.getName();
670     if(!htd.hasFamily(familyName)) {
671       throw new InvalidFamilyOperationException("Family '" +
672         Bytes.toString(familyName) + "' doesn't exists so cannot be modified");
673     }
674     htd.addFamily(hcd);
675     this.services.getTableDescriptors().add(htd);
676     return htd;
677   }
678 
679   /**
680    * Add column to a table
681    * @param tableName
682    * @param hcd
683    * @return Modified HTableDescriptor with new column added.
684    * @throws IOException
685    */
686   public HTableDescriptor addColumn(byte[] tableName, HColumnDescriptor hcd)
687       throws IOException {
688     LOG.info("AddColumn. Table = " + Bytes.toString(tableName) + " HCD = " +
689       hcd.toString());
690     HTableDescriptor htd = this.services.getTableDescriptors().get(tableName);
691     if (htd == null) {
692       throw new InvalidFamilyOperationException("Family '" +
693         hcd.getNameAsString() + "' cannot be modified as HTD is null");
694     }
695     htd.addFamily(hcd);
696     this.services.getTableDescriptors().add(htd);
697     return htd;
698   }
699 }