Add append to JournalContext

core/server/common/src/main/java/alluxio/master/AbstractMaster.java

@@ -18,7 +18,7 @@
 import alluxio.exception.PreconditionMessage;
 import alluxio.master.journal.AsyncJournalWriter;
 import alluxio.master.journal.Journal;
-import alluxio.master.journal.JournalEntryAppender;
+import alluxio.master.journal.JournalContext;
 import alluxio.master.journal.JournalInputStream;
 import alluxio.master.journal.JournalOutputStream;
 import alluxio.master.journal.JournalTailer;
@@ -258,7 +258,7 @@ protected void flushJournal() {
    */
   protected void appendJournalEntry(JournalEntry entry, JournalContext journalContext) {
     Preconditions.checkNotNull(mAsyncJournalWriter, PreconditionMessage.ASYNC_JOURNAL_WRITER_NULL);
-    journalContext.setFlushCounter(mAsyncJournalWriter.appendEntry(entry));
+    journalContext.append(entry);
   }
 
   /**
@@ -306,64 +306,43 @@ protected ExecutorService getExecutorService() {
    * @return new instance of {@link JournalContext}
    */
   protected JournalContext createJournalContext() {
-    return new JournalContext();
-  }
-
-  /**
-   * @return new instance of {@link JournalEntryAppender}
-   */
-  protected JournalEntryAppender createJournalAppender(JournalContext journalContext) {
-    return new MasterJournalEntryAppender(journalContext, mAsyncJournalWriter);
+    return new MasterJournalContext(mAsyncJournalWriter);
   }
 
   /**
    * Context for storing journaling information.
    */
   @NotThreadSafe
-  public final class JournalContext implements AutoCloseable {
+  public final class MasterJournalContext implements JournalContext {
+    private final AsyncJournalWriter mAsyncJournalWriter;
     private long mFlushCounter;
 
-    private JournalContext() {
+    /**
+     * Constructs a {@link MasterJournalContext}.
+     *
+     * @param asyncJournalWriter a {@link AsyncJournalWriter}
+     */
+    private MasterJournalContext(AsyncJournalWriter asyncJournalWriter) {
+      mAsyncJournalWriter = asyncJournalWriter;
       mFlushCounter = INVALID_FLUSH_COUNTER;
     }
 
-    private long getFlushCounter() {
+    @Override
+    public long getFlushCounter() {
       return mFlushCounter;
     }
 
-    private void setFlushCounter(long counter) {
-      mFlushCounter = counter;
-    }
-
     @Override
-    public void close() {
-      waitForJournalFlush(this);
-    }
-  }
-
-  /**
-   * This class appends journal entries to the async journal.
-   */
-  public final class MasterJournalEntryAppender implements JournalEntryAppender {
-    private final JournalContext mJournalContext;
-    private final AsyncJournalWriter mAsyncJournalWriter;
-
-    /**
-     * Constructs a {@link MasterJournalEntryAppender}.
-     *
-     * @param journalContext a {@link JournalContext}
-     * @param asyncJournalWriter a {@link AsyncJournalWriter}
-     */
-    public MasterJournalEntryAppender(JournalContext journalContext,
-        AsyncJournalWriter asyncJournalWriter) {
-      mJournalContext = journalContext;
-      mAsyncJournalWriter = Preconditions.checkNotNull(asyncJournalWriter);
+    public void append(alluxio.proto.journal.Journal.JournalEntry entry) {
+      if (mAsyncJournalWriter != null) {
+        mFlushCounter = mAsyncJournalWriter.appendEntry(entry);
+      }
     }
 
     @Override
-    public void append(alluxio.proto.journal.Journal.JournalEntry entry) {
-      if (mJournalContext != null) {
-        mJournalContext.setFlushCounter(mAsyncJournalWriter.appendEntry(entry));
+    public void close() {
+      if (mAsyncJournalWriter != null) {
+        waitForJournalFlush(this);
       }
     }
   }

core/server/common/src/main/java/alluxio/master/journal/JournalEntryAppender.java → core/server/common/src/main/java/alluxio/master/journal/JournalContext.java

@@ -13,12 +13,22 @@
 
 import alluxio.proto.journal.Journal;
 
+import java.io.Closeable;
+
 /**
- * This interface enables appending entries to the journal.
+ * Context for storing journaling information.
  */
-public interface JournalEntryAppender {
+public interface JournalContext extends Closeable {
   /**
    * @param entry the {@link Journal.JournalEntry} to append to the journal
    */
   void append(Journal.JournalEntry entry);
+
+  /**
+   * @return the journal flush counter
+   */
+  long getFlushCounter();
+
+  @Override
+  void close();
 }

core/server/master/src/test/java/alluxio/master/NoopJournalAppender.java → core/server/common/src/main/java/alluxio/master/journal/NoopJournalContext.java

@@ -9,21 +9,33 @@
  * See the NOTICE file distributed with this work for information regarding copyright ownership.
  */
 
-package alluxio.master;
+package alluxio.master.journal;
 
-import alluxio.master.journal.JournalEntryAppender;
 import alluxio.proto.journal.Journal;
 
 /**
- * NOOP version of JournalEntryAppender.
+ * Noop version of JournalContext.
  */
-public final class NoopJournalAppender implements JournalEntryAppender {
-  public NoopJournalAppender() {
+public final class NoopJournalContext implements JournalContext {
+  /**
+   * Constructs the {@link NoopJournalContext}.
+   */
+  public NoopJournalContext() {
     // Do nothing
   }
 
   @Override
   public void append(Journal.JournalEntry entry) {
     // Do nothing
   }
+
+  @Override
+  public long getFlushCounter() {
+    return -1;
+  }
+
+  @Override
+  public void close() {
+    // Do nothing
+  }
 }

core/server/master/src/main/java/alluxio/master/block/BlockMaster.java

@@ -32,6 +32,7 @@
 import alluxio.master.block.meta.MasterBlockInfo;
 import alluxio.master.block.meta.MasterBlockLocation;
 import alluxio.master.block.meta.MasterWorkerInfo;
+import alluxio.master.journal.JournalContext;
 import alluxio.master.journal.JournalFactory;
 import alluxio.master.journal.JournalInputStream;
 import alluxio.master.journal.JournalOutputStream;

core/server/master/src/main/java/alluxio/master/file/FileSystemMaster.java

@@ -66,6 +66,7 @@
 import alluxio.master.file.options.MountOptions;
 import alluxio.master.file.options.RenameOptions;
 import alluxio.master.file.options.SetAttributeOptions;
+import alluxio.master.journal.JournalContext;
 import alluxio.master.journal.JournalFactory;
 import alluxio.master.journal.JournalOutputStream;
 import alluxio.metrics.MetricsSystem;
@@ -1150,7 +1151,7 @@ InodeTree.CreatePathResult createFileInternal(LockedInodePath inodePath,
       options.setCacheable(true);
     }
     InodeTree.CreatePathResult createResult =
-        mInodeTree.createPath(inodePath, options, createJournalAppender(journalContext));
+        mInodeTree.createPath(inodePath, options, journalContext);
     // If the create succeeded, the list of created inodes will not be empty.
     List<Inode<?>> created = createResult.getCreated();
     InodeFile inode = (InodeFile) created.get(created.size() - 1);
@@ -1430,7 +1431,7 @@ private void deleteInternal(LockedInodePath inodePath, JournalContext journalCon
           mBlockMaster.removeBlocks(((InodeFile) delInode).getBlockIds(), true /* delete */);
         }
 
-        if (i == 0 && !replayed) {
+        if (i == 0) {
           // Journal right before deleting the "root" of the sub-tree from the parent, since the
           // parent is read locked.
           DeleteFileEntry deleteFile = DeleteFileEntry.newBuilder().setId(delInode.getId())
@@ -1706,7 +1707,7 @@ private InodeTree.CreatePathResult createDirectoryInternal(LockedInodePath inode
       FileDoesNotExistException {
     try {
       InodeTree.CreatePathResult createResult =
-          mInodeTree.createPath(inodePath, options, createJournalAppender(journalContext));
+          mInodeTree.createPath(inodePath, options, journalContext);
       InodeDirectory inodeDirectory = (InodeDirectory) inodePath.getInode();
       // If inodeDirectory's ttl not equals Constants.NO_TTL, it should insert into mTtlBuckets
       if (createResult.getCreated().size() > 0) {
@@ -1911,8 +1912,7 @@ private void renameInternal(LockedInodePath srcInodePath, LockedInodePath dstIno
         while (!sameMountDirs.empty()) {
           InodeDirectory dir = sameMountDirs.pop();
           if (!dir.isPersisted()) {
-            InodeUtils.syncPersistDirectory(dir, mInodeTree, mMountTable,
-                createJournalAppender(journalContext));
+            InodeUtils.syncPersistDirectory(dir, mInodeTree, mMountTable, journalContext);
           }
         }
 

core/server/master/src/main/java/alluxio/master/file/meta/InodeDirectoryIdGenerator.java

@@ -13,7 +13,7 @@
 
 import alluxio.master.block.BlockId;
 import alluxio.master.block.ContainerIdGenerable;
-import alluxio.master.journal.JournalEntryAppender;
+import alluxio.master.journal.JournalContext;
 import alluxio.master.journal.JournalEntryRepresentable;
 import alluxio.proto.journal.File.InodeDirectoryIdGeneratorEntry;
 import alluxio.proto.journal.Journal.JournalEntry;
@@ -52,10 +52,10 @@ synchronized long getNewDirectoryId() {
   /**
    * Returns the next directory id, and journals the state.
    *
-   * @param journalAppender the appender to journal to, if not null
+   * @param journalContext the journal context
    * @return the next directory id
    */
-  synchronized long getNewDirectoryId(JournalEntryAppender journalAppender) {
+  synchronized long getNewDirectoryId(JournalContext journalContext) {
     initialize();
     long directoryId = BlockId.createBlockId(mContainerId, mSequenceNumber);
     if (mSequenceNumber == BlockId.getMaxSequenceNumber()) {
@@ -65,8 +65,8 @@ synchronized long getNewDirectoryId(JournalEntryAppender journalAppender) {
     } else {
       mSequenceNumber++;
     }
-    if (journalAppender != null) {
-      journalAppender.append(toJournalEntry());
+    if (journalContext != null) {
+      journalContext.append(toJournalEntry());
     }
     return directoryId;
   }

core/server/master/src/main/java/alluxio/master/file/meta/InodeTree.java

@@ -29,7 +29,7 @@
 import alluxio.master.file.options.CreateFileOptions;
 import alluxio.master.file.options.CreatePathOptions;
 import alluxio.master.journal.JournalCheckpointStreamable;
-import alluxio.master.journal.JournalEntryAppender;
+import alluxio.master.journal.JournalContext;
 import alluxio.master.journal.JournalOutputStream;
 import alluxio.proto.journal.File;
 import alluxio.proto.journal.File.InodeDirectoryEntry;
@@ -466,7 +466,7 @@ public InodeDirectory getRoot() {
    *
    * @param inodePath the path
    * @param options method options
-   * @param journalAppender the journal appender
+   * @param journalContext the journal context
    * @return a {@link CreatePathResult} representing the modified inodes and created inodes during
    *         path creation
    * @throws FileAlreadyExistsException when there is already a file at path if we want to create a
@@ -480,7 +480,7 @@ public InodeDirectory getRoot() {
    *         option is false
    */
   public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<?> options,
-      JournalEntryAppender journalAppender)
+      JournalContext journalContext)
       throws FileAlreadyExistsException, BlockInfoException, InvalidPathException, IOException,
       FileDoesNotExistException {
     AlluxioURI path = inodePath.getUri();
@@ -538,7 +538,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
     if (options.isPersisted()) {
       // Synchronously persist directories. These inodes are already READ locked.
       for (Inode inode : traversalResult.getNonPersisted()) {
-        InodeUtils.syncPersistDirectory((InodeDirectory) inode, this, mMountTable, journalAppender);
+        InodeUtils.syncPersistDirectory((InodeDirectory) inode, this, mMountTable, journalContext);
       }
     }
     if (pathIndex < (pathComponents.length - 1) || currentInodeDirectory.getChild(name) == null) {
@@ -551,7 +551,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
       File.InodeLastModificationTimeEntry inodeLastModificationTime =
           File.InodeLastModificationTimeEntry.newBuilder().setId(currentInodeDirectory.getId())
               .setLastModificationTimeMs(options.getOperationTimeMs()).build();
-      journalAppender.append(
+      journalContext.append(
           Journal.JournalEntry.newBuilder().setInodeLastModificationTime(inodeLastModificationTime)
               .build());
     }
@@ -569,7 +569,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
     for (int k = pathIndex; k < (pathComponents.length - 1); k++) {
       InodeDirectory dir = null;
       while (dir == null) {
-        dir = InodeDirectory.create(mDirectoryIdGenerator.getNewDirectoryId(journalAppender),
+        dir = InodeDirectory.create(mDirectoryIdGenerator.getNewDirectoryId(journalContext),
             currentInodeDirectory.getId(), pathComponents[k], missingDirOptions);
         // Lock the newly created inode before subsequent operations, and add it to the lock group.
         lockList.lockWriteAndCheckNameAndParent(dir, currentInodeDirectory, pathComponents[k]);
@@ -593,7 +593,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
             InodeUtils.syncPersistDirectory(dir, this, mMountTable, null);
           }
           // Journal the new inode.
-          journalAppender.append(dir.toJournalEntry());
+          journalContext.append(dir.toJournalEntry());
 
           // After creation and journaling, downgrade to a read lock.
           lockList.unlockLast();
@@ -625,7 +625,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
         // The final path component already exists and is not persisted, so it should be added
         // to the non-persisted Inodes of traversalResult.
         InodeUtils
-            .syncPersistDirectory((InodeDirectory) lastInode, this, mMountTable, journalAppender);
+            .syncPersistDirectory((InodeDirectory) lastInode, this, mMountTable, journalContext);
       } else if (!lastInode.isDirectory() || !(options instanceof CreateDirectoryOptions
           && ((CreateDirectoryOptions) options).isAllowExists())) {
         String errorMessage = ExceptionMessage.FILE_ALREADY_EXISTS.getMessage(path);
@@ -636,7 +636,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
       // create the new inode
       if (options instanceof CreateDirectoryOptions) {
         CreateDirectoryOptions directoryOptions = (CreateDirectoryOptions) options;
-        lastInode = InodeDirectory.create(mDirectoryIdGenerator.getNewDirectoryId(journalAppender),
+        lastInode = InodeDirectory.create(mDirectoryIdGenerator.getNewDirectoryId(journalContext),
             currentInodeDirectory.getId(), name, directoryOptions);
         // Lock the created inode before subsequent operations, and add it to the lock group.
         lockList.lockWriteAndCheckNameAndParent(lastInode, currentInodeDirectory, name);
@@ -670,7 +670,7 @@ public CreatePathResult createPath(LockedInodePath inodePath, CreatePathOptions<
       }
 
       // Journal the new inode.
-      journalAppender.append(lastInode.toJournalEntry());
+      journalContext.append(lastInode.toJournalEntry());
 
       createdInodes.add(lastInode);
       extensibleInodePath.getInodes().add(lastInode);

core/server/master/src/main/java/alluxio/master/file/meta/InodeUtils.java

@@ -12,7 +12,7 @@
 package alluxio.master.file.meta;
 
 import alluxio.AlluxioURI;
-import alluxio.master.journal.JournalEntryAppender;
+import alluxio.master.journal.JournalContext;
 import alluxio.proto.journal.File;
 import alluxio.proto.journal.Journal;
 import alluxio.security.authorization.Mode;
@@ -38,10 +38,10 @@ private InodeUtils() {} // prevent instantiation
    * @param dir the {@link InodeDirectory} to persist
    * @param inodeTree the {@link InodeTree}
    * @param mountTable the {@link MountTable}
-   * @param journalAppender the appender to journal the persist entry to, if not null
+   * @param journalContext the journal context
    */
   public static void syncPersistDirectory(InodeDirectory dir, InodeTree inodeTree,
-      MountTable mountTable, JournalEntryAppender journalAppender) {
+      MountTable mountTable, JournalContext journalContext) {
     // TODO(gpang): use a max timeout.
     while (dir.getPersistenceState() != PersistenceState.PERSISTED) {
       if (dir.compareAndSwapPersistenceState(PersistenceState.NOT_PERSISTED,
@@ -58,13 +58,11 @@ public static void syncPersistDirectory(InodeDirectory dir, InodeTree inodeTree,
           ufs.mkdirs(ufsUri, mkdirsOptions);
           dir.setPersistenceState(PersistenceState.PERSISTED);
 
-          if (journalAppender != null) {
-            // Append the persist entry to the journal.
-            File.PersistDirectoryEntry persistDirectory =
-                File.PersistDirectoryEntry.newBuilder().setId(dir.getId()).build();
-            journalAppender.append(
-                Journal.JournalEntry.newBuilder().setPersistDirectory(persistDirectory).build());
-          }
+          // Append the persist entry to the journal.
+          File.PersistDirectoryEntry persistDirectory =
+              File.PersistDirectoryEntry.newBuilder().setId(dir.getId()).build();
+          journalContext.append(
+              Journal.JournalEntry.newBuilder().setPersistDirectory(persistDirectory).build());
           success = true;
         } catch (Exception e) {
           // Ignore