HDFS-3313. Create a protocol for journal service synchronziation. Brandon Li

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

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

@@ -23,6 +23,9 @@ HDFS-3092 branch changes
     HDFS-3201. Add GetJournalEditServlet for downloading edit logs from journal
     service.  (Brandon Li via szetszwo)
 
+    HDFS-3313. Create a protocol for journal service synchronziation.  (Brandon
+    Li via szetszwo)
+
   IMPROVEMENTS
 
   OPTIMIZATIONS

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/JournalSyncProtocolPB.java

@@ -0,0 +1,41 @@
+/**
+ * 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.protocolPB;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.protocol.proto.JournalSyncProtocolProtos.JournalSyncProtocolService;
+import org.apache.hadoop.ipc.ProtocolInfo;
+import org.apache.hadoop.security.KerberosInfo;
+
+/**
+ * Protocol used to synchronize journal edit logs.
+ * 
+ * Note: This extends the protocolbuffer service based interface to
+ * add annotations required for security.
+ */
+@KerberosInfo(
+    serverPrincipal = DFSConfigKeys.DFS_JOURNAL_USER_NAME_KEY,
+    clientPrincipal = DFSConfigKeys.DFS_JOURNAL_USER_NAME_KEY)
+@ProtocolInfo(protocolName = 
+    "org.apache.hadoop.hdfs.server.protocol.JournalSyncProtocol",
+    protocolVersion = 1)
+@InterfaceAudience.Private
+public interface JournalSyncProtocolPB extends
+    JournalSyncProtocolService.BlockingInterface {
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/JournalSyncProtocolServerSideTranslatorPB.java

@@ -0,0 +1,60 @@
+/**
+ * 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.protocolPB;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.protocol.proto.JournalSyncProtocolProtos.GetEditLogManifestRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.JournalSyncProtocolProtos.GetEditLogManifestResponseProto;
+import org.apache.hadoop.hdfs.server.protocol.JournalInfo;
+import org.apache.hadoop.hdfs.server.protocol.JournalSyncProtocol;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+
+/**
+ * Implementation for protobuf service that forwards requests
+ * received on {@link JournalSyncProtocolPB} to the 
+ * {@link JournalSyncProtocol} server implementation.
+ */
+@InterfaceAudience.Private
+public class JournalSyncProtocolServerSideTranslatorPB implements JournalSyncProtocolPB {
+  /** Server side implementation to delegate the requests to */
+  private final JournalSyncProtocol impl;
+
+  public JournalSyncProtocolServerSideTranslatorPB(JournalSyncProtocol impl) {
+    this.impl = impl;
+  }
+
+  @Override
+  public GetEditLogManifestResponseProto getEditLogManifest(
+      RpcController unused, GetEditLogManifestRequestProto request)
+      throws ServiceException {
+    RemoteEditLogManifest manifest;
+    try {
+      JournalInfo journalInfo = PBHelper.convert(request.getJournalInfo());
+      manifest = impl.getEditLogManifest(journalInfo, request.getSinceTxId());
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+    return GetEditLogManifestResponseProto.newBuilder()
+        .setManifest(PBHelper.convert(manifest)).build();
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/JournalSyncProtocolTranslatorPB.java

@@ -0,0 +1,79 @@
+/**
+ * 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.protocolPB;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.protocol.proto.JournalProtocolProtos.JournalInfoProto;
+import org.apache.hadoop.hdfs.protocol.proto.JournalSyncProtocolProtos.GetEditLogManifestRequestProto;
+import org.apache.hadoop.hdfs.server.protocol.JournalInfo;
+import org.apache.hadoop.hdfs.server.protocol.JournalSyncProtocol;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
+import org.apache.hadoop.ipc.ProtobufHelper;
+import org.apache.hadoop.ipc.ProtocolMetaInterface;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.RpcClientUtil;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+
+/**
+ * This class is the client side translator to translate the requests made on
+ * {@link JournalSyncProtocol} interfaces to the RPC server implementing
+ * {@link JournalSyncProtocolPB}.
+ */
+@InterfaceAudience.Private
+public class JournalSyncProtocolTranslatorPB implements ProtocolMetaInterface,
+    JournalSyncProtocol, Closeable {
+  /** RpcController is not used and hence is set to null */
+  private final static RpcController NULL_CONTROLLER = null;
+  private final JournalSyncProtocolPB rpcProxy;
+  
+  public JournalSyncProtocolTranslatorPB(JournalSyncProtocolPB rpcProxy) {
+    this.rpcProxy = rpcProxy;
+  }
+
+  @Override
+  public void close() {
+    RPC.stopProxy(rpcProxy);
+  }
+  
+  @Override
+  public RemoteEditLogManifest getEditLogManifest(JournalInfo journalInfo, long sinceTxId)
+      throws IOException {
+    JournalInfoProto journalInfoProto = PBHelper.convert(journalInfo);
+    GetEditLogManifestRequestProto req = GetEditLogManifestRequestProto
+        .newBuilder().setJournalInfo(journalInfoProto).setSinceTxId(sinceTxId).build();
+    
+    try {
+      return PBHelper.convert(rpcProxy.getEditLogManifest(NULL_CONTROLLER, req)
+          .getManifest());
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
+  @Override
+  public boolean isMethodSupported(String methodName) throws IOException {
+    return RpcClientUtil.isMethodSupported(rpcProxy, JournalSyncProtocolPB.class,
+        RPC.RpcKind.RPC_PROTOCOL_BUFFER,
+        RPC.getProtocolVersion(JournalSyncProtocolPB.class), methodName);
+  }
+}

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

@@ -38,6 +38,7 @@ import org.apache.hadoop.hdfs.server.namenode.FSEditLog;
 import org.apache.hadoop.hdfs.server.namenode.FSImage;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage;
 import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 
 /** The journal stored in local directories. */
 class Journal {
@@ -115,7 +116,11 @@ class Journal {
   FSEditLog getEditLog() {
     return image.getEditLog();
   }
-
+  
+  RemoteEditLogManifest getRemoteEditLogs(long sinceTxId) throws IOException {
+    return image.getEditLog().getEditLogManifest(sinceTxId);
+  }
+  
   static List<URI> getEditDirs(Configuration conf) throws IOException {
     final Collection<String> dirs = conf.getTrimmedStringCollection(
         DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY);

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

@@ -23,19 +23,26 @@ import java.net.InetSocketAddress;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.NameNodeProxies;
 import org.apache.hadoop.hdfs.protocol.UnregisteredNodeException;
 import org.apache.hadoop.hdfs.protocol.proto.JournalProtocolProtos.JournalProtocolService;
+import org.apache.hadoop.hdfs.protocol.proto.JournalSyncProtocolProtos.JournalSyncProtocolService;
 import org.apache.hadoop.hdfs.protocolPB.JournalProtocolPB;
 import org.apache.hadoop.hdfs.protocolPB.JournalProtocolServerSideTranslatorPB;
+import org.apache.hadoop.hdfs.protocolPB.JournalSyncProtocolPB;
+import org.apache.hadoop.hdfs.protocolPB.JournalSyncProtocolServerSideTranslatorPB;
+import org.apache.hadoop.hdfs.protocolPB.JournalSyncProtocolTranslatorPB;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
 import org.apache.hadoop.hdfs.server.protocol.FenceResponse;
 import org.apache.hadoop.hdfs.server.protocol.FencedException;
 import org.apache.hadoop.hdfs.server.protocol.JournalInfo;
-import org.apache.hadoop.hdfs.server.protocol.JournalProtocol;
+import org.apache.hadoop.hdfs.server.protocol.JournalServiceProtocols;
+import org.apache.hadoop.hdfs.server.protocol.JournalSyncProtocol;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeRegistration;
 import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 import org.apache.hadoop.ipc.ProtobufRpcEngine;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.net.NetUtils;
@@ -59,7 +66,7 @@ import com.google.protobuf.BlockingService;
  * listener implementation can handle the callbacks based on the application
  * requirement.
  */
-public class JournalService implements JournalProtocol {
+public class JournalService implements JournalServiceProtocols {
   public static final Log LOG = LogFactory.getLog(JournalService.class.getName());
 
   private final InetSocketAddress nnAddress;
@@ -292,17 +299,43 @@ public class JournalService implements JournalProtocol {
     return new FenceResponse(previousEpoch, 0, false);
   }
 
+  @Override
+  public RemoteEditLogManifest getEditLogManifest(JournalInfo journalInfo,
+      long sinceTxId) throws IOException {
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Received getEditLogManifest " + sinceTxId);
+    }
+    if (!journal.isFormatted()) {
+      throw new IOException("This journal service is not formatted.");
+    }
+    verify(journalInfo.getNamespaceId(), journalInfo.getClusterId());
+
+    //journal has only one storage directory
+    return journal.getRemoteEditLogs(sinceTxId);
+  }
+
   /** Create an RPC server. */
   private static RPC.Server createRpcServer(Configuration conf,
-      InetSocketAddress address, JournalProtocol impl) throws IOException {
+      InetSocketAddress address, JournalServiceProtocols impl) throws IOException {
+    
     RPC.setProtocolEngine(conf, JournalProtocolPB.class,
         ProtobufRpcEngine.class);
     JournalProtocolServerSideTranslatorPB xlator = 
         new JournalProtocolServerSideTranslatorPB(impl);
     BlockingService service = 
         JournalProtocolService.newReflectiveBlockingService(xlator);
-    return RPC.getServer(JournalProtocolPB.class, service,
+    
+    JournalSyncProtocolServerSideTranslatorPB syncXlator = 
+        new JournalSyncProtocolServerSideTranslatorPB(impl);   
+    BlockingService syncService = 
+        JournalSyncProtocolService.newReflectiveBlockingService(syncXlator);
+    
+    RPC.Server rpcServer = RPC.getServer(JournalProtocolPB.class, service,
         address.getHostName(), address.getPort(), 1, false, conf, null);
+    DFSUtil.addPBProtocol(conf, JournalSyncProtocolPB.class, syncService,
+        rpcServer);
+    
+    return rpcServer;
   }
   
   private void verifyEpoch(long e) throws FencedException {
@@ -390,4 +423,16 @@ public class JournalService implements JournalProtocol {
   long getEpoch() {
     return epoch;
   }
+  
+  public static JournalSyncProtocol createProxyWithJournalSyncProtocol(
+      InetSocketAddress address, Configuration conf, UserGroupInformation ugi)
+      throws IOException {
+    RPC.setProtocolEngine(conf, JournalSyncProtocolPB.class,
+        ProtobufRpcEngine.class);
+    Object proxy = RPC.getProxy(JournalSyncProtocolPB.class,
+        RPC.getProtocolVersion(JournalSyncProtocolPB.class), address, ugi,
+        conf, NetUtils.getDefaultSocketFactory(conf), 30000);
+
+    return new JournalSyncProtocolTranslatorPB((JournalSyncProtocolPB) proxy);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/JournalServiceProtocols.java

@@ -0,0 +1,27 @@
+/**
+ * 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.protocol;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/** The full set of RPC methods implemented by the Journal Service.  */
+@InterfaceAudience.Private
+public interface JournalServiceProtocols
+  extends JournalProtocol, JournalSyncProtocol {
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/JournalSyncProtocol.java

@@ -0,0 +1,58 @@
+/**
+ * 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.protocol;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.security.KerberosInfo;
+
+/**
+ * Protocol used to sync journal edits. 
+ * 
+ */
+@KerberosInfo(
+    serverPrincipal = DFSConfigKeys.DFS_JOURNAL_USER_NAME_KEY,
+    clientPrincipal = DFSConfigKeys.DFS_JOURNAL_USER_NAME_KEY)
+@InterfaceAudience.Private
+public interface JournalSyncProtocol {
+  /**
+   * 
+   * This class is used by both the Namenode and Journal Service to insulate
+   * from the protocol serialization.
+   * 
+   * If you are adding/changing journal sync protocol then you need to change both this
+   * class and ALSO related protocol buffer wire protocol definition in
+   * JournalSyncProtocol.proto.
+   * 
+   * For more details on protocol buffer wire protocol, please see
+   * .../org/apache/hadoop/hdfs/protocolPB/overview.html
+   */
+
+  /**
+   * Return a structure containing details about all edit logs available to be
+   * fetched from the JournalNode.
+   * 
+   * @param journalInfo journal information 
+   * @param sinceTxId return only logs containing transactions >= sinceTxI
+   * @throws IOException
+   */
+  public RemoteEditLogManifest getEditLogManifest(JournalInfo info, long sinceTxId)
+      throws IOException;
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/proto/JournalSyncProtocol.proto

@@ -0,0 +1,57 @@
+/**
+ * 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.
+ */
+
+// This file contains protocol buffers that are used throughout HDFS -- i.e.
+// by the client, server, and data transfer protocols.
+
+option java_package = "org.apache.hadoop.hdfs.protocol.proto";
+option java_outer_classname = "JournalSyncProtocolProtos";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+
+import "hdfs.proto";  // for RemoteEditLogManifestProto
+import "JournalProtocol.proto"; // for JournalInfoProto
+
+/**
+ * sinceTxId - return the editlog information for transactions >= sinceTxId
+ */
+message GetEditLogManifestRequestProto {
+  required JournalInfoProto journalInfo = 1; // Info about the journal
+  required uint64 sinceTxId = 2;  // Transaction ID
+}
+
+/**
+ * manifest - Enumeration of editlogs from journal service 
+ *            for logs >= sinceTxId in the request
+ */
+message GetEditLogManifestResponseProto {
+  required RemoteEditLogManifestProto manifest = 1; 
+}
+
+/**
+ * Protocol used to sync journal edits to a remote node.
+ *
+ * See the request and response for details of rpc call.
+ */
+service JournalSyncProtocolService {
+  /**
+   * Get editlog manifests from the active journalnode for all the editlogs
+   */
+  rpc getEditLogManifest(GetEditLogManifestRequestProto)
+      returns(GetEditLogManifestResponseProto);
+}

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

@@ -37,6 +37,8 @@ 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.NameNode;
+import org.apache.hadoop.hdfs.server.protocol.JournalInfo;
+import org.apache.hadoop.hdfs.server.protocol.JournalSyncProtocol;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 import org.apache.hadoop.net.NetUtils;
@@ -68,9 +70,9 @@ 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 = new File(hdfsDir, "j1dir");
+    path2 = new File(hdfsDir, "j2dir");
     path1.mkdir();
     path2.mkdir();
     if (!path1.exists() || !path2.exists()) {
@@ -79,7 +81,7 @@ public class TestJournalHttpServer {
     }
 
     System.out.println("configuring hdfsdir is " + hdfsDir.getAbsolutePath()
-        + "; jn1Dir = " + path1.getPath() + "; jn2Dir = " + path2.getPath());
+        + "; j1Dir = " + path1.getPath() + "; j2Dir = " + path2.getPath());
   }
 
   /**
@@ -90,15 +92,15 @@ public class TestJournalHttpServer {
   @Test
   public void testHttpServer() throws Exception {
     MiniDFSCluster cluster = null;
-    JournalHttpServer jns1 = null;
+    JournalHttpServer jh1 = null;
 
     try {
       cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
 
       conf.set(DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY, path1.getPath());
-      jns1 = new JournalHttpServer(conf, new Journal(conf),
+      jh1 = new JournalHttpServer(conf, new Journal(conf),
           NetUtils.createSocketAddr("localhost:50200"));
-      jns1.start();
+      jh1.start();
 
       String pageContents = DFSTestUtil.urlGet(new URL(
           "http://localhost:50200/journalstatus.jsp"));
@@ -108,26 +110,12 @@ public class TestJournalHttpServer {
       LOG.error("Error in TestHttpServer:", e);
       assertTrue(e.getLocalizedMessage(), false);
     } finally {
-      if (jns1 != null)
-        jns1.stop();
+      if (jh1 != null)
+        jh1.stop();
       if (cluster != null)
         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
@@ -138,58 +126,69 @@ public class TestJournalHttpServer {
   @Test
   public void testCopyEdits() throws Exception {
     MiniDFSCluster cluster = null;
-    JournalHttpServer jns1 = null, jns2 = null;
+    JournalService service = null;
+    JournalHttpServer jhs1 = null, jhs2 = null;
 
     try {
       cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
 
-      // restart namenode, so it will have one finalized edit segment
+      // restart namenode, so it will have finalized edit segments
       cluster.restartNameNode();
 
-      // get namenode clusterID/layoutVersion/namespaceID
+      conf.set(DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY, path1.getPath());
       InetSocketAddress nnAddr = cluster.getNameNode(0).getNameNodeAddress();
-      InetSocketAddress serverAddr = new InetSocketAddress(0);
+      InetSocketAddress serverAddr = new InetSocketAddress(50900);
       JournalListener listener = Mockito.mock(JournalListener.class);
-      JournalService service = new JournalService(conf, nnAddr, serverAddr,
-          listener);
+      service = new JournalService(conf, nnAddr, serverAddr, listener);
       service.start();
+      
+      // get namenode clusterID/layoutVersion/namespaceID
       StorageInfo si = service.getJournal().getStorage();
-      service.stop();
-        
+      JournalInfo journalInfo = new JournalInfo(si.layoutVersion, si.clusterID,
+          si.namespaceID);
+
       // 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,
+      jhs1 = new JournalHttpServer(conf, service.getJournal(),
           NetUtils.createSocketAddr("localhost:50200"));
-      jns1.start();
+      jhs1.start();
 
+      // get all edit segments
       InetSocketAddress srcRpcAddr = NameNode.getServiceAddress(conf, true);
-      RemoteEditLogManifest manifest = editsToDownload(srcRpcAddr, 1);
+      NamenodeProtocol namenode = NameNodeProxies.createNonHAProxy(conf,
+          srcRpcAddr, NamenodeProtocol.class,
+          UserGroupInformation.getCurrentUser(), true).getProxy();
 
-      String addr = conf.get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
-      jns1.downloadEditFiles(addr, manifest);
+      RemoteEditLogManifest manifest = namenode.getEditLogManifest(1);      
+      jhs1.downloadEditFiles(
+          conf.get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY), 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,
+      jhs2 = new JournalHttpServer(conf, journal2,
           NetUtils.createSocketAddr("localhost:50300"));
-      jns2.start();
+      jhs2.start();
 
-      jns2.downloadEditFiles("localhost:50200", manifest);
+      // transfer edit logs from j1 to j2
+      JournalSyncProtocol journalp = JournalService.createProxyWithJournalSyncProtocol(
+          NetUtils.createSocketAddr("localhost:50900"), conf,
+          UserGroupInformation.getCurrentUser());
+      RemoteEditLogManifest manifest1 = journalp.getEditLogManifest(journalInfo, 1);  
+      jhs2.downloadEditFiles("localhost:50200", manifest1);
 
     } 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 (jhs1 != null)
+        jhs1.stop();
+      if (jhs2 != null)
+        jhs2.stop();
       if (cluster != null)
         cluster.shutdown();
+      if (service != null)
+        service.stop();
     }
   }
 }