Improve comments and code for the master

common/src/main/java/tachyon/master/block/BlockId.java

@@ -32,12 +32,10 @@ public final class BlockId {
   private static final long SEQUENCE_NUMBER_MASK = (1L << SEQUENCE_NUMBER_BITS) - 1;
 
   private BlockId() {
-    // util class
+    // prevent instantiation of a util class
   }
 
   /**
-   * Returns a block id with the specified container id and sequence number.
-   *
    * @param containerId the container id to create the block id with
    * @param sequenceNumber the sequence number to create the block id with
    * @return the block id constructed with the container id and sequence number
@@ -49,29 +47,23 @@ public static long createBlockId(long containerId, long sequenceNumber) {
   }
 
   /**
-   * Returns the container id of the specified block id.
-   *
    * @param blockId the block id to get the container id for
-   * @return the container id
+   * @return the container id of a specified block id
    */
   public static long getContainerId(long blockId) {
     return (blockId >> SEQUENCE_NUMBER_BITS) & CONTAINER_ID_MASK;
   }
 
   /**
-   * Returns the sequence number of the specified block id.
-   *
    * @param blockId the block id to get the sequene number for
-   * @return the sequence number
+   * @return the sequence number of the specified block id
    */
   public static long getSequenceNumber(long blockId) {
     return blockId & SEQUENCE_NUMBER_MASK;
   }
 
   /**
-   * Returns the maximum possible sequence number for block ids.
-   *
-   * @return the maximum possible sequence number
+   * @return the maximum possible sequence number for block ids
    */
   public static long getMaxSequenceNumber() {
     return SEQUENCE_NUMBER_MASK;

common/src/thrift/tachyon.thrift

@@ -44,7 +44,6 @@ struct FileBlockInfo {
   4: list<NetAddress> locations
 }
 
-// TODO: make fileId into i64
 struct FileInfo {
   1: i64 fileId
   2: string name

servers/src/main/java/tachyon/master/Master.java

@@ -36,7 +36,7 @@ public interface Master extends JournalCheckpointStreamable {
    *
    * @return a {@link String} representing this master service name.
    */
-  String getProcessorName();
+  String getServiceName();
 
   /**
    * Processes the journal checkpoint file.

servers/src/main/java/tachyon/master/MasterBase.java

@@ -60,15 +60,18 @@ protected MasterBase(Journal journal, ExecutorService executorService) {
   @Override
   public void processJournalCheckpoint(JournalInputStream inputStream) throws IOException {
     JournalEntry entry;
-    while ((entry = inputStream.getNextEntry()) != null) {
-      processJournalEntry(entry);
+    try {
+      while ((entry = inputStream.getNextEntry()) != null) {
+        processJournalEntry(entry);
+      }
+    } finally {
+      inputStream.close();
     }
-    inputStream.close();
   }
 
   @Override
   public void start(boolean isLeader) throws IOException {
-    LOG.info(getProcessorName() + ": Starting master. isLeader: " + isLeader);
+    LOG.info(getServiceName() + ": Starting master. isLeader: " + isLeader);
     mIsLeader = isLeader;
     if (mIsLeader) {
       mJournalWriter = mJournal.getNewWriter();
@@ -96,7 +99,7 @@ public void start(boolean isLeader) throws IOException {
       // Phase 2: Replay all the state of the checkpoint and the completed log files.
       // TODO: only do this if this is a fresh start, not if this master had already been tailing
       // the journal.
-      LOG.info(getProcessorName() + ": process completed logs before becoming master.");
+      LOG.info(getServiceName() + ": process completed logs before becoming master.");
       JournalTailer catchupTailer = new JournalTailer(this, mJournal);
       if (catchupTailer.checkpointExists()) {
         catchupTailer.processJournalCheckpoint(true);
@@ -119,11 +122,12 @@ public void start(boolean isLeader) throws IOException {
 
   @Override
   public void stop() throws IOException {
-    LOG.info(getProcessorName() + ":Stopping master. isLeader: " + isLeaderMode());
+    LOG.info(getServiceName() + ":Stopping master. isLeader: " + isLeaderMode());
     if (isStandbyMode()) {
       if (mStandbyJournalTailer != null) {
         // stop and wait for the journal tailer thread.
         mStandbyJournalTailer.shutdownAndJoin();
+        mStandbyJournalTailer = null;
       }
     } else {
       // Stop this master.
@@ -143,9 +147,7 @@ protected boolean isStandbyMode() {
   }
 
   protected void writeJournalEntry(JournalEntry entry) {
-    if (mJournalWriter == null) {
-      throw new RuntimeException("Cannot write entry: journal writer is null.");
-    }
+    Preconditions.checkNotNull(mJournalWriter, "Cannot write entry: journal writer is null.");
     try {
       mJournalWriter.getEntryOutputStream().writeEntry(entry);
     } catch (IOException ioe) {
@@ -154,9 +156,7 @@ protected void writeJournalEntry(JournalEntry entry) {
   }
 
   protected void flushJournal() {
-    if (mJournalWriter == null) {
-      throw new RuntimeException("Cannot flush journal: Journal writer is null.");
-    }
+    Preconditions.checkNotNull(mJournalWriter, "Cannot write entry: journal writer is null.");
     try {
       mJournalWriter.getEntryOutputStream().flush();
     } catch (IOException ioe) {

servers/src/main/java/tachyon/master/MasterSource.java

@@ -121,6 +121,7 @@ public Integer getValue() {
       }
     });
 
+    // TODO: renable when metrics is fully implemented on the master.
     /*
     mMetricRegistry.register(MetricRegistry.name("FilesTotal"), new Gauge<Integer>() {
       @Override

servers/src/main/java/tachyon/master/TachyonMaster.java

@@ -58,6 +58,7 @@ public static void main(String[] args) {
     }
 
     try {
+      // TODO: create a master context with the tachyon conf.
       TachyonConf conf = new TachyonConf();
       TachyonMaster master;
       if (conf.getBoolean(Constants.USE_ZOOKEEPER)) {
@@ -150,80 +151,76 @@ public TachyonMaster(TachyonConf tachyonConf) {
   }
 
   /**
-   * Returns the underlying {@link TachyonConf} instance for the Worker.
-   *
-   * @return TachyonConf of the Master
+   * @return the underlying {@link TachyonConf} instance for the master.
    */
   public TachyonConf getTachyonConf() {
     return mTachyonConf;
   }
 
   /**
-   * Gets the externally resolvable address of this master.
+   * @return the externally resolvable address of this master.
    */
   public InetSocketAddress getMasterAddress() {
     return mMasterAddress;
   }
 
   /**
-   * Gets the actual bind hostname on RPC service (used by unit test only).
+   * @return the actual bind hostname on RPC service (used by unit test only).
    */
   public String getRPCBindHost() {
     return NetworkAddressUtils.getThriftSocket(mTServerSocket).getLocalSocketAddress().toString();
   }
 
   /**
-   * Gets the actual port that the RPC service is listening on (used by unit test only)
+   * @return the actual port that the RPC service is listening on (used by unit test only)
    */
   public int getRPCLocalPort() {
     return mPort;
   }
 
   /**
-   * Gets the actual bind hostname on web service (used by unit test only).
+   * @return the actual bind hostname on web service (used by unit test only).
    */
   public String getWebBindHost() {
     return mWebServer.getBindHost();
   }
 
   /**
-   * Gets the actual port that the web service is listening on (used by unit test only)
+   * @return the actual port that the web service is listening on (used by unit test only)
    */
   public int getWebLocalPort() {
     return mWebServer.getLocalPort();
   }
 
   /**
-   * Gets internal {@link FileSystemMaster}, for unit test only.
+   * @return internal {@link FileSystemMaster}, for unit test only.
    */
   public FileSystemMaster getFileSystemMaster() {
     return mFileSystemMaster;
   }
 
   /**
-   * Gets internal {@link RawTableMaster}, for unit test only.
+   * @return internal {@link RawTableMaster}, for unit test only.
    */
   public RawTableMaster getRawTableMaster() {
     return mRawTableMaster;
   }
 
   /**
-   * Gets internal {@link BlockMaster}, for unit test only.
+   * @return internal {@link BlockMaster}, for unit test only.
    */
   public BlockMaster getBlockMaster() {
     return mBlockMaster;
   }
 
   /**
-   * Gets the millisecond when Tachyon Master starts serving, return -1 when not started.
+   * @return the millisecond when Tachyon Master starts serving, return -1 when not started.
    */
   public long getStarttimeMs() {
     return mStartTimeMs;
   }
 
   /**
-   * Gets whether the system is serving the rpc server.
-   *
    * @return true if the system is the leader (serving the rpc server), false otherwise.
    */
   boolean isServing() {
@@ -293,10 +290,10 @@ protected void startServingWebServer() {
   protected void startServingRPCServer() {
     // set up multiplexed thrift processors
     TMultiplexedProcessor processor = new TMultiplexedProcessor();
-    processor.registerProcessor(mBlockMaster.getProcessorName(), mBlockMaster.getProcessor());
-    processor.registerProcessor(mFileSystemMaster.getProcessorName(),
+    processor.registerProcessor(mBlockMaster.getServiceName(), mBlockMaster.getProcessor());
+    processor.registerProcessor(mFileSystemMaster.getServiceName(),
         mFileSystemMaster.getProcessor());
-    processor.registerProcessor(mRawTableMaster.getProcessorName(), mRawTableMaster.getProcessor());
+    processor.registerProcessor(mRawTableMaster.getServiceName(), mRawTableMaster.getProcessor());
 
     // create master thrift service with the multiplexed processor.
     mMasterServiceServer = new TThreadPoolServer(new TThreadPoolServer.Args(mTServerSocket)
@@ -313,9 +310,11 @@ protected void startServingRPCServer() {
   protected void stopServing() throws Exception {
     if (mMasterServiceServer != null) {
       mMasterServiceServer.stop();
+      mMasterServiceServer = null;
     }
     if (mWebServer != null) {
       mWebServer.shutdownWebServer();
+      mWebServer = null;
     }
     mIsServing = false;
   }
@@ -330,6 +329,9 @@ protected void stopServing() throws Exception {
   private boolean isJournalFormatted(String journalDirectory) throws IOException {
     UnderFileSystem ufs = UnderFileSystem.get(journalDirectory, mTachyonConf);
     if (!ufs.providesStorage()) {
+      // TODO: Should the journal really be allowed on a ufs without storage?
+      // This ufs doesn't provide storage. Allow the master to use this ufs for the journal.
+      LOG.info("Journal directory doesn't provide storage: " + journalDirectory);
       return true;
     }
     String[] files = ufs.list(journalDirectory);

servers/src/main/java/tachyon/master/TachyonMasterFaultTolerant.java

@@ -20,6 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 
 import tachyon.Constants;
@@ -43,6 +44,7 @@ public final class TachyonMasterFaultTolerant extends TachyonMaster {
 
   public TachyonMasterFaultTolerant(TachyonConf tachyonConf) {
     super(tachyonConf);
+    Preconditions.checkArgument(tachyonConf.getBoolean(Constants.USE_ZOOKEEPER));
 
     // Set up zookeeper specific functionality.
     try {
@@ -78,15 +80,14 @@ public void start() throws Exception {
 
     while (true) {
       if (mLeaderSelectorClient.isLeader()) {
-        if (started) {
-          stopMasters();
-        }
+        stopMasters();
         startMasters(true);
         started = true;
         startServing();
       } else {
         // This master should be standby, and not the leader
         if (isServing() || !started) {
+          // Need to transition this master to standby mode.
           stopServing();
           stopMasters();
 
@@ -100,6 +101,7 @@ public void start() throws Exception {
           startMasters(false);
           started = true;
         }
+        // This master is already in standby mode. No further actions needed.
       }
 
       CommonUtils.sleepMs(LOG, 100);