HDFS-1948 Forward port 'hdfs-1520 lightweight namenode operation to trigger lease reccovery'
git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/trunk@1133108 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/CHANGES.txt b/CHANGES.txt
index a71b891..cba750c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -944,6 +944,9 @@
HDFS-1619. Remove AC_TYPE* from the libhdfs. (Roman Shaposhnik via eli)
+ HDFS-1948 Forward port 'hdfs-1520 lightweight namenode operation to
+ trigger lease recovery' (stack)
+
OPTIMIZATIONS
HDFS-1140. Speedup INode.getPathComponents. (Dmytro Molkov via shv)
diff --git a/src/java/org/apache/hadoop/hdfs/DFSClient.java b/src/java/org/apache/hadoop/hdfs/DFSClient.java
index 3b03674..17a9829 100644
--- a/src/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/src/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -539,6 +539,23 @@
}
/**
+ * Recover a file's lease
+ * @param src a file's path
+ * @return true if the file is already closed
+ * @throws IOException
+ */
+ boolean recoverLease(String src) throws IOException {
+ checkOpen();
+
+ try {
+ return namenode.recoverLease(src, clientName);
+ } catch (RemoteException re) {
+ throw re.unwrapRemoteException(FileNotFoundException.class,
+ AccessControlException.class);
+ }
+ }
+
+ /**
* Get block location info about file
*
* getBlockLocations() returns a list of hostnames that store
diff --git a/src/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/src/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
index 9d54954..c4e916e 100644
--- a/src/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
+++ b/src/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
@@ -218,6 +218,17 @@
this.verifyChecksum = verifyChecksum;
}
+ /**
+ * Start the lease recovery of a file
+ *
+ * @param f a file
+ * @return true if the file is already closed
+ * @throws IOException if an error occurs
+ */
+ public boolean recoverLease(Path f) throws IOException {
+ return dfs.recoverLease(getPathName(f));
+ }
+
@SuppressWarnings("deprecation")
@Override
public FSDataInputStream open(Path f, int bufferSize) throws IOException {
diff --git a/src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java b/src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
index 9db6fa4..1520668 100644
--- a/src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
+++ b/src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
@@ -537,6 +537,17 @@
public void renewLease(String clientName) throws AccessControlException,
IOException;
+ /**
+ * Start lease recovery.
+ * Lightweight NameNode operation to trigger lease recovery
+ *
+ * @param src path of the file to start lease recovery
+ * @param clientName name of the current client
+ * @return true if the file is already closed
+ * @throws IOException
+ */
+ public boolean recoverLease(String src, String clientName) throws IOException;
+
public int GET_STATS_CAPACITY_IDX = 0;
public int GET_STATS_USED_IDX = 1;
public int GET_STATS_REMAINING_IDX = 2;
diff --git a/src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index d34fa47..d0a2afe 100644
--- a/src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -1395,62 +1395,7 @@
try {
INode myFile = dir.getFileINode(src);
- if (myFile != null && myFile.isUnderConstruction()) {
- INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction) myFile;
- //
- // If the file is under construction , then it must be in our
- // leases. Find the appropriate lease record.
- //
- Lease lease = leaseManager.getLeaseByPath(src);
- if (lease == null) {
- throw new AlreadyBeingCreatedException(
- "failed to create file " + src + " for " + holder +
- " on client " + clientMachine +
- " because pendingCreates is non-null but no leases found.");
- }
- //
- // We found the lease for this file. And surprisingly the original
- // holder is trying to recreate this file. This should never occur.
- //
- if (lease.getHolder().equals(holder)) {
- throw new AlreadyBeingCreatedException(
- "failed to create file " + src + " for " + holder +
- " on client " + clientMachine +
- " because current leaseholder is trying to recreate file.");
- }
- assert lease.getHolder().equals(pendingFile.getClientName()) :
- "Current lease holder " + lease.getHolder() +
- " does not match file creator " + pendingFile.getClientName();
- //
- // Current lease holder is different from the requester.
- // If the original holder has not renewed in the last SOFTLIMIT
- // period, then start lease recovery, otherwise fail.
- //
- if (lease.expiredSoftLimit()) {
- LOG.info("startFile: recover lease " + lease + ", src=" + src);
- boolean isClosed = internalReleaseLease(lease, src, null);
- if(!isClosed)
- throw new RecoveryInProgressException(
- "Failed to close file " + src +
- ". Lease recovery is in progress. Try again later.");
-
- } else {
- BlockInfoUnderConstruction lastBlock=pendingFile.getLastBlock();
- if(lastBlock != null && lastBlock.getBlockUCState() ==
- BlockUCState.UNDER_RECOVERY) {
- throw new RecoveryInProgressException(
- "Recovery in progress, file [" + src + "], " +
- "lease owner [" + lease.getHolder() + "]");
- } else {
- throw new AlreadyBeingCreatedException(
- "Failed to create file [" + src + "] for [" + holder +
- "] on client [" + clientMachine +
- "], because this file is already being created by [" +
- pendingFile.getClientName() + "] on [" +
- pendingFile.getClientMachine() + "]");
- }
- }
- }
+ recoverLeaseInternal(myFile, src, holder, clientMachine, false);
try {
blockManager.verifyReplication(src, replication, clientMachine);
@@ -1537,6 +1482,120 @@
}
/**
+ * Recover lease;
+ * Immediately revoke the lease of the current lease holder and start lease
+ * recovery so that the file can be forced to be closed.
+ *
+ * @param src the path of the file to start lease recovery
+ * @param holder the lease holder's name
+ * @param clientMachine the client machine's name
+ * @return true if the file is already closed
+ * @throws IOException
+ */
+ synchronized boolean recoverLease(String src, String holder, String clientMachine)
+ throws IOException {
+ if (isInSafeMode()) {
+ throw new SafeModeException(
+ "Cannot recover the lease of " + src, safeMode);
+ }
+ if (!DFSUtil.isValidName(src)) {
+ throw new IOException("Invalid file name: " + src);
+ }
+
+ INode inode = dir.getFileINode(src);
+ if (inode == null) {
+ throw new FileNotFoundException("File not found " + src);
+ }
+
+ if (!inode.isUnderConstruction()) {
+ return true;
+ }
+ if (isPermissionEnabled) {
+ checkPathAccess(src, FsAction.WRITE);
+ }
+
+ recoverLeaseInternal(inode, src, holder, clientMachine, true);
+ return false;
+ }
+
+ private void recoverLeaseInternal(INode fileInode,
+ String src, String holder, String clientMachine, boolean force)
+ throws IOException {
+ if (fileInode != null && fileInode.isUnderConstruction()) {
+ INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction) fileInode;
+ //
+ // If the file is under construction , then it must be in our
+ // leases. Find the appropriate lease record.
+ //
+ Lease lease = leaseManager.getLease(holder);
+ //
+ // We found the lease for this file. And surprisingly the original
+ // holder is trying to recreate this file. This should never occur.
+ //
+ if (!force && lease != null) {
+ Lease leaseFile = leaseManager.getLeaseByPath(src);
+ if ((leaseFile != null && leaseFile.equals(lease)) ||
+ lease.getHolder().equals(holder)) {
+ throw new AlreadyBeingCreatedException(
+ "failed to create file " + src + " for " + holder +
+ " on client " + clientMachine +
+ " because current leaseholder is trying to recreate file.");
+ }
+ }
+ //
+ // Find the original holder.
+ //
+ lease = leaseManager.getLease(pendingFile.getClientName());
+ if (lease == null) {
+ throw new AlreadyBeingCreatedException(
+ "failed to create file " + src + " for " + holder +
+ " on client " + clientMachine +
+ " because pendingCreates is non-null but no leases found.");
+ }
+ if (force) {
+ // close now: no need to wait for soft lease expiration and
+ // close only the file src
+ LOG.info("recoverLease: recover lease " + lease + ", src=" + src +
+ " from client " + pendingFile.getClientName());
+ internalReleaseLease(lease, src, holder);
+ } else {
+ assert lease.getHolder().equals(pendingFile.getClientName()) :
+ "Current lease holder " + lease.getHolder() +
+ " does not match file creator " + pendingFile.getClientName();
+ //
+ // If the original holder has not renewed in the last SOFTLIMIT
+ // period, then start lease recovery.
+ //
+ if (lease.expiredSoftLimit()) {
+ LOG.info("startFile: recover lease " + lease + ", src=" + src +
+ " from client " + pendingFile.getClientName());
+ boolean isClosed = internalReleaseLease(lease, src, null);
+ if(!isClosed)
+ throw new RecoveryInProgressException(
+ "Failed to close file " + src +
+ ". Lease recovery is in progress. Try again later.");
+ } else {
+ BlockInfoUnderConstruction lastBlock=pendingFile.getLastBlock();
+ if(lastBlock != null && lastBlock.getBlockUCState() ==
+ BlockUCState.UNDER_RECOVERY) {
+ throw new RecoveryInProgressException(
+ "Recovery in progress, file [" + src + "], " +
+ "lease owner [" + lease.getHolder() + "]");
+ } else {
+ throw new AlreadyBeingCreatedException(
+ "Failed to create file [" + src + "] for [" + holder +
+ "] on client [" + clientMachine +
+ "], because this file is already being created by [" +
+ pendingFile.getClientName() + "] on [" +
+ pendingFile.getClientMachine() + "]");
+ }
+ }
+ }
+ }
+
+ }
+
+ /**
* Append to an existing file in the namespace.
*/
LocatedBlock appendFile(String src, String holder, String clientMachine)
diff --git a/src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java b/src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
index 108ac6e..8a62810 100644
--- a/src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
+++ b/src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
@@ -836,6 +836,12 @@
}
/** {@inheritDoc} */
+ public boolean recoverLease(String src, String clientName) throws IOException {
+ String clientMachine = getClientMachine();
+ return namesystem.recoverLease(src, clientName, clientMachine);
+ }
+
+ /** {@inheritDoc} */
public boolean setReplication(String src, short replication)
throws IOException {
return namesystem.setReplication(src, replication);
diff --git a/src/test/hdfs/org/apache/hadoop/hdfs/TestLeaseRecovery2.java b/src/test/hdfs/org/apache/hadoop/hdfs/TestLeaseRecovery2.java
index 532023c..0902b49 100644
--- a/src/test/hdfs/org/apache/hadoop/hdfs/TestLeaseRecovery2.java
+++ b/src/test/hdfs/org/apache/hadoop/hdfs/TestLeaseRecovery2.java
@@ -17,7 +17,9 @@
*/
package org.apache.hadoop.hdfs;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.HashMap;
@@ -27,6 +29,7 @@
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileSystem;
@@ -95,7 +98,147 @@
IOUtils.closeStream(dfs);
if (cluster != null) {cluster.shutdown();}
}
-
+
+ /**
+ * Test the NameNode's revoke lease on current lease holder function.
+ * @throws Exception
+ */
+ @Test
+ public void testImmediateRecoveryOfLease() throws Exception {
+ //create a file
+ // write bytes into the file.
+ byte [] actual = new byte[FILE_SIZE];
+ int size = AppendTestUtil.nextInt(FILE_SIZE);
+ Path filepath = createFile("/immediateRecoverLease-shortlease", size, true);
+ // set the soft limit to be 1 second so that the
+ // namenode triggers lease recovery on next attempt to write-for-open.
+ cluster.setLeasePeriod(SHORT_LEASE_PERIOD, LONG_LEASE_PERIOD);
+
+ recoverLeaseUsingCreate(filepath);
+ verifyFile(dfs, filepath, actual, size);
+
+ //test recoverLease
+ // set the soft limit to be 1 hour but recoverLease should
+ // close the file immediately
+ cluster.setLeasePeriod(LONG_LEASE_PERIOD, LONG_LEASE_PERIOD);
+ size = AppendTestUtil.nextInt(FILE_SIZE);
+ filepath = createFile("/immediateRecoverLease-longlease", size, false);
+
+ // test recoverLese from a different client
+ recoverLease(filepath, null);
+ verifyFile(dfs, filepath, actual, size);
+
+ // test recoverlease from the same client
+ size = AppendTestUtil.nextInt(FILE_SIZE);
+ filepath = createFile("/immediateRecoverLease-sameclient", size, false);
+
+ // create another file using the same client
+ Path filepath1 = new Path(filepath.toString() + AppendTestUtil.nextInt());
+ FSDataOutputStream stm = dfs.create(filepath1, true, BUF_SIZE,
+ REPLICATION_NUM, BLOCK_SIZE);
+
+ // recover the first file
+ recoverLease(filepath, dfs);
+ verifyFile(dfs, filepath, actual, size);
+
+ // continue to write to the second file
+ stm.write(buffer, 0, size);
+ stm.close();
+ verifyFile(dfs, filepath1, actual, size);
+ }
+
+ private Path createFile(final String filestr, final int size,
+ final boolean triggerLeaseRenewerInterrupt)
+ throws IOException, InterruptedException {
+ AppendTestUtil.LOG.info("filestr=" + filestr);
+ Path filepath = new Path(filestr);
+ FSDataOutputStream stm = dfs.create(filepath, true, BUF_SIZE,
+ REPLICATION_NUM, BLOCK_SIZE);
+ assertTrue(dfs.dfs.exists(filestr));
+
+ AppendTestUtil.LOG.info("size=" + size);
+ stm.write(buffer, 0, size);
+
+ // hflush file
+ AppendTestUtil.LOG.info("hflush");
+ stm.hflush();
+
+ if (triggerLeaseRenewerInterrupt) {
+ AppendTestUtil.LOG.info("leasechecker.interruptAndJoin()");
+ dfs.dfs.leaserenewer.interruptAndJoin();
+ }
+ return filepath;
+ }
+
+ private void recoverLease(Path filepath, DistributedFileSystem dfs)
+ throws Exception {
+ if (dfs == null) {
+ dfs = (DistributedFileSystem)getFSAsAnotherUser(conf);
+ }
+
+ while (!dfs.recoverLease(filepath)) {
+ AppendTestUtil.LOG.info("sleep " + 5000 + "ms");
+ Thread.sleep(5000);
+ }
+ }
+
+ private FileSystem getFSAsAnotherUser(final Configuration c)
+ throws IOException, InterruptedException {
+ return FileSystem.get(FileSystem.getDefaultUri(c), c,
+ UserGroupInformation.createUserForTesting(fakeUsername,
+ new String [] {fakeGroup}).getUserName());
+ }
+
+ private void recoverLeaseUsingCreate(Path filepath)
+ throws IOException, InterruptedException {
+ FileSystem dfs2 = getFSAsAnotherUser(conf);
+
+ boolean done = false;
+ for(int i = 0; i < 10 && !done; i++) {
+ AppendTestUtil.LOG.info("i=" + i);
+ try {
+ dfs2.create(filepath, false, BUF_SIZE, (short)1, BLOCK_SIZE);
+ fail("Creation of an existing file should never succeed.");
+ } catch (IOException ioe) {
+ final String message = ioe.getMessage();
+ if (message.contains("file exists")) {
+ AppendTestUtil.LOG.info("done", ioe);
+ done = true;
+ }
+ else if (message.contains(AlreadyBeingCreatedException.class.getSimpleName())) {
+ AppendTestUtil.LOG.info("GOOD! got " + message);
+ }
+ else {
+ AppendTestUtil.LOG.warn("UNEXPECTED IOException", ioe);
+ }
+ }
+
+ if (!done) {
+ AppendTestUtil.LOG.info("sleep " + 5000 + "ms");
+ try {Thread.sleep(5000);} catch (InterruptedException e) {}
+ }
+ }
+ assertTrue(done);
+ }
+
+ private void verifyFile(FileSystem dfs, Path filepath, byte[] actual,
+ int size) throws IOException {
+ AppendTestUtil.LOG.info("Lease for file " + filepath + " is recovered. "
+ + "Validating its contents now...");
+
+ // verify that file-size matches
+ assertTrue("File should be " + size + " bytes, but is actually " +
+ " found to be " + dfs.getFileStatus(filepath).getLen() +
+ " bytes",
+ dfs.getFileStatus(filepath).getLen() == size);
+
+ // verify that there is enough data to read.
+ System.out.println("File size is good. Now validating sizes from datanodes...");
+ FSDataInputStream stmin = dfs.open(filepath);
+ stmin.readFully(0, actual, 0, size);
+ stmin.close();
+ }
+
/**
* This test makes the client does not renew its lease and also
* set the hard lease expiration period to be short 1s. Thus triggering
