HADOOP-12774. s3a should use UGI.getCurrentUser.getShortname() for username. Contributed by Steve Loughran.

(cherry picked from commit 3372e94030)
(cherry picked from commit 5c2f67bdae)

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java

@@ -298,7 +298,7 @@ public class Listing {
         // Skip over keys that are ourselves and old S3N _$folder$ files
         if (acceptor.accept(keyPath, summary) && filter.accept(keyPath)) {
           FileStatus status = createFileStatus(keyPath, summary,
-              owner.getDefaultBlockSize(keyPath));
+              owner.getDefaultBlockSize(keyPath), owner.getUsername());
           LOG.debug("Adding: {}", status);
           stats.add(status);
           added++;
@@ -312,7 +312,8 @@ public class Listing {
       for (String prefix : objects.getCommonPrefixes()) {
         Path keyPath = owner.keyToQualifiedPath(prefix);
         if (acceptor.accept(keyPath, prefix) && filter.accept(keyPath)) {
-          FileStatus status = new S3AFileStatus(true, false, keyPath);
+          FileStatus status = new S3AFileStatus(false, keyPath,
+              owner.getUsername());
           LOG.debug("Adding directory: {}", status);
           added++;
           stats.add(status);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java

@@ -33,28 +33,41 @@ import org.apache.hadoop.fs.Path;
 public class S3AFileStatus extends FileStatus {
   private boolean isEmptyDirectory;
 
-  // Directories
+  /**
-  public S3AFileStatus(boolean isdir, boolean isemptydir, Path path) {
+   * Create a directory status.
-    super(0, isdir, 1, 0, 0, path);
+   * @param isemptydir is this an empty directory?
+   * @param path the path
+   * @param owner the owner
+   */
+  public S3AFileStatus(boolean isemptydir,
+      Path path,
+      String owner) {
+    super(0, true, 1, 0, 0, path);
     isEmptyDirectory = isemptydir;
+    setOwner(owner);
+    setGroup(owner);
   }
 
-  // Files
+  /**
+   * A simple file.
+   * @param length file length
+   * @param modification_time mod time
+   * @param path path
+   * @param blockSize block size
+   * @param owner owner
+   */
   public S3AFileStatus(long length, long modification_time, Path path,
-      long blockSize) {
+      long blockSize, String owner) {
     super(length, false, 1, blockSize, modification_time, path);
     isEmptyDirectory = false;
+    setOwner(owner);
+    setGroup(owner);
   }
 
   public boolean isEmptyDirectory() {
     return isEmptyDirectory;
   }
 
-  @Override
-  public String getOwner() {
-    return System.getProperty("user.name");
-  }
-
   /** Compare if this object is equal to another object.
    * @param   o the object to be compared.
    * @return  true if two file status has the same path name; false if not.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -86,6 +86,7 @@ import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.StorageStatistics;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.s3native.S3xLoginHelper;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Progressable;
 import org.apache.hadoop.util.ReflectionUtils;
 
@@ -119,6 +120,7 @@ public class S3AFileSystem extends FileSystem {
   public static final int DEFAULT_BLOCKSIZE = 32 * 1024 * 1024;
   private URI uri;
   private Path workingDir;
+  private String username;
   private AmazonS3 s3;
   private String bucket;
   private int maxKeys;
@@ -159,7 +161,9 @@ public class S3AFileSystem extends FileSystem {
       instrumentation = new S3AInstrumentation(name);
 
       uri = S3xLoginHelper.buildFSURI(name);
-      workingDir = new Path("/user", System.getProperty("user.name"))
+      // Username is the current user at the time the FS was instantiated.
+      username = UserGroupInformation.getCurrentUser().getShortUserName();
+      workingDir = new Path("/user", username)
           .makeQualified(this.uri, this.getWorkingDirectory());
 
       bucket = name.getHost();
@@ -1386,6 +1390,14 @@ public class S3AFileSystem extends FileSystem {
     return workingDir;
   }
 
+  /**
+   * Get the username of the FS.
+   * @return the short name of the user who instantiated the FS
+   */
+  public String getUsername() {
+    return username;
+  }
+
   /**
    *
    * Make the given path and all non-existent parents into
@@ -1477,14 +1489,14 @@ public class S3AFileSystem extends FileSystem {
 
         if (objectRepresentsDirectory(key, meta.getContentLength())) {
           LOG.debug("Found exact file: fake directory");
-          return new S3AFileStatus(true, true,
+          return new S3AFileStatus(true, path, username);
-              path);
         } else {
           LOG.debug("Found exact file: normal file");
           return new S3AFileStatus(meta.getContentLength(),
               dateToLong(meta.getLastModified()),
               path,
-              getDefaultBlockSize(path));
+              getDefaultBlockSize(path),
+              username);
         }
       } catch (AmazonServiceException e) {
         if (e.getStatusCode() != 404) {
@@ -1502,7 +1514,7 @@ public class S3AFileSystem extends FileSystem {
 
           if (objectRepresentsDirectory(newKey, meta.getContentLength())) {
             LOG.debug("Found file (with /): fake directory");
-            return new S3AFileStatus(true, true, path);
+            return new S3AFileStatus(true, path, username);
           } else {
             LOG.warn("Found file (with /): real file? should not happen: {}",
                 key);
@@ -1510,7 +1522,8 @@ public class S3AFileSystem extends FileSystem {
             return new S3AFileStatus(meta.getContentLength(),
                 dateToLong(meta.getLastModified()),
                 path,
-                getDefaultBlockSize(path));
+                getDefaultBlockSize(path),
+                username);
           }
         } catch (AmazonServiceException e) {
           if (e.getStatusCode() != 404) {
@@ -1547,10 +1560,10 @@ public class S3AFileSystem extends FileSystem {
           }
         }
 
-        return new S3AFileStatus(true, false, path);
+        return new S3AFileStatus(false, path, username);
       } else if (key.isEmpty()) {
         LOG.debug("Found root directory");
-        return new S3AFileStatus(true, true, path);
+        return new S3AFileStatus(true, path, username);
       }
     } catch (AmazonServiceException e) {
       if (e.getStatusCode() != 404) {

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java

@@ -241,17 +241,19 @@ public final class S3AUtils {
    * @param keyPath path to entry
    * @param summary summary from AWS
    * @param blockSize block size to declare.
+   * @param owner owner of the file
    * @return a status entry
    */
   public static S3AFileStatus createFileStatus(Path keyPath,
       S3ObjectSummary summary,
-      long blockSize) {
+      long blockSize,
+      String owner) {
     if (objectRepresentsDirectory(summary.getKey(), summary.getSize())) {
-      return new S3AFileStatus(true, true, keyPath);
+      return new S3AFileStatus(true, keyPath, owner);
     } else {
       return new S3AFileStatus(summary.getSize(),
           dateToLong(summary.getLastModified()), keyPath,
-          blockSize);
+          blockSize, owner);
     }
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java

@@ -44,8 +44,10 @@ import static org.junit.Assert.fail;
 
 import java.io.File;
 import java.net.URI;
+import java.security.PrivilegedExceptionAction;
 
 import org.apache.hadoop.security.ProviderUtils;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.alias.CredentialProvider;
 import org.apache.hadoop.security.alias.CredentialProviderFactory;
 import org.apache.hadoop.util.VersionInfo;
@@ -434,7 +436,7 @@ public class ITestS3AConfiguration {
     dir1.mkdirs();
     dir2.mkdirs();
     conf = new Configuration();
-    conf.set(Constants.BUFFER_DIR, dir1 +", " + dir2);
+    conf.set(Constants.BUFFER_DIR, dir1 + ", " + dir2);
     fs = S3ATestUtils.createTestFileSystem(conf);
     File tmp1 = fs.createTmpFileForWrite("out-", 1024, conf);
     tmp1.delete();
@@ -444,6 +446,25 @@ public class ITestS3AConfiguration {
         tmp1.getParent(), tmp2.getParent());
   }
 
+  @Test
+  public void testUsernameFromUGI() throws Throwable {
+    final String alice = "alice";
+    UserGroupInformation fakeUser =
+        UserGroupInformation.createUserForTesting(alice,
+            new String[]{"users", "administrators"});
+    conf = new Configuration();
+    fs = fakeUser.doAs(new PrivilegedExceptionAction<S3AFileSystem>() {
+      @Override
+      public S3AFileSystem run() throws Exception{
+        return S3ATestUtils.createTestFileSystem(conf);
+      }
+    });
+    assertEquals("username", alice, fs.getUsername());
+    S3AFileStatus status = fs.getFileStatus(new Path("/"));
+    assertEquals("owner in " + status, alice, status.getOwner());
+    assertEquals("group in " + status, alice, status.getGroup());
+  }
+
   /**
    * Reads and returns a field from an object using reflection.  If the field
    * cannot be found, is null, or is not the expected type, then this method