HDFS-3201. Add GetJournalEditServlet for downloading edit logs from journal service. Contributed by Brandon Li

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-3092@1327830 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.HDFS-3092.txt

@@ -20,6 +20,9 @@ HDFS-3092 branch changes
     HDFS-3196. Add Journal and JournalDiskWriter for journal service.
     (szetszwo)
 
+    HDFS-3201. Add GetJournalEditServlet for downloading edit logs from journal
+    service.  (Brandon Li via szetszwo)
+
   IMPROVEMENTS
 
   OPTIMIZATIONS

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -202,12 +202,13 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final String  DFS_CLIENT_LOCAL_INTERFACES = "dfs.client.local.interfaces";
   
   // This is a comma separated host:port list of addresses hosting the journal service
-  public static final String  DFS_JOURNAL_ADDRESS_KEY = "dfs.journalnode.addresses";
+  public static final String  DFS_JOURNAL_ADDRESS_KEY = "dfs.journal.addresses";
   public static final String  DFS_JOURNAL_EDITS_DIR_KEY = "dfs.journal.edits.dir";
   public static final String  DFS_JOURNAL_HTTPS_PORT_KEY = "dfs.journal.https-port";
   public static final int     DFS_JOURNAL_HTTPS_PORT_DEFAULT = 50510;
   public static final String  DFS_JOURNAL_KRB_HTTPS_USER_NAME_KEY = "dfs.journal.kerberos.https.principal";
   public static final String  DFS_JOURNAL_KEYTAB_FILE_KEY = "dfs.journal.keytab.file";
+  public static final String  DFS_JOURNAL_USER_NAME_KEY = "dfs.journal.kerberos.principal";
 
   // Much code in hdfs is not yet updated to use these keys.
   public static final String  DFS_CLIENT_BLOCK_WRITE_LOCATEFOLLOWINGBLOCK_RETRIES_KEY = "dfs.client.block.write.locateFollowingBlock.retries";

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/journalservice/GetJournalEditServlet.java

@@ -0,0 +1,163 @@
+/**
+ * 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.server.journalservice;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+
+import javax.servlet.ServletContext;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.server.common.JspHelper;
+import org.apache.hadoop.hdfs.server.namenode.GetImageServlet;
+import org.apache.hadoop.hdfs.server.namenode.GetImageServlet.GetImageParams;
+import org.apache.hadoop.hdfs.server.namenode.NNStorage;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
+import org.apache.hadoop.hdfs.server.namenode.TransferFsImage;
+import org.apache.hadoop.hdfs.util.DataTransferThrottler;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.StringUtils;
+
+/**
+ * This class is used by the lagging Journal service to retrieve edit file from
+ * another Journal service for sync up.
+ */
+@InterfaceAudience.Private
+public class GetJournalEditServlet extends HttpServlet {
+
+  private static final long serialVersionUID = -4635891628211723009L;
+  private static final Log LOG = LogFactory.getLog(GetJournalEditServlet.class);
+
+  // TODO: create security tests
+  protected boolean isValidRequestor(String remoteUser, Configuration conf)
+      throws IOException {
+    if (remoteUser == null) { // This really shouldn't happen...
+      LOG.warn("Received null remoteUser while authorizing access to GetJournalEditServlet");
+      return false;
+    }
+
+    String[] validRequestors = {
+        SecurityUtil.getServerPrincipal(
+            conf.get(DFSConfigKeys.DFS_NAMENODE_KRB_HTTPS_USER_NAME_KEY),
+            NameNode.getAddress(conf).getHostName()),
+        SecurityUtil.getServerPrincipal(conf
+            .get(DFSConfigKeys.DFS_NAMENODE_USER_NAME_KEY), NameNode
+            .getAddress(conf).getHostName()),
+        SecurityUtil.getServerPrincipal(
+            conf.get(DFSConfigKeys.DFS_JOURNAL_KRB_HTTPS_USER_NAME_KEY),
+            NameNode.getAddress(conf).getHostName()),
+        SecurityUtil.getServerPrincipal(conf
+            .get(DFSConfigKeys.DFS_JOURNAL_USER_NAME_KEY),
+            NameNode.getAddress(conf).getHostName()) };
+
+    for (String v : validRequestors) {
+      if (v != null && v.equals(remoteUser)) {
+        if (LOG.isDebugEnabled())
+          LOG.debug("isValidRequestor is allowing: " + remoteUser);
+        return true;
+      }
+    }
+    if (LOG.isDebugEnabled())
+      LOG.debug("isValidRequestor is rejecting: " + remoteUser);
+    return false;
+  }
+
+  @Override
+  public void doGet(final HttpServletRequest request,
+      final HttpServletResponse response) throws ServletException, IOException {
+    try {
+      ServletContext context = getServletContext();
+      final NNStorage storage = JournalHttpServer
+          .getJournalFromContext(context).getStorage();
+
+      final GetImageParams parsedParams = new GetImageParams(request, response);
+
+      final Configuration conf = (Configuration) getServletContext()
+          .getAttribute(JspHelper.CURRENT_CONF);
+
+      if (UserGroupInformation.isSecurityEnabled()
+          && !isValidRequestor(request.getRemoteUser(), conf)) {
+        response
+            .sendError(HttpServletResponse.SC_FORBIDDEN,
+                "Only Namenode and another Journal service may access this servlet");
+        LOG.warn("Received non-NN/Journal request for edits from "
+            + request.getRemoteHost());
+        return;
+      }
+
+      String myStorageInfoString = storage.toColonSeparatedString();
+      String theirStorageInfoString = parsedParams.getStorageInfoString();
+      if (theirStorageInfoString != null
+          && !myStorageInfoString.equals(theirStorageInfoString)) {
+        response
+            .sendError(HttpServletResponse.SC_FORBIDDEN,
+                "This node has storage info " + myStorageInfoString
+                    + " but the requesting node expected "
+                    + theirStorageInfoString);
+        LOG.warn("Received an invalid request file transfer request "
+            + " with storage info " + theirStorageInfoString);
+        return;
+      }
+
+      UserGroupInformation.getCurrentUser().doAs(
+          new PrivilegedExceptionAction<Void>() {
+            @Override
+            public Void run() throws Exception {
+              if (parsedParams.isGetEdit()) {
+                long startTxId = parsedParams.getStartTxId();
+                long endTxId = parsedParams.getEndTxId();
+                File editFile = storage.findFinalizedEditsFile(startTxId,
+                    endTxId);
+
+                GetImageServlet.setVerificationHeaders(response, editFile);
+                GetImageServlet.setFileNameHeaders(response, editFile);
+                
+                DataTransferThrottler throttler = GetImageServlet.getThrottler(conf);
+
+                // send edits
+                TransferFsImage.getFileServer(response.getOutputStream(),
+                    editFile, throttler);
+              } else {
+                response
+                    .sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED,
+                        "The server only accepts getedit request. This request is not getedit.");
+              }
+              return null;
+            }
+          });
+
+    } catch (Exception ie) {
+      String errMsg = "getedit failed. " + StringUtils.stringifyException(ie);
+      response.sendError(HttpServletResponse.SC_GONE, errMsg);
+      throw new IOException(errMsg);
+    } finally {
+      response.getOutputStream().close();
+    }
+  }
+
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/journalservice/Journal.java

@@ -79,10 +79,10 @@ class Journal {
     return isFormatted;
   }
   
-  StorageInfo getStorageInfo() {
+  NNStorage getStorage() {
     return image.getStorage();
   }
-
+  
   synchronized void verifyVersion(JournalService service, NamespaceInfo info
       ) throws IOException {
     if (!isFormatted) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/journalservice/JournalHttpServer.java

@@ -34,7 +34,9 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.server.common.JspHelper;
-import org.apache.hadoop.hdfs.server.namenode.NNStorage;
+import org.apache.hadoop.hdfs.server.namenode.TransferFsImage;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 
 import org.apache.hadoop.http.HttpServer;
 import org.apache.hadoop.net.NetUtils;
@@ -47,34 +49,30 @@ import org.apache.hadoop.security.authorize.AccessControlList;
  */
 @InterfaceAudience.Private
 public class JournalHttpServer {
+  public static final Log LOG = LogFactory.getLog(JournalHttpServer.class);
+
+  public static final String JOURNAL_ATTRIBUTE_KEY = "localjournal";
 
   private HttpServer httpServer;
   private InetSocketAddress httpAddress;
   private String infoBindAddress;
   private int infoPort;
-  private int imagePort;
+  private int httpsPort;
+  private Journal localJournal;
 
   private final Configuration conf;
 
-  public static final Log LOG = LogFactory.getLog(JournalHttpServer.class
-      .getName());
-
-  public static final String NNSTORAGE_ATTRIBUTE_KEY = "name.system.storage";
-  protected static final String NAMENODE_ATTRIBUTE_KEY = "name.node";
-
-  private NNStorage storage = null;
-
-  public JournalHttpServer(Configuration conf, NNStorage storage,
+  JournalHttpServer(Configuration conf, Journal journal,
       InetSocketAddress bindAddress) throws Exception {
     this.conf = conf;
-    this.storage = storage;
+    this.localJournal = journal;
     this.httpAddress = bindAddress;
   }
 
-  public void start() throws IOException {
+  void start() throws IOException {
     infoBindAddress = httpAddress.getHostName();
 
-    // initialize the webserver for uploading files.
+    // initialize the webserver for uploading/downloading files.
     // Kerberized SSL servers must be run from the host principal...
     UserGroupInformation httpUGI = UserGroupInformation
         .loginUserFromKeytabAndReturnUGI(SecurityUtil.getServerPrincipal(
@@ -99,44 +97,79 @@ public class JournalHttpServer {
                     + ":"
                     + conf.getInt(DFS_JOURNAL_HTTPS_PORT_KEY,
                         DFS_JOURNAL_HTTPS_PORT_DEFAULT));
-            imagePort = secInfoSocAddr.getPort();
+            httpsPort = secInfoSocAddr.getPort();
             httpServer.addSslListener(secInfoSocAddr, conf, false, true);
           }
-          httpServer.setAttribute("journal.node", JournalHttpServer.this);
-          httpServer.setAttribute("name.system.storage", storage);
+          httpServer.setAttribute(JOURNAL_ATTRIBUTE_KEY, localJournal);
           httpServer.setAttribute(JspHelper.CURRENT_CONF, conf);
+          // use "/getimage" because GetJournalEditServlet uses some
+          // GetImageServlet methods.
+          // TODO: change getimage to getedit
+          httpServer.addInternalServlet("getimage", "/getimage",
+              GetJournalEditServlet.class, true);
           httpServer.start();
           return httpServer;
         }
       });
     } catch (InterruptedException e) {
-      throw new RuntimeException(e);
+      throw new IOException(e);
     }
 
-    LOG.info("Journal web server init done");
     // The web-server port can be ephemeral... ensure we have the correct info
     infoPort = httpServer.getPort();
     if (!UserGroupInformation.isSecurityEnabled()) {
-      imagePort = infoPort;
+      httpsPort = infoPort;
     }
 
-    LOG.info("Journal Web-server up at: " + infoBindAddress + ":"
-        + infoPort);
-    LOG.info("Journal image servlet up at: " + infoBindAddress + ":"
-        + imagePort);
+    LOG.info("Journal Web-server up at: " + infoBindAddress + ":" + infoPort
+        + " and https port is: " + httpsPort);
   }
 
-  public void stop() throws Exception {
+  void stop() throws Exception {
     if (httpServer != null) {
       httpServer.stop();
     }
   }
 
-  public static NNStorage getNNStorageFromContext(ServletContext context) {
-    return (NNStorage) context.getAttribute(NNSTORAGE_ATTRIBUTE_KEY);
+  public static Journal getJournalFromContext(ServletContext context) {
+    return (Journal) context.getAttribute(JOURNAL_ATTRIBUTE_KEY);
   }
 
   public static Configuration getConfFromContext(ServletContext context) {
     return (Configuration) context.getAttribute(JspHelper.CURRENT_CONF);
   }
+
+  /**
+   * Download <code>edits</code> files from another journal service
+   * 
+   * @return true if a new image has been downloaded and needs to be loaded
+   * @throws IOException
+   */
+  public boolean downloadEditFiles(final String jnHostPort,
+      final RemoteEditLogManifest manifest) throws IOException {
+
+    // Sanity check manifest
+    if (manifest.getLogs().isEmpty()) {
+      throw new IOException("Found no edit logs to download");
+    }
+
+    try {
+      Boolean b = UserGroupInformation.getCurrentUser().doAs(
+          new PrivilegedExceptionAction<Boolean>() {
+
+            @Override
+            public Boolean run() throws Exception {
+              // get edits file
+              for (RemoteEditLog log : manifest.getLogs()) {
+                TransferFsImage.downloadEditsToStorage(jnHostPort, log,
+                    localJournal.getStorage());
+              }
+              return true;
+            }
+          });
+      return b.booleanValue();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/journalservice/JournalService.java

@@ -187,7 +187,7 @@ public class JournalService implements JournalProtocol {
     if (registration == null) {
       registration = new NamenodeRegistration(
           NetUtils.getHostPortString(rpcServer.getListenerAddress()), "",
-          journal.getStorageInfo(), NamenodeRole.BACKUP);
+          journal.getStorage(), NamenodeRole.BACKUP);
     }
     return registration;
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/GetImageServlet.java

@@ -195,7 +195,7 @@ public class GetImageServlet extends HttpServlet {
     }
   }
   
-  private static void setFileNameHeaders(HttpServletResponse response,
+  public static void setFileNameHeaders(HttpServletResponse response,
       File file) {
     response.setHeader(CONTENT_DISPOSITION, "attachment; filename=" +
         file.getName());
@@ -207,7 +207,7 @@ public class GetImageServlet extends HttpServlet {
    * @param conf configuration
    * @return a data transfer throttler
    */
-  private final DataTransferThrottler getThrottler(Configuration conf) {
+  public static DataTransferThrottler getThrottler(Configuration conf) {
     long transferBandwidth = 
       conf.getLong(DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY,
                    DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_DEFAULT);
@@ -253,7 +253,7 @@ public class GetImageServlet extends HttpServlet {
    * Set headers for content length, and, if available, md5.
    * @throws IOException 
    */
-  private void setVerificationHeaders(HttpServletResponse response, File file)
+  public static void setVerificationHeaders(HttpServletResponse response, File file)
   throws IOException {
     response.setHeader(TransferFsImage.CONTENT_LENGTH,
         String.valueOf(file.length()));
@@ -294,7 +294,7 @@ public class GetImageServlet extends HttpServlet {
   }
 
   
-  static class GetImageParams {
+  public static class GetImageParams {
     private boolean isGetImage;
     private boolean isGetEdit;
     private boolean isPutImage;
@@ -373,7 +373,7 @@ public class GetImageServlet extends HttpServlet {
       return endTxId;
     }
 
-    boolean isGetEdit() {
+    public boolean isGetEdit() {
       return isGetEdit;
     }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java

@@ -734,7 +734,7 @@ public class NNStorage extends Storage implements Closeable {
   /**
    * Return the first readable finalized edits file for the given txid.
    */
-  File findFinalizedEditsFile(long startTxId, long endTxId)
+  public File findFinalizedEditsFile(long startTxId, long endTxId)
   throws IOException {
     File ret = findFile(NameNodeDirType.EDITS,
         getFinalizedEditsFileName(startTxId, endTxId));

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java

@@ -79,7 +79,7 @@ public class TransferFsImage {
     return hash;
   }
   
-  static void downloadEditsToStorage(String fsName, RemoteEditLog log,
+  public static void downloadEditsToStorage(String fsName, RemoteEditLog log,
       NNStorage dstStorage) throws IOException {
     assert log.getStartTxId() > 0 && log.getEndTxId() > 0 :
       "bad log: " + log;
@@ -146,7 +146,7 @@ public class TransferFsImage {
    * A server-side method to respond to a getfile http request
    * Copies the contents of the local file into the output stream.
    */
-  static void getFileServer(OutputStream outstream, File localfile,
+  public static void getFileServer(OutputStream outstream, File localfile,
       DataTransferThrottler throttler) 
     throws IOException {
     byte buf[] = new byte[HdfsConstants.IO_FILE_BUFFER_SIZE];

hadoop-hdfs-project/hadoop-hdfs/src/main/webapps/journal/journalstatus.jsp

@@ -36,9 +36,6 @@
 <h1>JournalNode</h1>
 <%= JspHelper.getVersionTable() %>
 <hr />
-<pre>
-<%= application.getAttribute("journal.node").toString() %>
-</pre>
 
 <br />
 <b><a href="/logs/">Logs</a></b>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/journalservice/TestJournal.java

@@ -44,7 +44,7 @@ public class TestJournal {
   public void testFormat() throws Exception {
     final Configuration conf = newConf("testFormat");
     final Journal j = new Journal(conf);
-    LOG.info("Initial  : " + j.getStorageInfo());
+    LOG.info("Initial  : " + j.getStorage());
     Assert.assertFalse(j.isFormatted());
 
     //format
@@ -53,7 +53,7 @@ public class TestJournal {
     j.format(namespaceId, clusterId);
     Assert.assertTrue(j.isFormatted());
 
-    final StorageInfo info = j.getStorageInfo();
+    final StorageInfo info = j.getStorage();
     LOG.info("Formatted: " + info);
     
     Assert.assertEquals(namespaceId, info.getNamespaceID());
@@ -61,7 +61,7 @@ public class TestJournal {
     j.close();
     
     //create another Journal object
-    final StorageInfo another = new Journal(conf).getStorageInfo();
+    final StorageInfo another = new Journal(conf).getStorage();
     Assert.assertEquals(info.toString(), another.toString());
   }
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/journalservice/TestJournalHttpServer.java

@@ -21,10 +21,8 @@ import static org.junit.Assert.*;
 
 import java.io.File;
 import java.io.IOException;
-import java.net.URI;
+import java.net.InetSocketAddress;
 import java.net.URL;
-import java.util.ArrayList;
-import java.util.List;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -35,16 +33,21 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSTestUtil;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.NameNodeProxies;
+import org.apache.hadoop.hdfs.server.common.StorageInfo;
 import org.apache.hadoop.hdfs.server.journalservice.JournalHttpServer;
-import org.apache.hadoop.hdfs.server.namenode.NNStorage;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
+import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.log4j.Level;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class TestJournalHttpServer {
-  public static final Log LOG = LogFactory
-      .getLog(TestJournalHttpServer.class);
+  public static final Log LOG = LogFactory.getLog(TestJournalHttpServer.class);
 
   static {
     ((Log4JLogger) JournalHttpServer.LOG).getLogger().setLevel(Level.ALL);
@@ -52,7 +55,7 @@ public class TestJournalHttpServer {
 
   private Configuration conf;
   private File hdfsDir = null;
-  private File path1;
+  private File path1, path2;
 
   @Before
   public void setUp() throws Exception {
@@ -67,24 +70,20 @@ public class TestJournalHttpServer {
     hdfsDir.mkdirs();
     // TODO: remove the manual setting storage when JN is fully implemented
     path1 = new File(hdfsDir, "jn1dir");
+    path2 = new File(hdfsDir, "jn2dir");
     path1.mkdir();
-    if (!path1.exists()) {
+    path2.mkdir();
+    if (!path1.exists() || !path2.exists()) {
       throw new IOException("Couldn't create path in "
           + hdfsDir.getAbsolutePath());
     }
 
     System.out.println("configuring hdfsdir is " + hdfsDir.getAbsolutePath()
-        + "; jn1Dir = " + path1.getPath());
-
-    File path1current = new File(path1, "current");
-    path1current.mkdir();
-    if (!path1current.exists()) {
-      throw new IOException("Couldn't create path " + path1current);
-    }
+        + "; jn1Dir = " + path1.getPath() + "; jn2Dir = " + path2.getPath());
   }
 
   /**
-   * Test JN Http Server
+   * Test Journal service Http Server
    * 
    * @throws Exception
    */
@@ -97,12 +96,7 @@ public class TestJournalHttpServer {
       cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
 
       conf.set(DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY, path1.getPath());
-      // TODO: remove the manual setting storage when JN is fully implemented
-      URI uri = new URI(new String("file:" + path1.getPath()));
-      List<URI> editsDirs = new ArrayList<URI>();
-      editsDirs.add(uri);
-      NNStorage storage = new NNStorage(conf, new ArrayList<URI>(), editsDirs);
-      jns1 = new JournalHttpServer(conf, storage,
+      jns1 = new JournalHttpServer(conf, new Journal(conf),
           NetUtils.createSocketAddr("localhost:50200"));
       jns1.start();
 
@@ -120,4 +114,82 @@ public class TestJournalHttpServer {
         cluster.shutdown();
     }
   }
+  
+  //TODO: remove this method when the same rpc is supported by journal service
+  private RemoteEditLogManifest editsToDownload(InetSocketAddress srcRpcAddr,
+      long txid) throws IOException {
+
+    NamenodeProtocol namenode = NameNodeProxies.createNonHAProxy(conf,
+        srcRpcAddr, NamenodeProtocol.class,
+        UserGroupInformation.getCurrentUser(), true).getProxy();
+
+    // get all edit segments
+    RemoteEditLogManifest manifest = namenode.getEditLogManifest(txid);
+
+    return manifest;
+  }
+
+  /**
+   * Test lagging Journal service copies edit segments from another Journal
+   * service
+   * 
+   * @throws Exception
+   */
+  @Test
+  public void testCopyEdits() throws Exception {
+    MiniDFSCluster cluster = null;
+    JournalHttpServer jns1 = null, jns2 = null;
+
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
+
+      // restart namenode, so it will have one finalized edit segment
+      cluster.restartNameNode();
+
+      // get namenode clusterID/layoutVersion/namespaceID
+      InetSocketAddress nnAddr = cluster.getNameNode(0).getNameNodeAddress();
+      InetSocketAddress serverAddr = new InetSocketAddress(0);
+      JournalListener listener = Mockito.mock(JournalListener.class);
+      JournalService service = new JournalService(conf, nnAddr, serverAddr,
+          listener);
+      service.start();
+      StorageInfo si = service.getJournal().getStorage();
+      service.stop();
+        
+      // start jns1 with path1
+      conf.set(DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY, path1.getPath());
+      Journal journal1 = new Journal(conf);
+      journal1.format(si.namespaceID, si.clusterID);   
+      jns1 = new JournalHttpServer(conf, journal1,
+          NetUtils.createSocketAddr("localhost:50200"));
+      jns1.start();
+
+      InetSocketAddress srcRpcAddr = NameNode.getServiceAddress(conf, true);
+      RemoteEditLogManifest manifest = editsToDownload(srcRpcAddr, 1);
+
+      String addr = conf.get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
+      jns1.downloadEditFiles(addr, manifest);
+
+      // start jns2 with path2
+      conf.set(DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY, path2.getPath());
+      Journal journal2 = new Journal(conf);
+      journal2.format(si.namespaceID, si.clusterID);   
+      jns2 = new JournalHttpServer(conf, journal2,
+          NetUtils.createSocketAddr("localhost:50300"));
+      jns2.start();
+
+      jns2.downloadEditFiles("localhost:50200", manifest);
+
+    } catch (IOException e) {
+      LOG.error("Error in TestCopyEdits:", e);
+      assertTrue(e.getLocalizedMessage(), false);
+    } finally {
+      if (jns1 != null)
+        jns1.stop();
+      if (jns2 != null)
+        jns2.stop();
+      if (cluster != null)
+        cluster.shutdown();
+    }
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/journalservice/TestJournalService.java

@@ -98,11 +98,11 @@ public class TestJournalService {
     }
 
     //test restart journal service
-    StorageInfo before = service.getJournal().getStorageInfo();
+    StorageInfo before = service.getJournal().getStorage();
     LOG.info("before: " + before);
     service.stop();
     service = newJournalService(nnAddress, listener, conf);
-    StorageInfo after = service.getJournal().getStorageInfo();
+    StorageInfo after = service.getJournal().getStorage();
     LOG.info("after : " + after);
     Assert.assertEquals(before.toString(), after.toString());
   }