Feature/grant lease in create

test/namenode/CreationTest.cc

@@ -144,3 +144,24 @@ TEST_F(NamenodeTest, creationPerformance) {
     }
 }
 
+TEST_F(NamenodeTest, createLease) {
+    hadoop::hdfs::CreateRequestProto create_req =
+      getCreateRequestProto("basic_file_1");
+    create_req.set_clientname("test_client1");
+    hadoop::hdfs::CreateResponseProto create_resp;
+    ASSERT_EQ(client->create_file(create_req, create_resp),
+              zkclient::ZkNnClient::CreateResponse::Ok);
+
+    hadoop::hdfs::HdfsFileStatusProto file_status = create_resp.fs();
+    ASSERT_EQ(file_status.filetype(),
+              hadoop::hdfs::HdfsFileStatusProto::IS_FILE);
+    ASSERT_TRUE(client->file_exists("basic_file_1"));
+
+    hadoop::hdfs::RecoverLeaseRequestProto recover_req;
+    recover_req.set_clientname("test_client2");
+    recover_req.set_src("basic_file_1");
+    hadoop::hdfs::RecoverLeaseResponseProto recover_res;
+    client->recover_lease(recover_req, recover_res);
+    ASSERT_FALSE(recover_res.result());
+}
+

test/namenode/LeaseTest.cc

@@ -40,6 +40,7 @@ TEST_F(NamenodeTest, recoverLeaseCorrectnessTest) {
   // Create the file
   hadoop::hdfs::CreateRequestProto create_req =
     getCreateRequestProto("test_filename");
+  create_req.set_clientname("test_client");
   hadoop::hdfs::CreateResponseProto create_resp;
   ASSERT_EQ(client->create_file(create_req, create_resp),
             zkclient::ZkNnClient::CreateResponse::Ok);

zookeeper/include/zk_nn_client.h

@@ -613,7 +613,8 @@ class ZkNnClient : public ZkClientCommon {
    * Crate a znode corresponding to a file of "filetype", with path "path", with
    * znode data contained in "znode_data"
    */
-  bool create_file_znode(const std::string &path, FileZNode *znode_data);
+  bool create_file_znode(const std::string &path, FileZNode *znode_data,
+                         const std::string &client_name);
 
   /**
    * Set the default information in a directory znode struct

zookeeper/source/zk_nn_client.cc

@@ -404,7 +404,7 @@ void ZkNnClient::recover_lease(RecoverLeaseRequestProto &req,
     res.set_result(false);
     return;
   }
-  if (children.size() > 0) {
+  if (children.size() > 1) {
     for (std::string child : children) {
       LOG(ERROR) << "[recover_lease] Child: " + child;
     }
@@ -420,6 +420,10 @@ void ZkNnClient::recover_lease(RecoverLeaseRequestProto &req,
     return;
   }
   std::string lease_holder_client = children[0];
+  if (lease_holder_client == client_name) {
+    res.set_result(true);
+    return;
+  }
   if (lease_expired(lease_holder_client)) {
     // TODO(elfyingying): start the lease recovery process
     // that closes the file for the
@@ -800,7 +804,8 @@ ZkNnClient::GetFileInfoResponse ZkNnClient::get_info(
  * Create a node in zookeeper corresponding to a file
  */
 bool ZkNnClient::create_file_znode(const std::string &path,
-                                  FileZNode *znode_data) {
+                                  FileZNode *znode_data,
+                                   const std::string &client_name) {
   int error_code;
   if (!file_exists(path)) {
     LOG(INFO) << "[create_file_znode] Creating file znode at " << path;
@@ -837,6 +842,19 @@ bool ZkNnClient::create_file_znode(const std::string &path,
                    << error_code;
         return false;
       }
+      // grant the client the lease
+      if (!zk->create(LeaseZookeeperPath(path) + '/' +
+                        client_name, ZKWrapper::EMPTY_VECTOR,
+                      error_code, false)) {
+        LOG(ERROR) << "[create_file_znode] ZK create lease client path failed"
+        << error_code;
+        return false;
+      }
+      // set the timestamp for the client that created the file
+      RenewLeaseRequestProto req;
+      req.set_clientname(client_name);
+      RenewLeaseResponseProto res;
+      renew_lease(req, res);
     }
     return true;
   }
@@ -1237,7 +1255,7 @@ ZkNnClient::CreateResponse ZkNnClient::create_file(
   }
 
   // if we failed, then do not set any status
-  if (!create_file_znode(path, &znode_data))
+  if (!create_file_znode(path, &znode_data, request.clientname()))
       return CreateResponse::FailedCreateZnode;
 
   HdfsFileStatusProto *status = response.mutable_fs();
@@ -1358,6 +1376,7 @@ ZkNnClient::MkdirResponse ZkNnClient::mkdir(MkdirsRequestProto &request,
 ZkNnClient::MkdirResponse ZkNnClient::mkdir_helper(const std::string &path,
                            bool create_parent) {
   LOG(INFO) << "[mkdir_helper] mkdir_helper called with input " << path;
+    std::string empty_string;
   if (create_parent) {
     std::vector<std::string> split_path;
     boost::split(split_path, path, boost::is_any_of("/"));
@@ -1377,7 +1396,7 @@ ZkNnClient::MkdirResponse ZkNnClient::mkdir_helper(const std::string &path,
         not_exist = true;
         FileZNode znode_data;
         set_mkdir_znode(&znode_data);
-        if (!create_file_znode(p_path, &znode_data)) {
+        if (!create_file_znode(p_path, &znode_data, empty_string)) {
           // TODO(2016) unroll the created directories
           return MkdirResponse::FailedZnodeCreation;
         }
@@ -1386,7 +1405,7 @@ ZkNnClient::MkdirResponse ZkNnClient::mkdir_helper(const std::string &path,
   } else {
     FileZNode znode_data;
     set_mkdir_znode(&znode_data);
-    return create_file_znode(path, &znode_data) ?
+    return create_file_znode(path, &znode_data, empty_string) ?
          MkdirResponse::Ok : MkdirResponse::FailedZnodeCreation;
   }
   return MkdirResponse::Ok;