hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From w...@apache.org
Subject [2/2] git commit: HDFS-6634. inotify in HDFS. Contributed by James Thomas.
Date Tue, 02 Sep 2014 21:18:04 GMT
HDFS-6634. inotify in HDFS. Contributed by James Thomas.

(cherry picked from commit faa4455be512e070fa420084be8d1be5c72f3b08)

Conflicts:
	hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
	hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/IPCLoggerChannel.java
	hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/958c9b50
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/958c9b50
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/958c9b50

Branch: refs/heads/branch-2
Commit: 958c9b5080d584c2e60fa2e1a041e5dc5f7498eb
Parents: c50371c
Author: Andrew Wang <wang@apache.org>
Authored: Tue Sep 2 14:02:29 2014 -0700
Committer: Andrew Wang <wang@apache.org>
Committed: Tue Sep 2 14:17:51 2014 -0700

----------------------------------------------------------------------
 hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt     |   2 +
 .../dev-support/findbugsExcludeFile.xml         |   9 +
 hadoop-hdfs-project/hadoop-hdfs/pom.xml         |   1 +
 .../bkjournal/BookKeeperEditLogInputStream.java |   5 +
 .../java/org/apache/hadoop/hdfs/DFSClient.java  |   9 +
 .../org/apache/hadoop/hdfs/DFSConfigKeys.java   |   6 +
 .../hadoop/hdfs/DFSInotifyEventInputStream.java | 220 +++++++++
 .../hadoop/hdfs/DistributedFileSystem.java      |   9 +
 .../apache/hadoop/hdfs/client/HdfsAdmin.java    |  50 ++
 .../org/apache/hadoop/hdfs/inotify/Event.java   | 452 +++++++++++++++++++
 .../apache/hadoop/hdfs/inotify/EventsList.java  |  63 +++
 .../hdfs/inotify/MissingEventsException.java    |  54 +++
 .../hadoop/hdfs/protocol/ClientProtocol.java    |  18 +
 ...tNamenodeProtocolServerSideTranslatorPB.java |  25 +
 .../ClientNamenodeProtocolTranslatorPB.java     |  25 +
 .../apache/hadoop/hdfs/protocolPB/PBHelper.java | 245 ++++++++++
 .../hdfs/qjournal/client/IPCLoggerChannel.java  |  80 ++--
 .../hadoop/hdfs/qjournal/server/Journal.java    |   3 +-
 .../namenode/EditLogBackupInputStream.java      |   5 +
 .../server/namenode/EditLogFileInputStream.java |   5 +
 .../server/namenode/EditLogInputStream.java     |   6 +
 .../hadoop/hdfs/server/namenode/FSEditLog.java  |  67 ++-
 .../server/namenode/FileJournalManager.java     |  16 +-
 .../namenode/InotifyFSEditLogOpTranslator.java  | 146 ++++++
 .../hadoop/hdfs/server/namenode/JournalSet.java |  38 +-
 .../hdfs/server/namenode/NameNodeRpcServer.java | 114 +++++
 .../namenode/RedundantEditLogInputStream.java   |   5 +
 .../hdfs/server/namenode/TransferFsImage.java   |   5 +-
 .../src/main/proto/ClientNamenodeProtocol.proto |  20 +
 .../hadoop-hdfs/src/main/proto/inotify.proto    | 117 +++++
 .../src/main/resources/hdfs-default.xml         |  10 +
 .../hdfs/TestDFSInotifyEventInputStream.java    | 430 ++++++++++++++++++
 .../hadoop/hdfs/qjournal/MiniQJMHACluster.java  |   6 +-
 .../hdfs/qjournal/client/TestQJMWithFaults.java |   2 +-
 .../client/TestQuorumJournalManager.java        |   2 +-
 .../hdfs/server/namenode/TestEditLog.java       |   4 +
 36 files changed, 2215 insertions(+), 59 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 6e385db..12d8098 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -171,6 +171,8 @@ Release 2.6.0 - UNRELEASED
     HDFS-6774. Make FsDataset and DataStore support removing volumes. (Lei Xu
     via atm)
 
+    HDFS-6634. inotify in HDFS. (James Thomas via wang)
+
   OPTIMIZATIONS
 
     HDFS-6690. Deduplicate xattr names in memory. (wang)

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml b/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml
index 29702d4..bbfb9e9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml
@@ -106,6 +106,15 @@
        <Field name="metrics" />
        <Bug pattern="IS2_INCONSISTENT_SYNC" />
      </Match>
+    <!--
+     We use a separate lock to protect modifications to journalSet so that
+     FSEditLog#selectInputStreams does not need to be a synchronized method.
+    -->
+    <Match>
+        <Class name="org.apache.hadoop.hdfs.server.namenode.FSEditLog" />
+        <Field name="journalSet" />
+        <Bug pattern="IS2_INCONSISTENT_SYNC" />
+    </Match>
      <!--
       This method isn't performance-critical and is much clearer to write as it's written.
       -->

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/pom.xml b/hadoop-hdfs-project/hadoop-hdfs/pom.xml
index f3fef4c..d4d61b0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/pom.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/pom.xml
@@ -453,6 +453,7 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd">
                   <include>fsimage.proto</include>
                   <include>hdfs.proto</include>
                   <include>encryption.proto</include>
+                  <include>inotify.proto</include>
                 </includes>
               </source>
               <output>${project.build.directory}/generated-sources/java</output>

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/BookKeeperEditLogInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/BookKeeperEditLogInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/BookKeeperEditLogInputStream.java
index bd3ccd4..e2098dd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/BookKeeperEditLogInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/main/java/org/apache/hadoop/contrib/bkjournal/BookKeeperEditLogInputStream.java
@@ -168,6 +168,11 @@ class BookKeeperEditLogInputStream extends EditLogInputStream {
     reader.setMaxOpSize(maxOpSize);
   }
 
+  @Override
+  public boolean isLocalLog() {
+    return false;
+  }
+
   /**
    * Input stream implementation which can be used by 
    * FSEditLogOp.Reader

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index c49d210..ce0f133 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -2990,6 +2990,15 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
     }
   }
 
+  public DFSInotifyEventInputStream getInotifyEventStream() throws IOException {
+    return new DFSInotifyEventInputStream(namenode);
+  }
+
+  public DFSInotifyEventInputStream getInotifyEventStream(long lastReadTxid)
+      throws IOException {
+    return new DFSInotifyEventInputStream(namenode, lastReadTxid);
+  }
+
   @Override // RemotePeerFactory
   public Peer newConnectedPeer(InetSocketAddress addr,
       Token<BlockTokenIdentifier> blockToken, DatanodeID datanodeId)

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index 3b6df70..f1a9705 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -677,4 +677,10 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
    public static final String DFS_DATANODE_SLOW_IO_WARNING_THRESHOLD_KEY =
      "dfs.datanode.slow.io.warning.threshold.ms";
    public static final long DFS_DATANODE_SLOW_IO_WARNING_THRESHOLD_DEFAULT = 300;
+
+  public static final String DFS_NAMENODE_INOTIFY_MAX_EVENTS_PER_RPC_KEY =
+      "dfs.namenode.inotify.max.events.per.rpc";
+  public static final int DFS_NAMENODE_INOTIFY_MAX_EVENTS_PER_RPC_DEFAULT =
+      1000;
+
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInotifyEventInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInotifyEventInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInotifyEventInputStream.java
new file mode 100644
index 0000000..73c5f55
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInotifyEventInputStream.java
@@ -0,0 +1,220 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs;
+
+import com.google.common.collect.Iterators;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.inotify.Event;
+import org.apache.hadoop.hdfs.inotify.EventsList;
+import org.apache.hadoop.hdfs.inotify.MissingEventsException;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Stream for reading inotify events. DFSInotifyEventInputStreams should not
+ * be shared among multiple threads.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class DFSInotifyEventInputStream {
+  public static Logger LOG = LoggerFactory.getLogger(DFSInotifyEventInputStream
+      .class);
+
+  private final ClientProtocol namenode;
+  private Iterator<Event> it;
+  private long lastReadTxid;
+  /**
+   * The most recent txid the NameNode told us it has sync'ed -- helps us
+   * determine how far behind we are in the edit stream.
+   */
+  private long syncTxid;
+  /**
+   * Used to generate wait times in {@link DFSInotifyEventInputStream#take()}.
+   */
+  private Random rng = new Random();
+
+  private static final int INITIAL_WAIT_MS = 10;
+
+  DFSInotifyEventInputStream(ClientProtocol namenode) throws IOException {
+    this(namenode, namenode.getCurrentEditLogTxid()); // only consider new txn's
+  }
+
+  DFSInotifyEventInputStream(ClientProtocol namenode, long lastReadTxid)
+      throws IOException {
+    this.namenode = namenode;
+    this.it = Iterators.emptyIterator();
+    this.lastReadTxid = lastReadTxid;
+  }
+
+  /**
+   * Returns the next event in the stream or null if no new events are currently
+   * available.
+   *
+   * @throws IOException because of network error or edit log
+   * corruption. Also possible if JournalNodes are unresponsive in the
+   * QJM setting (even one unresponsive JournalNode is enough in rare cases),
+   * so catching this exception and retrying at least a few times is
+   * recommended.
+   * @throws MissingEventsException if we cannot return the next event in the
+   * stream because the data for the event (and possibly some subsequent events)
+   * has been deleted (generally because this stream is a very large number of
+   * events behind the current state of the NameNode). It is safe to continue
+   * reading from the stream after this exception is thrown -- the next
+   * available event will be returned.
+   */
+  public Event poll() throws IOException, MissingEventsException {
+    // need to keep retrying until the NN sends us the latest committed txid
+    if (lastReadTxid == -1) {
+      LOG.debug("poll(): lastReadTxid is -1, reading current txid from NN");
+      lastReadTxid = namenode.getCurrentEditLogTxid();
+      return null;
+    }
+    if (!it.hasNext()) {
+      EventsList el = namenode.getEditsFromTxid(lastReadTxid + 1);
+      if (el.getLastTxid() != -1) {
+        // we only want to set syncTxid when we were actually able to read some
+        // edits on the NN -- otherwise it will seem like edits are being
+        // generated faster than we can read them when the problem is really
+        // that we are temporarily unable to read edits
+        syncTxid = el.getSyncTxid();
+        it = el.getEvents().iterator();
+        long formerLastReadTxid = lastReadTxid;
+        lastReadTxid = el.getLastTxid();
+        if (el.getFirstTxid() != formerLastReadTxid + 1) {
+          throw new MissingEventsException(formerLastReadTxid + 1,
+              el.getFirstTxid());
+        }
+      } else {
+        LOG.debug("poll(): read no edits from the NN when requesting edits " +
+          "after txid {}", lastReadTxid);
+        return null;
+      }
+    }
+
+    if (it.hasNext()) { // can be empty if el.getLastTxid != -1 but none of the
+      // newly seen edit log ops actually got converted to events
+      return it.next();
+    } else {
+      return null;
+    }
+  }
+
+  /**
+   * Return a estimate of how many events behind the NameNode's current state
+   * this stream is. Clients should periodically call this method and check if
+   * its result is steadily increasing, which indicates that they are falling
+   * behind (i.e. events are being generated faster than the client is reading
+   * them). If a client falls too far behind events may be deleted before the
+   * client can read them.
+   * <p/>
+   * A return value of -1 indicates that an estimate could not be produced, and
+   * should be ignored. The value returned by this method is really only useful
+   * when compared to previous or subsequent returned values.
+   */
+  public long getEventsBehindEstimate() {
+    if (syncTxid == 0) {
+      return -1;
+    } else {
+      assert syncTxid >= lastReadTxid;
+      // this gives the difference between the last txid we have fetched to the
+      // client and syncTxid at the time we last fetched events from the
+      // NameNode
+      return syncTxid - lastReadTxid;
+    }
+  }
+
+  /**
+   * Returns the next event in the stream, waiting up to the specified amount of
+   * time for a new event. Returns null if a new event is not available at the
+   * end of the specified amount of time. The time before the method returns may
+   * exceed the specified amount of time by up to the time required for an RPC
+   * to the NameNode.
+   *
+   * @param time number of units of the given TimeUnit to wait
+   * @param tu the desired TimeUnit
+   * @throws IOException see {@link DFSInotifyEventInputStream#poll()}
+   * @throws MissingEventsException
+   * see {@link DFSInotifyEventInputStream#poll()}
+   * @throws InterruptedException if the calling thread is interrupted
+   */
+  public Event poll(long time, TimeUnit tu) throws IOException,
+      InterruptedException, MissingEventsException {
+    long initialTime = Time.monotonicNow();
+    long totalWait = TimeUnit.MILLISECONDS.convert(time, tu);
+    long nextWait = INITIAL_WAIT_MS;
+    Event next = null;
+    while ((next = poll()) == null) {
+      long timeLeft = totalWait - (Time.monotonicNow() - initialTime);
+      if (timeLeft <= 0) {
+        LOG.debug("timed poll(): timed out");
+        break;
+      } else if (timeLeft < nextWait * 2) {
+        nextWait = timeLeft;
+      } else {
+        nextWait *= 2;
+      }
+      LOG.debug("timed poll(): poll() returned null, sleeping for {} ms",
+          nextWait);
+      Thread.sleep(nextWait);
+    }
+
+    return next;
+  }
+
+  /**
+   * Returns the next event in the stream, waiting indefinitely if a new event
+   * is not immediately available.
+   *
+   * @throws IOException see {@link DFSInotifyEventInputStream#poll()}
+   * @throws MissingEventsException see
+   * {@link DFSInotifyEventInputStream#poll()}
+   * @throws InterruptedException if the calling thread is interrupted
+   */
+  public Event take() throws IOException, InterruptedException,
+      MissingEventsException {
+    Event next = null;
+    int nextWaitMin = INITIAL_WAIT_MS;
+    while ((next = poll()) == null) {
+      // sleep for a random period between nextWaitMin and nextWaitMin * 2
+      // to avoid stampedes at the NN if there are multiple clients
+      int sleepTime = nextWaitMin + rng.nextInt(nextWaitMin);
+      LOG.debug("take(): poll() returned null, sleeping for {} ms", sleepTime);
+      Thread.sleep(sleepTime);
+      // the maximum sleep is 2 minutes
+      nextWaitMin = Math.min(60000, nextWaitMin * 2);
+    }
+
+    return next;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
index 57c886a..af7d307 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
@@ -1988,4 +1988,13 @@ public class DistributedFileSystem extends FileSystem {
       }
     }.resolve(this, absF);
   }
+
+  public DFSInotifyEventInputStream getInotifyEventStream() throws IOException {
+    return dfs.getInotifyEventStream();
+  }
+
+  public DFSInotifyEventInputStream getInotifyEventStream(long lastReadTxid)
+      throws IOException {
+    return dfs.getInotifyEventStream(lastReadTxid);
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsAdmin.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsAdmin.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsAdmin.java
index 1adfc1b..fdc466a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsAdmin.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsAdmin.java
@@ -29,6 +29,7 @@ import org.apache.hadoop.fs.CacheFlag;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hdfs.DFSInotifyEventInputStream;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveEntry;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveInfo;
@@ -275,4 +276,53 @@ public class HdfsAdmin {
       throws IOException {
     return dfs.listEncryptionZones();
   }
+
+  /**
+   * Exposes a stream of namesystem events. Only events occurring after the
+   * stream is created are available.
+   * See {@link org.apache.hadoop.hdfs.DFSInotifyEventInputStream}
+   * for information on stream usage.
+   * See {@link org.apache.hadoop.hdfs.inotify.Event}
+   * for information on the available events.
+   * <p/>
+   * Inotify users may want to tune the following HDFS parameters to
+   * ensure that enough extra HDFS edits are saved to support inotify clients
+   * that fall behind the current state of the namespace while reading events.
+   * The default parameter values should generally be reasonable. If edits are
+   * deleted before their corresponding events can be read, clients will see a
+   * {@link org.apache.hadoop.hdfs.inotify.MissingEventsException} on
+   * {@link org.apache.hadoop.hdfs.DFSInotifyEventInputStream} method calls.
+   *
+   * It should generally be sufficient to tune these parameters:
+   * dfs.namenode.num.extra.edits.retained
+   * dfs.namenode.max.extra.edits.segments.retained
+   *
+   * Parameters that affect the number of created segments and the number of
+   * edits that are considered necessary, i.e. do not count towards the
+   * dfs.namenode.num.extra.edits.retained quota):
+   * dfs.namenode.checkpoint.period
+   * dfs.namenode.checkpoint.txns
+   * dfs.namenode.num.checkpoints.retained
+   * dfs.ha.log-roll.period
+   * <p/>
+   * It is recommended that local journaling be configured
+   * (dfs.namenode.edits.dir) for inotify (in addition to a shared journal)
+   * so that edit transfers from the shared journal can be avoided.
+   *
+   * @throws IOException If there was an error obtaining the stream.
+   */
+  public DFSInotifyEventInputStream getInotifyEventStream() throws IOException {
+    return dfs.getInotifyEventStream();
+  }
+
+  /**
+   * A version of {@link HdfsAdmin#getInotifyEventStream()} meant for advanced
+   * users who are aware of HDFS edits up to lastReadTxid (e.g. because they
+   * have access to an FSImage inclusive of lastReadTxid) and only want to read
+   * events after this point.
+   */
+  public DFSInotifyEventInputStream getInotifyEventStream(long lastReadTxid)
+      throws IOException {
+    return dfs.getInotifyEventStream(lastReadTxid);
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/Event.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/Event.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/Event.java
new file mode 100644
index 0000000..c7129ca
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/Event.java
@@ -0,0 +1,452 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.inotify;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.XAttr;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.FsPermission;
+
+import java.util.List;
+
+/**
+ * Events sent by the inotify system. Note that no events are necessarily sent
+ * when a file is opened for read (although a MetadataUpdateEvent will be sent
+ * if the atime is updated).
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public abstract class Event {
+  public static enum EventType {
+    CREATE, CLOSE, APPEND, RENAME, METADATA, UNLINK
+  }
+
+  private EventType eventType;
+
+  public EventType getEventType() {
+    return eventType;
+  }
+
+  public Event(EventType eventType) {
+    this.eventType = eventType;
+  }
+
+  /**
+   * Sent when a file is closed after append or create.
+   */
+  public static class CloseEvent extends Event {
+    private String path;
+    private long fileSize;
+    private long timestamp;
+
+    public CloseEvent(String path, long fileSize, long timestamp) {
+      super(EventType.CLOSE);
+      this.path = path;
+      this.fileSize = fileSize;
+      this.timestamp = timestamp;
+    }
+
+    public String getPath() {
+      return path;
+    }
+
+    /**
+     * The size of the closed file in bytes. May be -1 if the size is not
+     * available (e.g. in the case of a close generated by a concat operation).
+     */
+    public long getFileSize() {
+      return fileSize;
+    }
+
+    /**
+     * The time when this event occurred, in milliseconds since the epoch.
+     */
+    public long getTimestamp() {
+      return timestamp;
+    }
+  }
+
+  /**
+   * Sent when a new file is created (including overwrite).
+   */
+  public static class CreateEvent extends Event {
+
+    public static enum INodeType {
+      FILE, DIRECTORY, SYMLINK;
+    }
+
+    private INodeType iNodeType;
+    private String path;
+    private long ctime;
+    private int replication;
+    private String ownerName;
+    private String groupName;
+    private FsPermission perms;
+    private String symlinkTarget;
+
+    public static class Builder {
+      private INodeType iNodeType;
+      private String path;
+      private long ctime;
+      private int replication;
+      private String ownerName;
+      private String groupName;
+      private FsPermission perms;
+      private String symlinkTarget;
+
+      public Builder iNodeType(INodeType type) {
+        this.iNodeType = type;
+        return this;
+      }
+
+      public Builder path(String path) {
+        this.path = path;
+        return this;
+      }
+
+      public Builder ctime(long ctime) {
+        this.ctime = ctime;
+        return this;
+      }
+
+      public Builder replication(int replication) {
+        this.replication = replication;
+        return this;
+      }
+
+      public Builder ownerName(String ownerName) {
+        this.ownerName = ownerName;
+        return this;
+      }
+
+      public Builder groupName(String groupName) {
+        this.groupName = groupName;
+        return this;
+      }
+
+      public Builder perms(FsPermission perms) {
+        this.perms = perms;
+        return this;
+      }
+
+      public Builder symlinkTarget(String symlinkTarget) {
+        this.symlinkTarget = symlinkTarget;
+        return this;
+      }
+
+      public CreateEvent build() {
+        return new CreateEvent(this);
+      }
+    }
+
+    private CreateEvent(Builder b) {
+      super(EventType.CREATE);
+      this.iNodeType = b.iNodeType;
+      this.path = b.path;
+      this.ctime = b.ctime;
+      this.replication = b.replication;
+      this.ownerName = b.ownerName;
+      this.groupName = b.groupName;
+      this.perms = b.perms;
+      this.symlinkTarget = b.symlinkTarget;
+    }
+
+    public INodeType getiNodeType() {
+      return iNodeType;
+    }
+
+    public String getPath() {
+      return path;
+    }
+
+    /**
+     * Creation time of the file, directory, or symlink.
+     */
+    public long getCtime() {
+      return ctime;
+    }
+
+    /**
+     * Replication is zero if the CreateEvent iNodeType is directory or symlink.
+     */
+    public int getReplication() {
+      return replication;
+    }
+
+    public String getOwnerName() {
+      return ownerName;
+    }
+
+    public String getGroupName() {
+      return groupName;
+    }
+
+    public FsPermission getPerms() {
+      return perms;
+    }
+
+    /**
+     * Symlink target is null if the CreateEvent iNodeType is not symlink.
+     */
+    public String getSymlinkTarget() {
+      return symlinkTarget;
+    }
+  }
+
+  /**
+   * Sent when there is an update to directory or file (none of the metadata
+   * tracked here applies to symlinks) that is not associated with another
+   * inotify event. The tracked metadata includes atime/mtime, replication,
+   * owner/group, permissions, ACLs, and XAttributes. Fields not relevant to the
+   * metadataType of the MetadataUpdateEvent will be null or will have their default
+   * values.
+   */
+  public static class MetadataUpdateEvent extends Event {
+
+    public static enum MetadataType {
+      TIMES, REPLICATION, OWNER, PERMS, ACLS, XATTRS;
+    }
+
+    private String path;
+    private MetadataType metadataType;
+    private long mtime;
+    private long atime;
+    private int replication;
+    private String ownerName;
+    private String groupName;
+    private FsPermission perms;
+    private List<AclEntry> acls;
+    private List<XAttr> xAttrs;
+    private boolean xAttrsRemoved;
+
+    public static class Builder {
+      private String path;
+      private MetadataType metadataType;
+      private long mtime;
+      private long atime;
+      private int replication;
+      private String ownerName;
+      private String groupName;
+      private FsPermission perms;
+      private List<AclEntry> acls;
+      private List<XAttr> xAttrs;
+      private boolean xAttrsRemoved;
+
+      public Builder path(String path) {
+        this.path = path;
+        return this;
+      }
+
+      public Builder metadataType(MetadataType type) {
+        this.metadataType = type;
+        return this;
+      }
+
+      public Builder mtime(long mtime) {
+        this.mtime = mtime;
+        return this;
+      }
+
+      public Builder atime(long atime) {
+        this.atime = atime;
+        return this;
+      }
+
+      public Builder replication(int replication) {
+        this.replication = replication;
+        return this;
+      }
+
+      public Builder ownerName(String ownerName) {
+        this.ownerName = ownerName;
+        return this;
+      }
+
+      public Builder groupName(String groupName) {
+        this.groupName = groupName;
+        return this;
+      }
+
+      public Builder perms(FsPermission perms) {
+        this.perms = perms;
+        return this;
+      }
+
+      public Builder acls(List<AclEntry> acls) {
+        this.acls = acls;
+        return this;
+      }
+
+      public Builder xAttrs(List<XAttr> xAttrs) {
+        this.xAttrs = xAttrs;
+        return this;
+      }
+
+      public Builder xAttrsRemoved(boolean xAttrsRemoved) {
+        this.xAttrsRemoved = xAttrsRemoved;
+        return this;
+      }
+
+      public MetadataUpdateEvent build() {
+        return new MetadataUpdateEvent(this);
+      }
+    }
+
+    private MetadataUpdateEvent(Builder b) {
+      super(EventType.METADATA);
+      this.path = b.path;
+      this.metadataType = b.metadataType;
+      this.mtime = b.mtime;
+      this.atime = b.atime;
+      this.replication = b.replication;
+      this.ownerName = b.ownerName;
+      this.groupName = b.groupName;
+      this.perms = b.perms;
+      this.acls = b.acls;
+      this.xAttrs = b.xAttrs;
+      this.xAttrsRemoved = b.xAttrsRemoved;
+    }
+
+    public String getPath() {
+      return path;
+    }
+
+    public MetadataType getMetadataType() {
+      return metadataType;
+    }
+
+    public long getMtime() {
+      return mtime;
+    }
+
+    public long getAtime() {
+      return atime;
+    }
+
+    public int getReplication() {
+      return replication;
+    }
+
+    public String getOwnerName() {
+      return ownerName;
+    }
+
+    public String getGroupName() {
+      return groupName;
+    }
+
+    public FsPermission getPerms() {
+      return perms;
+    }
+
+    /**
+     * The full set of ACLs currently associated with this file or directory.
+     * May be null if all ACLs were removed.
+     */
+    public List<AclEntry> getAcls() {
+      return acls;
+    }
+
+    public List<XAttr> getxAttrs() {
+      return xAttrs;
+    }
+
+    /**
+     * Whether the xAttrs returned by getxAttrs() were removed (as opposed to
+     * added).
+     */
+    public boolean isxAttrsRemoved() {
+      return xAttrsRemoved;
+    }
+
+  }
+
+  /**
+   * Sent when a file, directory, or symlink is renamed.
+   */
+  public static class RenameEvent extends Event {
+    private String srcPath;
+    private String dstPath;
+    private long timestamp;
+
+    public RenameEvent(String srcPath, String dstPath, long timestamp) {
+      super(EventType.RENAME);
+      this.srcPath = srcPath;
+      this.dstPath = dstPath;
+      this.timestamp = timestamp;
+    }
+
+    public String getSrcPath() {
+      return srcPath;
+    }
+
+    public String getDstPath() {
+      return dstPath;
+    }
+
+    /**
+     * The time when this event occurred, in milliseconds since the epoch.
+     */
+    public long getTimestamp() {
+      return timestamp;
+    }
+  }
+
+  /**
+   * Sent when an existing file is opened for append.
+   */
+  public static class AppendEvent extends Event {
+    private String path;
+
+    public AppendEvent(String path) {
+      super(EventType.APPEND);
+      this.path = path;
+    }
+
+    public String getPath() {
+      return path;
+    }
+  }
+
+  /**
+   * Sent when a file, directory, or symlink is deleted.
+   */
+  public static class UnlinkEvent extends Event {
+    private String path;
+    private long timestamp;
+
+    public UnlinkEvent(String path, long timestamp) {
+      super(EventType.UNLINK);
+      this.path = path;
+      this.timestamp = timestamp;
+    }
+
+    public String getPath() {
+      return path;
+    }
+
+    /**
+     * The time when this event occurred, in milliseconds since the epoch.
+     */
+    public long getTimestamp() {
+      return timestamp;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/EventsList.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/EventsList.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/EventsList.java
new file mode 100644
index 0000000..6d02d3c
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/EventsList.java
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.inotify;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+import java.util.List;
+
+/**
+ * Contains a set of events, the transaction ID in the edit log up to which we
+ * read to produce these events, and the first txid we observed when producing
+ * these events (the last of which is for the purpose of determining whether we
+ * have missed events due to edit deletion). Also contains the most recent txid
+ * that the NameNode has sync'ed, so the client can determine how far behind in
+ * the edit log it is.
+ */
+@InterfaceAudience.Private
+public class EventsList {
+  private List<Event> events;
+  private long firstTxid;
+  private long lastTxid;
+  private long syncTxid;
+
+  public EventsList(List<Event> events, long firstTxid, long lastTxid,
+      long syncTxid) {
+    this.events = events;
+    this.firstTxid = firstTxid;
+    this.lastTxid = lastTxid;
+    this.syncTxid = syncTxid;
+  }
+
+  public List<Event> getEvents() {
+    return events;
+  }
+
+  public long getFirstTxid() {
+    return firstTxid;
+  }
+
+  public long getLastTxid() {
+    return lastTxid;
+  }
+
+  public long getSyncTxid() {
+    return syncTxid;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/MissingEventsException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/MissingEventsException.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/MissingEventsException.java
new file mode 100644
index 0000000..e4b51c5
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/inotify/MissingEventsException.java
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.inotify;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class MissingEventsException extends Exception {
+  private static final long serialVersionUID = 1L;
+
+  private long expectedTxid;
+  private long actualTxid;
+
+  public MissingEventsException() {}
+
+  public MissingEventsException(long expectedTxid, long actualTxid) {
+    this.expectedTxid = expectedTxid;
+    this.actualTxid = actualTxid;
+  }
+
+  public long getExpectedTxid() {
+    return expectedTxid;
+  }
+
+  public long getActualTxid() {
+    return actualTxid;
+  }
+
+  @Override
+  public String toString() {
+    return "We expected the next batch of events to start with transaction ID "
+        + expectedTxid + ", but it instead started with transaction ID " +
+        actualTxid + ". Most likely the intervening transactions were cleaned "
+        + "up as part of checkpointing.";
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
index ef0ac55..093afcf 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
@@ -43,10 +43,13 @@ import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.inotify.Event;
+import org.apache.hadoop.hdfs.inotify.EventsList;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants.RollingUpgradeAction;
 import org.apache.hadoop.hdfs.security.token.block.DataEncryptionKey;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenSelector;
+import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp;
 import org.apache.hadoop.hdfs.server.namenode.NotReplicatedYetException;
 import org.apache.hadoop.hdfs.server.namenode.SafeModeException;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
@@ -1372,4 +1375,19 @@ public interface ClientProtocol {
    */
   @Idempotent
   public void checkAccess(String path, FsAction mode) throws IOException;
+
+  /**
+   * Get the highest txid the NameNode knows has been written to the edit
+   * log, or -1 if the NameNode's edit log is not yet open for write. Used as
+   * the starting point for the inotify event stream.
+   */
+  @Idempotent
+  public long getCurrentEditLogTxid() throws IOException;
+
+  /**
+   * Get an ordered list of events corresponding to the edit log transactions
+   * from txid onwards.
+   */
+  @Idempotent
+  public EventsList getEditsFromTxid(long txid) throws IOException;
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
index 40dd8f0..a162ec5 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
@@ -91,12 +91,16 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetBlo
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetBlockLocationsResponseProto.Builder;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetContentSummaryRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetContentSummaryResponseProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetCurrentEditLogTxidRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetCurrentEditLogTxidResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDataEncryptionKeyRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDataEncryptionKeyResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDatanodeReportRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDatanodeReportResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDatanodeStorageReportRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDatanodeStorageReportResponseProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetEditsFromTxidRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetEditsFromTxidResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFileInfoRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFileInfoResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFileLinkInfoRequestProto;
@@ -1408,4 +1412,25 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
     }
     return VOID_CHECKACCESS_RESPONSE;
   }
+
+  public GetCurrentEditLogTxidResponseProto getCurrentEditLogTxid(RpcController controller,
+      GetCurrentEditLogTxidRequestProto req) throws ServiceException {
+    try {
+      return GetCurrentEditLogTxidResponseProto.newBuilder().setTxid(
+          server.getCurrentEditLogTxid()).build();
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
+
+  @Override
+  public GetEditsFromTxidResponseProto getEditsFromTxid(RpcController controller,
+      GetEditsFromTxidRequestProto req) throws ServiceException {
+    try {
+      return PBHelper.convertEditsResponse(server.getEditsFromTxid(
+          req.getTxid()));
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
index 210828d..79c4fcf 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
@@ -43,6 +43,7 @@ import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.inotify.EventsList;
 import org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveEntry;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveInfo;
@@ -95,10 +96,12 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetAdd
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetBlockLocationsRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetBlockLocationsResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetContentSummaryRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetCurrentEditLogTxidRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDataEncryptionKeyRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDataEncryptionKeyResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDatanodeReportRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetDatanodeStorageReportRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetEditsFromTxidRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFileInfoRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFileInfoResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFileLinkInfoRequestProto;
@@ -159,6 +162,7 @@ import org.apache.hadoop.hdfs.protocol.proto.XAttrProtos.RemoveXAttrRequestProto
 import org.apache.hadoop.hdfs.protocol.proto.XAttrProtos.SetXAttrRequestProto;
 import org.apache.hadoop.hdfs.security.token.block.DataEncryptionKey;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp;
 import org.apache.hadoop.hdfs.server.namenode.NotReplicatedYetException;
 import org.apache.hadoop.hdfs.server.namenode.SafeModeException;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
@@ -1430,4 +1434,25 @@ public class ClientNamenodeProtocolTranslatorPB implements
       throw ProtobufHelper.getRemoteException(e);
     }
   }
+
+  public long getCurrentEditLogTxid() throws IOException {
+    GetCurrentEditLogTxidRequestProto req = GetCurrentEditLogTxidRequestProto
+        .getDefaultInstance();
+    try {
+      return rpcProxy.getCurrentEditLogTxid(null, req).getTxid();
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
+  @Override
+  public EventsList getEditsFromTxid(long txid) throws IOException {
+    GetEditsFromTxidRequestProto req = GetEditsFromTxidRequestProto.newBuilder()
+        .setTxid(txid).build();
+    try {
+      return PBHelper.convert(rpcProxy.getEditsFromTxid(null, req));
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
index 4861518..deb52bc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
@@ -45,6 +45,8 @@ import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.StorageType;
+import org.apache.hadoop.hdfs.inotify.Event;
+import org.apache.hadoop.hdfs.inotify.EventsList;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveEntry;
 import org.apache.hadoop.hdfs.protocol.CacheDirectiveInfo;
@@ -95,6 +97,7 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.CacheP
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.CreateFlagProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.DatanodeReportTypeProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.DatanodeStorageReportProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetEditsFromTxidResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.GetFsStatsResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RollingUpgradeActionProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RollingUpgradeInfoProto;
@@ -157,6 +160,7 @@ import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.StorageReportProto;
 import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.StorageTypeProto;
 import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.StorageTypesProto;
 import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.StorageUuidsProto;
+import org.apache.hadoop.hdfs.protocol.proto.InotifyProtos;
 import org.apache.hadoop.hdfs.protocol.proto.JournalProtocolProtos.JournalInfoProto;
 import org.apache.hadoop.hdfs.protocol.proto.XAttrProtos.GetXAttrsResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.XAttrProtos.ListXAttrsResponseProto;
@@ -2334,6 +2338,247 @@ public class PBHelper {
     return new ShmId(shmId.getHi(), shmId.getLo());
   }
 
+  private static Event.CreateEvent.INodeType createTypeConvert(InotifyProtos.INodeType
+      type) {
+    switch (type) {
+    case I_TYPE_DIRECTORY:
+      return Event.CreateEvent.INodeType.DIRECTORY;
+    case I_TYPE_FILE:
+      return Event.CreateEvent.INodeType.FILE;
+    case I_TYPE_SYMLINK:
+      return Event.CreateEvent.INodeType.SYMLINK;
+    default:
+      return null;
+    }
+  }
+
+  private static InotifyProtos.MetadataUpdateType metadataUpdateTypeConvert(
+      Event.MetadataUpdateEvent.MetadataType type) {
+    switch (type) {
+    case TIMES:
+      return InotifyProtos.MetadataUpdateType.META_TYPE_TIMES;
+    case REPLICATION:
+      return InotifyProtos.MetadataUpdateType.META_TYPE_REPLICATION;
+    case OWNER:
+      return InotifyProtos.MetadataUpdateType.META_TYPE_OWNER;
+    case PERMS:
+      return InotifyProtos.MetadataUpdateType.META_TYPE_PERMS;
+    case ACLS:
+      return InotifyProtos.MetadataUpdateType.META_TYPE_ACLS;
+    case XATTRS:
+      return InotifyProtos.MetadataUpdateType.META_TYPE_XATTRS;
+    default:
+      return null;
+    }
+  }
+
+  private static Event.MetadataUpdateEvent.MetadataType metadataUpdateTypeConvert(
+      InotifyProtos.MetadataUpdateType type) {
+    switch (type) {
+    case META_TYPE_TIMES:
+      return Event.MetadataUpdateEvent.MetadataType.TIMES;
+    case META_TYPE_REPLICATION:
+      return Event.MetadataUpdateEvent.MetadataType.REPLICATION;
+    case META_TYPE_OWNER:
+      return Event.MetadataUpdateEvent.MetadataType.OWNER;
+    case META_TYPE_PERMS:
+      return Event.MetadataUpdateEvent.MetadataType.PERMS;
+    case META_TYPE_ACLS:
+      return Event.MetadataUpdateEvent.MetadataType.ACLS;
+    case META_TYPE_XATTRS:
+      return Event.MetadataUpdateEvent.MetadataType.XATTRS;
+    default:
+      return null;
+    }
+  }
+
+  private static InotifyProtos.INodeType createTypeConvert(Event.CreateEvent.INodeType
+      type) {
+    switch (type) {
+    case DIRECTORY:
+      return InotifyProtos.INodeType.I_TYPE_DIRECTORY;
+    case FILE:
+      return InotifyProtos.INodeType.I_TYPE_FILE;
+    case SYMLINK:
+      return InotifyProtos.INodeType.I_TYPE_SYMLINK;
+    default:
+      return null;
+    }
+  }
+
+  public static EventsList convert(GetEditsFromTxidResponseProto resp) throws
+    IOException {
+    List<Event> events = Lists.newArrayList();
+    for (InotifyProtos.EventProto p : resp.getEventsList().getEventsList()) {
+      switch(p.getType()) {
+      case EVENT_CLOSE:
+        InotifyProtos.CloseEventProto close =
+            InotifyProtos.CloseEventProto.parseFrom(p.getContents());
+        events.add(new Event.CloseEvent(close.getPath(), close.getFileSize(),
+            close.getTimestamp()));
+        break;
+      case EVENT_CREATE:
+        InotifyProtos.CreateEventProto create =
+            InotifyProtos.CreateEventProto.parseFrom(p.getContents());
+        events.add(new Event.CreateEvent.Builder()
+            .iNodeType(createTypeConvert(create.getType()))
+            .path(create.getPath())
+            .ctime(create.getCtime())
+            .ownerName(create.getOwnerName())
+            .groupName(create.getGroupName())
+            .perms(convert(create.getPerms()))
+            .replication(create.getReplication())
+            .symlinkTarget(create.getSymlinkTarget().isEmpty() ? null :
+            create.getSymlinkTarget()).build());
+        break;
+      case EVENT_METADATA:
+        InotifyProtos.MetadataUpdateEventProto meta =
+            InotifyProtos.MetadataUpdateEventProto.parseFrom(p.getContents());
+        events.add(new Event.MetadataUpdateEvent.Builder()
+            .path(meta.getPath())
+            .metadataType(metadataUpdateTypeConvert(meta.getType()))
+            .mtime(meta.getMtime())
+            .atime(meta.getAtime())
+            .replication(meta.getReplication())
+            .ownerName(
+                meta.getOwnerName().isEmpty() ? null : meta.getOwnerName())
+            .groupName(
+                meta.getGroupName().isEmpty() ? null : meta.getGroupName())
+            .perms(meta.hasPerms() ? convert(meta.getPerms()) : null)
+            .acls(meta.getAclsList().isEmpty() ? null : convertAclEntry(
+                meta.getAclsList()))
+            .xAttrs(meta.getXAttrsList().isEmpty() ? null : convertXAttrs(
+                meta.getXAttrsList()))
+            .xAttrsRemoved(meta.getXAttrsRemoved())
+            .build());
+        break;
+      case EVENT_RENAME:
+        InotifyProtos.RenameEventProto rename =
+            InotifyProtos.RenameEventProto.parseFrom(p.getContents());
+        events.add(new Event.RenameEvent(rename.getSrcPath(), rename.getDestPath(),
+            rename.getTimestamp()));
+        break;
+      case EVENT_APPEND:
+        InotifyProtos.AppendEventProto reopen =
+            InotifyProtos.AppendEventProto.parseFrom(p.getContents());
+        events.add(new Event.AppendEvent(reopen.getPath()));
+        break;
+      case EVENT_UNLINK:
+        InotifyProtos.UnlinkEventProto unlink =
+            InotifyProtos.UnlinkEventProto.parseFrom(p.getContents());
+        events.add(new Event.UnlinkEvent(unlink.getPath(), unlink.getTimestamp()));
+        break;
+      default:
+        throw new RuntimeException("Unexpected inotify event type: " +
+            p.getType());
+      }
+    }
+    return new EventsList(events, resp.getEventsList().getFirstTxid(),
+        resp.getEventsList().getLastTxid(), resp.getEventsList().getSyncTxid());
+  }
+
+  public static GetEditsFromTxidResponseProto convertEditsResponse(EventsList el) {
+    InotifyProtos.EventsListProto.Builder builder =
+        InotifyProtos.EventsListProto.newBuilder();
+    for (Event e : el.getEvents()) {
+      switch(e.getEventType()) {
+      case CLOSE:
+        Event.CloseEvent ce = (Event.CloseEvent) e;
+        builder.addEvents(InotifyProtos.EventProto.newBuilder()
+            .setType(InotifyProtos.EventType.EVENT_CLOSE)
+            .setContents(
+                InotifyProtos.CloseEventProto.newBuilder()
+                    .setPath(ce.getPath())
+                    .setFileSize(ce.getFileSize())
+                    .setTimestamp(ce.getTimestamp()).build().toByteString()
+            ).build());
+        break;
+      case CREATE:
+        Event.CreateEvent ce2 = (Event.CreateEvent) e;
+        builder.addEvents(InotifyProtos.EventProto.newBuilder()
+            .setType(InotifyProtos.EventType.EVENT_CREATE)
+            .setContents(
+                InotifyProtos.CreateEventProto.newBuilder()
+                    .setType(createTypeConvert(ce2.getiNodeType()))
+                    .setPath(ce2.getPath())
+                    .setCtime(ce2.getCtime())
+                    .setOwnerName(ce2.getOwnerName())
+                    .setGroupName(ce2.getGroupName())
+                    .setPerms(convert(ce2.getPerms()))
+                    .setReplication(ce2.getReplication())
+                    .setSymlinkTarget(ce2.getSymlinkTarget() == null ?
+                        "" : ce2.getSymlinkTarget()).build().toByteString()
+            ).build());
+        break;
+      case METADATA:
+        Event.MetadataUpdateEvent me = (Event.MetadataUpdateEvent) e;
+        InotifyProtos.MetadataUpdateEventProto.Builder metaB =
+            InotifyProtos.MetadataUpdateEventProto.newBuilder()
+                .setPath(me.getPath())
+                .setType(metadataUpdateTypeConvert(me.getMetadataType()))
+                .setMtime(me.getMtime())
+                .setAtime(me.getAtime())
+                .setReplication(me.getReplication())
+                .setOwnerName(me.getOwnerName() == null ? "" :
+                    me.getOwnerName())
+                .setGroupName(me.getGroupName() == null ? "" :
+                    me.getGroupName())
+                .addAllAcls(me.getAcls() == null ?
+                    Lists.<AclEntryProto>newArrayList() :
+                    convertAclEntryProto(me.getAcls()))
+                .addAllXAttrs(me.getxAttrs() == null ?
+                    Lists.<XAttrProto>newArrayList() :
+                    convertXAttrProto(me.getxAttrs()))
+                .setXAttrsRemoved(me.isxAttrsRemoved());
+        if (me.getPerms() != null) {
+          metaB.setPerms(convert(me.getPerms()));
+        }
+        builder.addEvents(InotifyProtos.EventProto.newBuilder()
+            .setType(InotifyProtos.EventType.EVENT_METADATA)
+            .setContents(metaB.build().toByteString())
+            .build());
+        break;
+      case RENAME:
+        Event.RenameEvent re = (Event.RenameEvent) e;
+        builder.addEvents(InotifyProtos.EventProto.newBuilder()
+            .setType(InotifyProtos.EventType.EVENT_RENAME)
+            .setContents(
+                InotifyProtos.RenameEventProto.newBuilder()
+                    .setSrcPath(re.getSrcPath())
+                    .setDestPath(re.getDstPath())
+                    .setTimestamp(re.getTimestamp()).build().toByteString()
+            ).build());
+        break;
+      case APPEND:
+        Event.AppendEvent re2 = (Event.AppendEvent) e;
+        builder.addEvents(InotifyProtos.EventProto.newBuilder()
+            .setType(InotifyProtos.EventType.EVENT_APPEND)
+            .setContents(
+                InotifyProtos.AppendEventProto.newBuilder()
+                    .setPath(re2.getPath()).build().toByteString()
+            ).build());
+        break;
+      case UNLINK:
+        Event.UnlinkEvent ue = (Event.UnlinkEvent) e;
+        builder.addEvents(InotifyProtos.EventProto.newBuilder()
+            .setType(InotifyProtos.EventType.EVENT_UNLINK)
+            .setContents(
+                InotifyProtos.UnlinkEventProto.newBuilder()
+                    .setPath(ue.getPath())
+                    .setTimestamp(ue.getTimestamp()).build().toByteString()
+            ).build());
+        break;
+      default:
+        throw new RuntimeException("Unexpected inotify event: " + e);
+      }
+    }
+    builder.setFirstTxid(el.getFirstTxid());
+    builder.setLastTxid(el.getLastTxid());
+    builder.setSyncTxid(el.getSyncTxid());
+    return GetEditsFromTxidResponseProto.newBuilder().setEventsList(
+        builder.build()).build();
+  }
+
   public static HdfsProtos.CipherSuite convert(CipherSuite suite) {
     switch (suite) {
     case UNKNOWN:

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/IPCLoggerChannel.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/IPCLoggerChannel.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/IPCLoggerChannel.java
index 8588de5..63d77bc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/IPCLoggerChannel.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/client/IPCLoggerChannel.java
@@ -79,7 +79,17 @@ public class IPCLoggerChannel implements AsyncLogger {
   protected final InetSocketAddress addr;
   private QJournalProtocol proxy;
 
-  private final ListeningExecutorService executor;
+  /**
+   * Executes tasks submitted to it serially, on a single thread, in FIFO order
+   * (generally used for write tasks that should not be reordered).
+   */
+  private final ListeningExecutorService singleThreadExecutor;
+  /**
+   * Executes tasks submitted to it in parallel with each other and with those
+   * submitted to singleThreadExecutor (generally used for read tasks that can
+   * be safely reordered and interleaved with writes).
+   */
+  private final ListeningExecutorService parallelExecutor;
   private long ipcSerial = 0;
   private long epoch = -1;
   private long committedTxId = HdfsConstants.INVALID_TXID;
@@ -160,8 +170,10 @@ public class IPCLoggerChannel implements AsyncLogger {
         DFSConfigKeys.DFS_QJOURNAL_QUEUE_SIZE_LIMIT_KEY,
         DFSConfigKeys.DFS_QJOURNAL_QUEUE_SIZE_LIMIT_DEFAULT);
     
-    executor = MoreExecutors.listeningDecorator(
-        createExecutor());
+    singleThreadExecutor = MoreExecutors.listeningDecorator(
+        createSingleThreadExecutor());
+    parallelExecutor = MoreExecutors.listeningDecorator(
+        createParallelExecutor());
     
     metrics = IPCLoggerChannelMetrics.create(this);
   }
@@ -183,7 +195,8 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public void close() {
     // No more tasks may be submitted after this point.
-    executor.shutdown();
+    singleThreadExecutor.shutdown();
+    parallelExecutor.shutdown();
     if (proxy != null) {
       // TODO: this can hang for quite some time if the client
       // is currently in the middle of a call to a downed JN.
@@ -230,15 +243,30 @@ public class IPCLoggerChannel implements AsyncLogger {
    * Separated out for easy overriding in tests.
    */
   @VisibleForTesting
-  protected ExecutorService createExecutor() {
+  protected ExecutorService createSingleThreadExecutor() {
     return Executors.newSingleThreadExecutor(
         new ThreadFactoryBuilder()
           .setDaemon(true)
-          .setNameFormat("Logger channel to " + addr)
+          .setNameFormat("Logger channel (from single-thread executor) to " +
+              addr)
           .setUncaughtExceptionHandler(
               UncaughtExceptionHandlers.systemExit())
           .build());
   }
+
+  /**
+   * Separated out for easy overriding in tests.
+   */
+  @VisibleForTesting
+  protected ExecutorService createParallelExecutor() {
+    return Executors.newCachedThreadPool(
+        new ThreadFactoryBuilder()
+            .setDaemon(true)
+            .setNameFormat("Logger channel (from parallel executor) to " + addr)
+            .setUncaughtExceptionHandler(
+                UncaughtExceptionHandlers.systemExit())
+            .build());
+  }
   
   @Override
   public URL buildURLToFetchLogs(long segmentTxId) {
@@ -286,7 +314,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @VisibleForTesting
   void waitForAllPendingCalls() throws InterruptedException {
     try {
-      executor.submit(new Runnable() {
+      singleThreadExecutor.submit(new Runnable() {
         @Override
         public void run() {
         }
@@ -299,7 +327,7 @@ public class IPCLoggerChannel implements AsyncLogger {
 
   @Override
   public ListenableFuture<Boolean> isFormatted() {
-    return executor.submit(new Callable<Boolean>() {
+    return singleThreadExecutor.submit(new Callable<Boolean>() {
       @Override
       public Boolean call() throws IOException {
         return getProxy().isFormatted(journalId);
@@ -309,7 +337,7 @@ public class IPCLoggerChannel implements AsyncLogger {
 
   @Override
   public ListenableFuture<GetJournalStateResponseProto> getJournalState() {
-    return executor.submit(new Callable<GetJournalStateResponseProto>() {
+    return singleThreadExecutor.submit(new Callable<GetJournalStateResponseProto>() {
       @Override
       public GetJournalStateResponseProto call() throws IOException {
         GetJournalStateResponseProto ret =
@@ -323,7 +351,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<NewEpochResponseProto> newEpoch(
       final long epoch) {
-    return executor.submit(new Callable<NewEpochResponseProto>() {
+    return singleThreadExecutor.submit(new Callable<NewEpochResponseProto>() {
       @Override
       public NewEpochResponseProto call() throws IOException {
         return getProxy().newEpoch(journalId, nsInfo, epoch);
@@ -347,7 +375,7 @@ public class IPCLoggerChannel implements AsyncLogger {
     
     ListenableFuture<Void> ret = null;
     try {
-      ret = executor.submit(new Callable<Void>() {
+      ret = singleThreadExecutor.submit(new Callable<Void>() {
         @Override
         public Void call() throws IOException {
           throwIfOutOfSync();
@@ -464,7 +492,7 @@ public class IPCLoggerChannel implements AsyncLogger {
 
   @Override
   public ListenableFuture<Void> format(final NamespaceInfo nsInfo) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws Exception {
         getProxy().format(journalId, nsInfo);
@@ -476,7 +504,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<Void> startLogSegment(final long txid,
       final int layoutVersion) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().startLogSegment(createReqInfo(), txid, layoutVersion);
@@ -497,7 +525,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<Void> finalizeLogSegment(
       final long startTxId, final long endTxId) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         throwIfOutOfSync();
@@ -510,7 +538,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   
   @Override
   public ListenableFuture<Void> purgeLogsOlderThan(final long minTxIdToKeep) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws Exception {
         getProxy().purgeLogsOlderThan(createReqInfo(), minTxIdToKeep);
@@ -522,7 +550,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<RemoteEditLogManifest> getEditLogManifest(
       final long fromTxnId, final boolean inProgressOk) {
-    return executor.submit(new Callable<RemoteEditLogManifest>() {
+    return parallelExecutor.submit(new Callable<RemoteEditLogManifest>() {
       @Override
       public RemoteEditLogManifest call() throws IOException {
         GetEditLogManifestResponseProto ret = getProxy().getEditLogManifest(
@@ -538,7 +566,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<PrepareRecoveryResponseProto> prepareRecovery(
       final long segmentTxId) {
-    return executor.submit(new Callable<PrepareRecoveryResponseProto>() {
+    return singleThreadExecutor.submit(new Callable<PrepareRecoveryResponseProto>() {
       @Override
       public PrepareRecoveryResponseProto call() throws IOException {
         if (!hasHttpServerEndPoint()) {
@@ -556,7 +584,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<Void> acceptRecovery(
       final SegmentStateProto log, final URL url) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().acceptRecovery(createReqInfo(), log, url);
@@ -567,7 +595,7 @@ public class IPCLoggerChannel implements AsyncLogger {
 
   @Override
   public ListenableFuture<Void> discardSegments(final long startTxId) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().discardSegments(journalId, startTxId);
@@ -578,7 +606,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   
   @Override
   public ListenableFuture<Void> doPreUpgrade() {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().doPreUpgrade(journalId);
@@ -589,7 +617,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   
   @Override
   public ListenableFuture<Void> doUpgrade(final StorageInfo sInfo) {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().doUpgrade(journalId, sInfo);
@@ -600,7 +628,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   
   @Override
   public ListenableFuture<Void> doFinalize() {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().doFinalize(journalId);
@@ -612,7 +640,7 @@ public class IPCLoggerChannel implements AsyncLogger {
   @Override
   public ListenableFuture<Boolean> canRollBack(final StorageInfo storage,
       final StorageInfo prevStorage, final int targetLayoutVersion) {
-    return executor.submit(new Callable<Boolean>() {
+    return singleThreadExecutor.submit(new Callable<Boolean>() {
       @Override
       public Boolean call() throws IOException {
         return getProxy().canRollBack(journalId, storage, prevStorage,
@@ -623,7 +651,7 @@ public class IPCLoggerChannel implements AsyncLogger {
 
   @Override
   public ListenableFuture<Void> doRollback() {
-    return executor.submit(new Callable<Void>() {
+    return singleThreadExecutor.submit(new Callable<Void>() {
       @Override
       public Void call() throws IOException {
         getProxy().doRollback(journalId);
@@ -631,10 +659,10 @@ public class IPCLoggerChannel implements AsyncLogger {
       }
     });
   }
-  
+
   @Override
   public ListenableFuture<Long> getJournalCTime() {
-    return executor.submit(new Callable<Long>() {
+    return singleThreadExecutor.submit(new Callable<Long>() {
       @Override
       public Long call() throws IOException {
         return getProxy().getJournalCTime(journalId);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java
index c117ee8..71c361d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java
@@ -651,7 +651,8 @@ public class Journal implements Closeable {
         }
       }
       if (log != null && log.isInProgress()) {
-        logs.add(new RemoteEditLog(log.getStartTxId(), getHighestWrittenTxId()));
+        logs.add(new RemoteEditLog(log.getStartTxId(), getHighestWrittenTxId(),
+            true));
       }
     }
     

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogBackupInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogBackupInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogBackupInputStream.java
index 0f63966..3649437 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogBackupInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogBackupInputStream.java
@@ -147,4 +147,9 @@ class EditLogBackupInputStream extends EditLogInputStream {
   public void setMaxOpSize(int maxOpSize) {
     reader.setMaxOpSize(maxOpSize);
   }
+
+  @Override
+  public boolean isLocalLog() {
+    return true;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileInputStream.java
index 748c694..c732181 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileInputStream.java
@@ -507,4 +507,9 @@ public class EditLogFileInputStream extends EditLogInputStream {
       reader.setMaxOpSize(maxOpSize);
     }
   }
+
+  @Override
+  public boolean isLocalLog() {
+    return log instanceof FileLog;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogInputStream.java
index 969668d..ac58616 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogInputStream.java
@@ -203,4 +203,10 @@ public abstract class EditLogInputStream implements Closeable {
    * Set the maximum opcode size in bytes.
    */
   public abstract void setMaxOpSize(int maxOpSize);
+
+  /**
+   * Returns true if we are currently reading the log from a local disk or an
+   * even faster data source (e.g. a byte buffer).
+   */
+  public abstract boolean isLocalLog();
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java
index 5b594ce..8b44628 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java
@@ -188,6 +188,13 @@ public class FSEditLog implements LogsPurgeable {
    */
   private final List<URI> sharedEditsDirs;
 
+  /**
+   * Take this lock when adding journals to or closing the JournalSet. Allows
+   * us to ensure that the JournalSet isn't closed or updated underneath us
+   * in selectInputStreams().
+   */
+  private final Object journalSetLock = new Object();
+
   private static class TransactionId {
     public long txid;
 
@@ -252,20 +259,22 @@ public class FSEditLog implements LogsPurgeable {
         DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_KEY,
         DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_DEFAULT);
 
-    journalSet = new JournalSet(minimumRedundantJournals);
-
-    for (URI u : dirs) {
-      boolean required = FSNamesystem.getRequiredNamespaceEditsDirs(conf)
-          .contains(u);
-      if (u.getScheme().equals(NNStorage.LOCAL_URI_SCHEME)) {
-        StorageDirectory sd = storage.getStorageDirectory(u);
-        if (sd != null) {
-          journalSet.add(new FileJournalManager(conf, sd, storage),
-              required, sharedEditsDirs.contains(u));
+    synchronized(journalSetLock) {
+      journalSet = new JournalSet(minimumRedundantJournals);
+
+      for (URI u : dirs) {
+        boolean required = FSNamesystem.getRequiredNamespaceEditsDirs(conf)
+            .contains(u);
+        if (u.getScheme().equals(NNStorage.LOCAL_URI_SCHEME)) {
+          StorageDirectory sd = storage.getStorageDirectory(u);
+          if (sd != null) {
+            journalSet.add(new FileJournalManager(conf, sd, storage),
+                required, sharedEditsDirs.contains(u));
+          }
+        } else {
+          journalSet.add(createJournal(u), required,
+              sharedEditsDirs.contains(u));
         }
-      } else {
-        journalSet.add(createJournal(u), required,
-            sharedEditsDirs.contains(u));
       }
     }
  
@@ -349,7 +358,9 @@ public class FSEditLog implements LogsPurgeable {
     } finally {
       if (journalSet != null && !journalSet.isEmpty()) {
         try {
-          journalSet.close();
+          synchronized(journalSetLock) {
+            journalSet.close();
+          }
         } catch (IOException ioe) {
           LOG.warn("Error closing journalSet", ioe);
         }
@@ -606,7 +617,9 @@ public class FSEditLog implements LogsPurgeable {
                 "due to " + e.getMessage() + ". " +
                 "Unsynced transactions: " + (txid - synctxid);
             LOG.fatal(msg, new Exception());
-            IOUtils.cleanup(LOG, journalSet);
+            synchronized(journalSetLock) {
+              IOUtils.cleanup(LOG, journalSet);
+            }
             terminate(1, msg);
           }
         } finally {
@@ -630,7 +643,9 @@ public class FSEditLog implements LogsPurgeable {
               "Could not sync enough journals to persistent storage. "
               + "Unsynced transactions: " + (txid - synctxid);
           LOG.fatal(msg, new Exception());
-          IOUtils.cleanup(LOG, journalSet);
+          synchronized(journalSetLock) {
+            IOUtils.cleanup(LOG, journalSet);
+          }
           terminate(1, msg);
         }
       }
@@ -1268,9 +1283,8 @@ public class FSEditLog implements LogsPurgeable {
 
   /**
    * Return the txid of the last synced transaction.
-   * For test use only
    */
-  synchronized long getSyncTxId() {
+  public synchronized long getSyncTxId() {
     return synctxid;
   }
 
@@ -1307,7 +1321,9 @@ public class FSEditLog implements LogsPurgeable {
     
     LOG.info("Registering new backup node: " + bnReg);
     BackupJournalManager bjm = new BackupJournalManager(bnReg, nnReg);
-    journalSet.add(bjm, true);
+    synchronized(journalSetLock) {
+      journalSet.add(bjm, false);
+    }
   }
   
   synchronized void releaseBackupStream(NamenodeRegistration registration)
@@ -1315,7 +1331,9 @@ public class FSEditLog implements LogsPurgeable {
     BackupJournalManager bjm = this.findBackupJournal(registration);
     if (bjm != null) {
       LOG.info("Removing backup journal " + bjm);
-      journalSet.remove(bjm);
+      synchronized(journalSetLock) {
+        journalSet.remove(bjm);
+      }
     }
   }
   
@@ -1443,11 +1461,16 @@ public class FSEditLog implements LogsPurgeable {
    * @param recovery recovery context
    * @param inProgressOk set to true if in-progress streams are OK
    */
-  public synchronized Collection<EditLogInputStream> selectInputStreams(
+  public Collection<EditLogInputStream> selectInputStreams(
       long fromTxId, long toAtLeastTxId, MetaRecoveryContext recovery,
       boolean inProgressOk) throws IOException {
+
     List<EditLogInputStream> streams = new ArrayList<EditLogInputStream>();
-    selectInputStreams(streams, fromTxId, inProgressOk);
+    synchronized(journalSetLock) {
+      Preconditions.checkState(journalSet.isOpen(), "Cannot call " +
+          "selectInputStreams() on closed FSEditLog");
+      selectInputStreams(streams, fromTxId, inProgressOk);
+    }
 
     try {
       checkForGaps(streams, fromTxId, toAtLeastTxId, inProgressOk);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/958c9b50/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java
index ccb5da8..101c42c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java
@@ -186,17 +186,27 @@ public class FileJournalManager implements JournalManager {
     List<EditLogFile> allLogFiles = matchEditLogs(currentDir);
     List<RemoteEditLog> ret = Lists.newArrayListWithCapacity(
         allLogFiles.size());
-
     for (EditLogFile elf : allLogFiles) {
       if (elf.hasCorruptHeader() || (!inProgressOk && elf.isInProgress())) {
         continue;
       }
+      if (elf.isInProgress()) {
+        try {
+          elf.validateLog();
+        } catch (IOException e) {
+          LOG.error("got IOException while trying to validate header of " +
+              elf + ".  Skipping.", e);
+          continue;
+        }
+      }
       if (elf.getFirstTxId() >= firstTxId) {
-        ret.add(new RemoteEditLog(elf.firstTxId, elf.lastTxId));
+        ret.add(new RemoteEditLog(elf.firstTxId, elf.lastTxId,
+            elf.isInProgress()));
       } else if (elf.getFirstTxId() < firstTxId && firstTxId <= elf.getLastTxId()) {
         // If the firstTxId is in the middle of an edit log segment. Return this
         // anyway and let the caller figure out whether it wants to use it.
-        ret.add(new RemoteEditLog(elf.firstTxId, elf.lastTxId));
+        ret.add(new RemoteEditLog(elf.firstTxId, elf.lastTxId,
+            elf.isInProgress()));
       }
     }
     


Mime
View raw message