diff --git a/CHANGES.txt b/CHANGES.txt
index 01837a6925b..7960f373e98 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -355,6 +355,9 @@ Trunk (unreleased changes)
     HADOOP-6719. Insert all missing methods in FilterFs.
     (Rodrigo Schmidt via dhruba)
 
+    HADOOP-6692. Add FileContext#listStatus that returns an iterator.
+    (hairong)
+
 Release 0.21.0 - Unreleased
 
   INCOMPATIBLE CHANGES
diff --git a/src/java/org/apache/hadoop/fs/AbstractFileSystem.java b/src/java/org/apache/hadoop/fs/AbstractFileSystem.java
index 5f66f608154..f1d6530e93f 100644
--- a/src/java/org/apache/hadoop/fs/AbstractFileSystem.java
+++ b/src/java/org/apache/hadoop/fs/AbstractFileSystem.java
@@ -25,7 +25,9 @@ import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.EnumSet;
 import java.util.IdentityHashMap;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.StringTokenizer;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -611,8 +613,8 @@ public abstract class AbstractFileSystem {
       }
       // Delete the destination that is a file or an empty directory
       if (dstStatus.isDir()) {
-        FileStatus[] list = listStatus(dst);
-        if (list != null && list.length != 0) {
+        Iterator<FileStatus> list = listStatusIterator(dst);
+        if (list != null && list.hasNext()) {
           throw new IOException(
               "rename cannot overwrite non empty destination directory " + dst);
         }
@@ -753,6 +755,38 @@ public abstract class AbstractFileSystem {
    * {@link FileContext#listStatus(Path)} except that Path f must be for this
    * file system.
    */
+  protected Iterator<FileStatus> listStatusIterator(final Path f)
+      throws AccessControlException, FileNotFoundException,
+      UnresolvedLinkException, IOException {
+    return new Iterator<FileStatus>() {
+      private int i = 0;
+      private FileStatus[] statusList = listStatus(f);
+      
+      @Override
+      public boolean hasNext() {
+        return i < statusList.length;
+      }
+      
+      @Override
+      public FileStatus next() {
+        if (!hasNext()) {
+          throw new NoSuchElementException();
+        }
+        return statusList[i++];
+      }
+      
+      @Override
+      public void remove() {
+        throw new UnsupportedOperationException("Remove is not supported");
+      }
+    };
+  }
+
+  /**
+   * The specification of this method matches that of
+   * {@link FileContext.Util#listStatus(Path)} except that Path f must be 
+   * for this file system.
+   */
   protected abstract FileStatus[] listStatus(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
diff --git a/src/java/org/apache/hadoop/fs/FileContext.java b/src/java/org/apache/hadoop/fs/FileContext.java
index fde725f7208..d358ee7e8fb 100644
--- a/src/java/org/apache/hadoop/fs/FileContext.java
+++ b/src/java/org/apache/hadoop/fs/FileContext.java
@@ -26,6 +26,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.IdentityHashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -1378,12 +1379,11 @@ public final class FileContext {
    * 
    * @param f is the path
    *
-   * @return the statuses of the files/directories in the given path
+   * @return an iterator that traverses statuses of the files/directories 
+   *         in the given path
    *
    * @throws AccessControlException If access is denied
    * @throws FileNotFoundException If <code>f</code> does not exist
-   * @throws UnresolvedLinkException If symbolic link <code>f</code> could not
-   *           be resolved
    * @throws UnsupportedFileSystemException If file system for <code>f</code> is
    *           not supported
    * @throws IOException If an I/O error occurred
@@ -1394,14 +1394,14 @@ public final class FileContext {
    * @throws UnexpectedServerException If server implementation throws 
    *           undeclared exception to RPC server
    */
-  public FileStatus[] listStatus(final Path f) throws AccessControlException,
-      FileNotFoundException, UnsupportedFileSystemException,
-      UnresolvedLinkException, IOException {
+  public Iterator<FileStatus> listStatus(final Path f) throws
+      AccessControlException, FileNotFoundException,
+      UnsupportedFileSystemException, IOException {
     final Path absF = fixRelativePart(f);
-    return new FSLinkResolver<FileStatus[]>() {
-      public FileStatus[] next(final AbstractFileSystem fs, final Path p) 
+    return new FSLinkResolver<Iterator<FileStatus>>() {
+      public Iterator<FileStatus> next(final AbstractFileSystem fs, final Path p) 
         throws IOException, UnresolvedLinkException {
-        return fs.listStatus(p);
+        return fs.listStatusIterator(p);
       }
     }.resolve(this, absF);
   }
@@ -1521,7 +1521,9 @@ public final class FileContext {
       }
       // f is a directory
       long[] summary = {0, 0, 1};
-      for(FileStatus s : FileContext.this.listStatus(f)) {
+      Iterator<FileStatus> statusIterator = FileContext.this.listStatus(f);
+      while(statusIterator.hasNext()) {
+        FileStatus s = statusIterator.next();
         ContentSummary c = s.isDir() ? getContentSummary(s.getPath()) :
                                        new ContentSummary(s.getLen(), 1, 0);
         summary[0] += c.getLength();
@@ -1606,7 +1608,7 @@ public final class FileContext {
     private void listStatus(ArrayList<FileStatus> results, Path f,
         PathFilter filter) throws AccessControlException,
         FileNotFoundException, IOException {
-      FileStatus[] listing = FileContext.this.listStatus(f);
+      FileStatus[] listing = listStatus(f);
       if (listing != null) {
         for (int i = 0; i < listing.length; i++) {
           if (filter.accept(listing[i].getPath())) {
@@ -1616,6 +1618,39 @@ public final class FileContext {
       }
     }
 
+    /**
+     * List the statuses of the files/directories in the given path 
+     * if the path is a directory.
+     * 
+     * @param f is the path
+     *
+     * @return an array that contains statuses of the files/directories 
+     *         in the given path
+     *
+     * @throws AccessControlException If access is denied
+     * @throws FileNotFoundException If <code>f</code> does not exist
+     * @throws UnsupportedFileSystemException If file system for <code>f</code> is
+     *           not supported
+     * @throws IOException If an I/O error occurred
+     * 
+     * Exceptions applicable to file systems accessed over RPC:
+     * @throws RpcClientException If an exception occurred in the RPC client
+     * @throws RpcServerException If an exception occurred in the RPC server
+     * @throws UnexpectedServerException If server implementation throws 
+     *           undeclared exception to RPC server
+     */
+    public FileStatus[] listStatus(final Path f) throws AccessControlException,
+        FileNotFoundException, UnsupportedFileSystemException,
+        IOException {
+      final Path absF = fixRelativePart(f);
+      return new FSLinkResolver<FileStatus[]>() {
+        public FileStatus[] next(final AbstractFileSystem fs, final Path p) 
+          throws IOException, UnresolvedLinkException {
+          return fs.listStatus(p);
+        }
+      }.resolve(FileContext.this, absF);
+    }
+
     /**
      * <p>Return all the files that match filePattern and are not checksum
      * files. Results are sorted by their names.
@@ -1911,7 +1946,7 @@ public final class FileContext {
       if (isDirectory(qSrc)) {
         checkDependencies(qSrc, qDst);
         mkdir(qDst, FsPermission.getDefault(), true);
-        FileStatus[] contents = FileContext.this.listStatus(qSrc);
+        FileStatus[] contents = listStatus(qSrc);
         for (FileStatus content : contents) {
           copy(content.getPath(), new Path(qDst, content.getPath()),
                deleteSource, overwrite);
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java b/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java
index 5be803a6699..0a6ab8140f6 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java
@@ -21,6 +21,7 @@ package org.apache.hadoop.fs;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.EnumSet;
+import java.util.Iterator;
 
 import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.fs.Options.Rename;
@@ -262,11 +263,12 @@ public abstract class FileContextMainOperationsBaseTest  {
       fc.mkdir(path, FsPermission.getDefault(), true);
     }
 
-    FileStatus[] paths = fc.listStatus(getTestRootPath(fc, "test"));
+    // test listStatus that returns an array
+    FileStatus[] paths = fc.util().listStatus(getTestRootPath(fc, "test"));
     Assert.assertEquals(1, paths.length);
     Assert.assertEquals(getTestRootPath(fc, "test/hadoop"), paths[0].getPath());
 
-    paths = fc.listStatus(getTestRootPath(fc, "test/hadoop"));
+    paths = fc.util().listStatus(getTestRootPath(fc, "test/hadoop"));
     Assert.assertEquals(3, paths.length);
 
     Assert.assertTrue(containsPath(getTestRootPath(fc, "test/hadoop/a"),
@@ -276,8 +278,34 @@ public abstract class FileContextMainOperationsBaseTest  {
     Assert.assertTrue(containsPath(getTestRootPath(fc, "test/hadoop/c"),
         paths));
 
-    paths = fc.listStatus(getTestRootPath(fc, "test/hadoop/a"));
+    paths = fc.util().listStatus(getTestRootPath(fc, "test/hadoop/a"));
     Assert.assertEquals(0, paths.length);
+    
+    // test listStatus that returns an iterator
+    Iterator<FileStatus> pathsIterator = 
+      fc.listStatus(getTestRootPath(fc, "test"));
+    Assert.assertEquals(getTestRootPath(fc, "test/hadoop"), 
+        pathsIterator.next().getPath());
+    Assert.assertFalse(pathsIterator.hasNext());
+
+    pathsIterator = fc.listStatus(getTestRootPath(fc, "test/hadoop"));
+    FileStatus[] subdirs = new FileStatus[3];
+    int i=0;
+    while(i<3 && pathsIterator.hasNext()) {
+      subdirs[i++] = pathsIterator.next();
+    }
+    Assert.assertFalse(pathsIterator.hasNext());
+    Assert.assertTrue(i==3);
+    
+    Assert.assertTrue(containsPath(getTestRootPath(fc, "test/hadoop/a"),
+        subdirs));
+    Assert.assertTrue(containsPath(getTestRootPath(fc, "test/hadoop/b"),
+        subdirs));
+    Assert.assertTrue(containsPath(getTestRootPath(fc, "test/hadoop/c"),
+        subdirs));
+
+    pathsIterator = fc.listStatus(getTestRootPath(fc, "test/hadoop/a"));
+    Assert.assertFalse(pathsIterator.hasNext());
   }
   
   @Test
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java b/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java
index ded148e0c0a..226bb4aa04d 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.fs;
 
 import java.io.*;
 import java.net.URI;
+import java.util.Iterator;
 import java.util.Random;
 import java.util.EnumSet;
 import org.apache.hadoop.fs.FileContext;
@@ -606,8 +607,15 @@ public abstract class FileContextSymlinkBaseTest {
     fc.createSymlink(new Path(testBaseDir1()), link, false);
     // The size of the result is file system dependent, Hdfs is 2 (file 
     // and link) and LocalFs is 3 (file, link, file crc).
-    assertTrue(fc.listStatus(link).length == 2 ||
-               fc.listStatus(link).length == 3);
+    FileStatus[] stats = fc.util().listStatus(link);
+    assertTrue(stats.length == 2 || stats.length == 3);
+    Iterator<FileStatus> statsItor = fc.listStatus(link);
+    int dirLen = 0;
+    while(statsItor.hasNext()) {
+      statsItor.next();
+      dirLen++;
+    }
+    assertTrue(dirLen == 2 || dirLen == 3);
   }
   
   @Test
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextURIBase.java b/src/test/core/org/apache/hadoop/fs/FileContextURIBase.java
index 44a54ca7ef5..5df82459794 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextURIBase.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextURIBase.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.fs;
 
 import java.io.*;
 import java.util.ArrayList;
+import java.util.Iterator;
 
 import junit.framework.Assert;
 
@@ -501,11 +502,12 @@ public abstract class FileContextURIBase {
      fc1.mkdir(path, FsPermission.getDefault(), true);
     }
 
-    FileStatus[] paths = fc1.listStatus(qualifiedPath("test", fc1));
+    // test listStatus that returns an array of FileStatus
+    FileStatus[] paths = fc1.util().listStatus(qualifiedPath("test", fc1));
     Assert.assertEquals(1, paths.length);
     Assert.assertEquals(qualifiedPath(hPrefix, fc1), paths[0].getPath());
 
-    paths = fc1.listStatus(qualifiedPath(hPrefix, fc1));
+    paths = fc1.util().listStatus(qualifiedPath(hPrefix, fc1));
     Assert.assertEquals(6, paths.length);
     for (int i = 0; i < dirs.length; i++) {
       boolean found = false;
@@ -517,7 +519,30 @@ public abstract class FileContextURIBase {
       Assert.assertTrue(dirs[i] + " not found", found);
     }
 
-    paths = fc1.listStatus(qualifiedPath(dirs[0], fc1));
+    paths = fc1.util().listStatus(qualifiedPath(dirs[0], fc1));
     Assert.assertEquals(0, paths.length);
+    
+    // test listStatus that returns an iterator of FileStatus
+    Iterator<FileStatus> pathsItor = fc1.listStatus(qualifiedPath("test", fc1));
+    Assert.assertEquals(qualifiedPath(hPrefix, fc1), pathsItor.next().getPath());
+    Assert.assertFalse(pathsItor.hasNext());
+
+    pathsItor = fc1.listStatus(qualifiedPath(hPrefix, fc1));
+    int dirLen = 0;
+    for (; pathsItor.hasNext(); dirLen++) {
+      boolean found = false;
+      FileStatus stat = pathsItor.next();
+      for (int j = 0; j < dirs.length; j++) {
+        if (qualifiedPath(dirs[j],fc1).equals(stat.getPath())) {
+          found = true;
+          break;
+        }
+      }
+      Assert.assertTrue(stat.getPath() + " not found", found);
+    }
+    Assert.assertEquals(6, dirLen);
+
+    pathsItor = fc1.listStatus(qualifiedPath(dirs[0], fc1));
+    Assert.assertFalse(pathsItor.hasNext());
   }
 }
diff --git a/src/test/core/org/apache/hadoop/fs/TestFilterFs.java b/src/test/core/org/apache/hadoop/fs/TestFilterFs.java
index 757b3394645..352cf10907e 100644
--- a/src/test/core/org/apache/hadoop/fs/TestFilterFs.java
+++ b/src/test/core/org/apache/hadoop/fs/TestFilterFs.java
@@ -21,6 +21,7 @@ package org.apache.hadoop.fs;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.net.URI;
+import java.util.Iterator;
 
 import junit.framework.TestCase;
 import org.apache.commons.logging.Log;
@@ -31,6 +32,9 @@ public class TestFilterFs extends TestCase {
 
   public static class DontCheck {
     public void checkScheme(URI uri, String supportedScheme) { }
+    public Iterator<FileStatus> listStatusIterator(Path f) {
+      return null;
+    }
   }
   
   public void testFilterFileSystem() throws Exception {