org.apache.hadoop.fs.FileStatus#getOwner ( )源码实例Demo

private static void updatePermissions(FileStatus src, FileStatus dst,
    EnumSet<FileAttribute> preseved, FileSystem destFileSys
    ) throws IOException {
  String owner = null;
  String group = null;
  if (preseved.contains(FileAttribute.USER)
      && !src.getOwner().equals(dst.getOwner())) {
    owner = src.getOwner();
  }
  if (preseved.contains(FileAttribute.GROUP)
      && !src.getGroup().equals(dst.getGroup())) {
    group = src.getGroup();
  }
  if (owner != null || group != null) {
    destFileSys.setOwner(dst.getPath(), owner, group);
  }
  if (preseved.contains(FileAttribute.PERMISSION)
      && !src.getPermission().equals(dst.getPermission())) {
    destFileSys.setPermission(dst.getPath(), src.getPermission());
  }
}
 

private void cleanUpFilesPerUserDir(FileContext lfs, DeletionService del,
    Path userDirPath) throws IOException {
  RemoteIterator<FileStatus> userDirStatus = lfs.listStatus(userDirPath);
  FileDeletionTask dependentDeletionTask =
      del.createFileDeletionTask(null, userDirPath, new Path[] {});
  if (userDirStatus != null && userDirStatus.hasNext()) {
    List<FileDeletionTask> deletionTasks = new ArrayList<FileDeletionTask>();
    while (userDirStatus.hasNext()) {
      FileStatus status = userDirStatus.next();
      String owner = status.getOwner();
      FileDeletionTask deletionTask =
          del.createFileDeletionTask(owner, null,
            new Path[] { status.getPath() });
      deletionTask.addFileDeletionTaskDependency(dependentDeletionTask);
      deletionTasks.add(deletionTask);
    }
    for (FileDeletionTask task : deletionTasks) {
      del.scheduleFileDeletionTask(task);
    }
  } else {
    del.scheduleFileDeletionTask(dependentDeletionTask);
  }
}
 

@BeforeClass
public static void makeTestFile() throws Exception {
  Configuration conf = new Configuration();
  fs = FileSystem.getLocal(conf).getRaw();
  testFilePathIs =
      new File((new Path("target", TestSecureIOUtils.class.getSimpleName()
          + "1")).toUri().getRawPath());
  testFilePathRaf =
      new File((new Path("target", TestSecureIOUtils.class.getSimpleName()
          + "2")).toUri().getRawPath());
  testFilePathFadis =
      new File((new Path("target", TestSecureIOUtils.class.getSimpleName()
          + "3")).toUri().getRawPath());
  for (File f : new File[] { testFilePathIs, testFilePathRaf,
      testFilePathFadis }) {
    FileOutputStream fos = new FileOutputStream(f);
    fos.write("hello".getBytes("UTF-8"));
    fos.close();
  }

  FileStatus stat = fs.getFileStatus(
      new Path(testFilePathIs.toString()));
  // RealOwner and RealGroup would be same for all three files.
  realOwner = stat.getOwner();
  realGroup = stat.getGroup();
}
 

/**
 * Obtain the owner of the log dir. This is 
 * determined by checking the job's log directory.
 */
static String obtainLogDirOwner(TaskAttemptID taskid) throws IOException {
  Configuration conf = new Configuration();
  FileSystem raw = FileSystem.getLocal(conf).getRaw();
  Path jobLogDir = new Path(getJobDir(taskid.getJobID()).getAbsolutePath());
  FileStatus jobStat = raw.getFileStatus(jobLogDir);
  return jobStat.getOwner();
}
 

/**
 * Initializes the staging directory and returns the path. It also
 * keeps track of all necessary ownership and permissions
 * @param cluster
 * @param conf
 */
public static Path getStagingDir(Cluster cluster, Configuration conf) 
throws IOException,InterruptedException {
  Path stagingArea = cluster.getStagingAreaDir();
  FileSystem fs = stagingArea.getFileSystem(conf);
  String realUser;
  String currentUser;
  UserGroupInformation ugi = UserGroupInformation.getLoginUser();
  realUser = ugi.getShortUserName();
  currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
  if (fs.exists(stagingArea)) {
    FileStatus fsStatus = fs.getFileStatus(stagingArea);
    String owner = fsStatus.getOwner();
    if (!(owner.equals(currentUser) || owner.equals(realUser))) {
       throw new IOException("The ownership on the staging directory " +
                    stagingArea + " is not as expected. " +
                    "It is owned by " + owner + ". The directory must " +
                    "be owned by the submitter " + currentUser + " or " +
                    "by " + realUser);
    }
    if (!fsStatus.getPermission().equals(JOB_DIR_PERMISSION)) {
      LOG.info("Permissions on staging directory " + stagingArea + " are " +
        "incorrect: " + fsStatus.getPermission() + ". Fixing permissions " +
        "to correct value " + JOB_DIR_PERMISSION);
      fs.setPermission(stagingArea, JOB_DIR_PERMISSION);
    }
  } else {
    fs.mkdirs(stagingArea, 
        new FsPermission(JOB_DIR_PERMISSION));
  }
  return stagingArea;
}
 

@Test
public void testWrite() throws Exception {
  String streamString = "testContents";

  FileStatus status = fs.getFileStatus(testTempPath);
  OwnerAndPermission ownerAndPermission =
      new OwnerAndPermission(status.getOwner(), status.getGroup(), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
  CopyableFile cf = CopyableFileUtils.getTestCopyableFile(ownerAndPermission);

  CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(new TestCopyableDataset(new Path("/source")));

  WorkUnitState state = TestUtils.createTestWorkUnitState();
  state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, new Path(testTempPath, "staging").toString());
  state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, new Path(testTempPath, "output").toString());
  state.setProp(ConfigurationKeys.WRITER_FILE_PATH, RandomStringUtils.randomAlphabetic(5));
  CopySource.serializeCopyEntity(state, cf);
  CopySource.serializeCopyableDataset(state, metadata);

  FileAwareInputStreamDataWriter dataWriter = new FileAwareInputStreamDataWriter(state, 1, 0);

  FileAwareInputStream fileAwareInputStream = FileAwareInputStream.builder().file(cf)
      .inputStream(StreamUtils.convertStream(IOUtils.toInputStream(streamString))).build();
  dataWriter.write(fileAwareInputStream);
  dataWriter.commit();
  Path writtenFilePath = new Path(new Path(state.getProp(ConfigurationKeys.WRITER_OUTPUT_DIR),
      cf.getDatasetAndPartition(metadata).identifier()), cf.getDestination());
  Assert.assertEquals(IOUtils.toString(new FileInputStream(writtenFilePath.toString())), streamString);
}
 

/**
 * Converts a Hadoop {@link FileStatus} instance into a protobuf
 * {@link DFSProtos.FileStatus}
 *
 * @param status
 *          the Hadoop status instance to convert
 * @return a protobuf status instance
 * @throws IOException
 */
static DFS.FileStatus toProtoFileStatus(FileStatus status) throws IOException {
  DFS.FileStatus.Builder builder = DFS.FileStatus.newBuilder();

  builder
    .setLength(status.getLen())
    .setIsDirectory(status.isDirectory())
    .setBlockReplication(status.getReplication())
    .setBlockSize(status.getBlockSize())
    .setModificationTime(status.getModificationTime())
    .setAccessTime(status.getAccessTime());

  // Handling potential null values
  if (status.getPath() != null) {
    builder = builder.setPath(status.getPath().toUri().getPath());
  }
  if (status.getPermission() != null) {
    builder = builder.setPermission(status.getPermission().toExtendedShort());
  }
  if (status.getOwner() != null) {
    builder = builder.setOwner(status.getOwner());
  }
  if (status.getGroup() != null) {
    builder = builder.setGroup(status.getGroup());
  }
  if (status.isSymlink()) {
    builder = builder.setSymlink(status.getSymlink().toString());
  }

  return builder.build();
}
 

/**
 * Verify or create the Staging area directory on the configured Filesystem
 * @param stagingArea Staging area directory path
 * @return the FileSytem for the staging area directory
 * @throws IOException
 */
public static FileSystem ensureStagingDirExists(Configuration conf,
    Path stagingArea)
    throws IOException {
  FileSystem fs = stagingArea.getFileSystem(conf);
  String realUser;
  String currentUser;
  UserGroupInformation ugi = UserGroupInformation.getLoginUser();
  realUser = ugi.getShortUserName();
  currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
  if (fs.exists(stagingArea)) {
    FileStatus fsStatus = fs.getFileStatus(stagingArea);
    String owner = fsStatus.getOwner();
    if (!(owner.equals(currentUser) || owner.equals(realUser))) {
      throw new IOException("The ownership on the staging directory "
          + stagingArea + " is not as expected. " + "It is owned by " + owner
          + ". The directory must " + "be owned by the submitter "
          + currentUser + " or " + "by " + realUser);
    }
    if (!fsStatus.getPermission().equals(TezCommonUtils.TEZ_AM_DIR_PERMISSION)) {
      LOG.info("Permissions on staging directory " + stagingArea + " are "
          + "incorrect: " + fsStatus.getPermission()
          + ". Fixing permissions " + "to correct value "
          + TezCommonUtils.TEZ_AM_DIR_PERMISSION);
      fs.setPermission(stagingArea, TezCommonUtils.TEZ_AM_DIR_PERMISSION);
    }
  } else {
    TezCommonUtils.mkDirForAM(fs, stagingArea);
  }
  return fs;
}
 

/**
 * Obtain the owner of the log dir. This is 
 * determined by checking the job's log directory.
 */
static String obtainLogDirOwner(TaskAttemptID taskid) throws IOException {
  Configuration conf = new Configuration();
  FileSystem raw = FileSystem.getLocal(conf).getRaw();
  Path jobLogDir = new Path(getJobDir(taskid.getJobID()).getAbsolutePath());
  FileStatus jobStat = raw.getFileStatus(jobLogDir);
  return jobStat.getOwner();
}
 

@Test
public void testBlockWrite() throws Exception {
  String streamString = "testContents";

  FileStatus status = fs.getFileStatus(testTempPath);
  OwnerAndPermission ownerAndPermission =
      new OwnerAndPermission(status.getOwner(), status.getGroup(), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
  CopyableFile cf = CopyableFileUtils.getTestCopyableFile(ownerAndPermission);

  CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(new TestCopyableDataset(new Path("/source")));

  WorkUnitState state = TestUtils.createTestWorkUnitState();
  state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, new Path(testTempPath, "staging").toString());
  state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, new Path(testTempPath, "output").toString());
  state.setProp(ConfigurationKeys.WRITER_FILE_PATH, RandomStringUtils.randomAlphabetic(5));
  state.setProp(DistcpFileSplitter.SPLIT_ENABLED, true);
  CopySource.serializeCopyEntity(state, cf);
  CopySource.serializeCopyableDataset(state, metadata);

  FileAwareInputStreamDataWriter dataWriter = new FileAwareInputStreamDataWriter(state, 1, 0);

  long splitLen = 4;
  int splits = (int) (streamString.length() / splitLen + 1);
  DistcpFileSplitter.Split split = new DistcpFileSplitter.Split(0, splitLen, 0, splits,
      String.format("%s.__PART%d__", cf.getDestination().getName(), 0));
  FSDataInputStream dataInputStream = StreamUtils.convertStream(IOUtils.toInputStream(streamString));
  dataInputStream.seek(split.getLowPosition());
  FileAwareInputStream fileAwareInputStream = FileAwareInputStream.builder().file(cf)
      .inputStream(dataInputStream)
      .split(Optional.of(split))
      .build();
  dataWriter.write(fileAwareInputStream);
  dataWriter.commit();
  Path writtenFilePath = new Path(new Path(state.getProp(ConfigurationKeys.WRITER_OUTPUT_DIR),
      cf.getDatasetAndPartition(metadata).identifier()), cf.getDestination());
  Assert.assertEquals(IOUtils.toString(new FileInputStream(writtenFilePath.toString())),
      streamString.substring(0, (int) splitLen));
}
 

@Test
public void testWriteWithEncryption() throws Exception {
  byte[] streamString = "testEncryptedContents".getBytes("UTF-8");
  byte[] expectedContents = new byte[streamString.length];
  for (int i = 0; i < streamString.length; i++) {
    expectedContents[i] = (byte)((streamString[i] + 1) % 256);
  }

  FileStatus status = fs.getFileStatus(testTempPath);
  OwnerAndPermission ownerAndPermission =
      new OwnerAndPermission(status.getOwner(), status.getGroup(), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
  CopyableFile cf = CopyableFileUtils.getTestCopyableFile(ownerAndPermission);

  CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(new TestCopyableDataset(new Path("/source")));

  WorkUnitState state = TestUtils.createTestWorkUnitState();
  state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, new Path(testTempPath, "staging").toString());
  state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, new Path(testTempPath, "output").toString());
  state.setProp(ConfigurationKeys.WRITER_FILE_PATH, RandomStringUtils.randomAlphabetic(5));
  state.setProp("writer.encrypt." + EncryptionConfigParser.ENCRYPTION_ALGORITHM_KEY, "insecure_shift");

  CopySource.serializeCopyEntity(state, cf);
  CopySource.serializeCopyableDataset(state, metadata);

  FileAwareInputStreamDataWriter dataWriter = new FileAwareInputStreamDataWriter(state, 1, 0);

  FileAwareInputStream fileAwareInputStream = FileAwareInputStream.builder().file(cf)
      .inputStream(StreamUtils.convertStream(new ByteArrayInputStream(streamString))).build();
  dataWriter.write(fileAwareInputStream);
  dataWriter.commit();

  Path writtenFilePath = new Path(new Path(state.getProp(ConfigurationKeys.WRITER_OUTPUT_DIR),
      cf.getDatasetAndPartition(metadata).identifier()), cf.getDestination());
  Assert.assertTrue(writtenFilePath.getName().endsWith("insecure_shift"),
      "Expected encryption name to be appended to destination");
  Assert.assertEquals(IOUtils.toByteArray(new FileInputStream(writtenFilePath.toString())), expectedContents);
}
 

public FileStatusExtended(FileStatus stat, Block[] blocks, String leaseHolder) {
  super(stat.getLen(), stat.isDir(), stat.getReplication(),
      stat.getBlockSize(), stat.getModificationTime(), stat.getAccessTime(),
      stat.getPermission(), stat.getOwner(), stat.getGroup(), 
      stat.getPath());
  this.blocks = blocks;
  this.leaseHolder = (leaseHolder == null) ? "" : leaseHolder;
}
 

public FileStatusEntity(FileStatus status) throws IOException {
    //this.path = status.getPath();
    this.length = status.getLen();
    this.isdir = status.isDirectory();
    this.block_replication = status.getReplication();
    this.blocksize = status.getBlockSize();
    this.modification_time = status.getModificationTime();
    this.access_time = status.getAccessTime();
    this.permission = status.getPermission();
    this.owner = status.getOwner();
    this.group = status.getGroup();
    if(status.isSymlink()) {
        this.symlink = status.getSymlink();
    }
}
 

protected static String getTraceSignature(String input) throws IOException {
  Path inputPath = new Path(input);
  FileSystem fs = inputPath.getFileSystem(new Configuration());
  FileStatus status = fs.getFileStatus(inputPath);
  Path qPath = fs.makeQualified(status.getPath());
  String traceID = status.getModificationTime() + qPath.toString()
                   + status.getOwner() + status.getLen();
  return MD5Hash.digest(traceID).toString();
}
 

@Override
public FileStatus getFileStatus(Path inlinePath) throws IOException {
  Path outerPath = InLineFSUtils.getOuterfilePathFromInlinePath(inlinePath);
  FileSystem outerFs = outerPath.getFileSystem(conf);
  FileStatus status = outerFs.getFileStatus(outerPath);
  FileStatus toReturn = new FileStatus(InLineFSUtils.length(inlinePath), status.isDirectory(), status.getReplication(), status.getBlockSize(),
      status.getModificationTime(), status.getAccessTime(), status.getPermission(), status.getOwner(),
      status.getGroup(), inlinePath);
  return toReturn;
}
 

ChPermissionStatus(FileStatus filestatus, String owner, String group, String permission) {
  super("".equals(owner)? filestatus.getOwner(): owner, 
      "".equals(group)? filestatus.getGroup(): group,
      "".equals(permission)? filestatus.getPermission(): new FsPermission(Short.parseShort(permission, 8)));
}
 

ChPermissionStatus(FileStatus filestatus, String owner, String group, String permission) {
  super("".equals(owner)? filestatus.getOwner(): owner, 
      "".equals(group)? filestatus.getGroup(): group,
      "".equals(permission)? filestatus.getPermission(): new FsPermission(Short.parseShort(permission, 8)));
}
 

ChPermissionStatus(FileStatus filestatus, String owner, String group, String permission) {
  super("".equals(owner)? filestatus.getOwner(): owner, 
      "".equals(group)? filestatus.getGroup(): group,
      "".equals(permission)? filestatus.getPermission(): new FsPermission(Short.parseShort(permission, 8)));
  defaultPerm = permission == null || "".equals(permission);
}
 

/**
 * Preserve attribute on file matching that of the file status being sent
 * as argument. Barring the block size, all the other attributes are preserved
 * by this function
 *
 * @param targetFS - File system
 * @param path - Path that needs to preserve original file status
 * @param srcFileStatus - Original file status
 * @param attributes - Attribute set that needs to be preserved
 * @param preserveRawXattrs if true, raw.* xattrs should be preserved
 * @throws IOException - Exception if any (particularly relating to group/owner
 *                       change or any transient error)
 */
public static void preserve(FileSystem targetFS, Path path,
                            CopyListingFileStatus srcFileStatus,
                            EnumSet<FileAttribute> attributes,
                            boolean preserveRawXattrs) throws IOException {

  FileStatus targetFileStatus = targetFS.getFileStatus(path);
  String group = targetFileStatus.getGroup();
  String user = targetFileStatus.getOwner();
  boolean chown = false;

  if (attributes.contains(FileAttribute.ACL)) {
    List<AclEntry> srcAcl = srcFileStatus.getAclEntries();
    List<AclEntry> targetAcl = getAcl(targetFS, targetFileStatus);
    if (!srcAcl.equals(targetAcl)) {
      targetFS.setAcl(path, srcAcl);
    }
    // setAcl doesn't preserve sticky bit, so also call setPermission if needed.
    if (srcFileStatus.getPermission().getStickyBit() !=
        targetFileStatus.getPermission().getStickyBit()) {
      targetFS.setPermission(path, srcFileStatus.getPermission());
    }
  } else if (attributes.contains(FileAttribute.PERMISSION) &&
    !srcFileStatus.getPermission().equals(targetFileStatus.getPermission())) {
    targetFS.setPermission(path, srcFileStatus.getPermission());
  }

  final boolean preserveXAttrs = attributes.contains(FileAttribute.XATTR);
  if (preserveXAttrs || preserveRawXattrs) {
    final String rawNS =
        StringUtils.toLowerCase(XAttr.NameSpace.RAW.name());
    Map<String, byte[]> srcXAttrs = srcFileStatus.getXAttrs();
    Map<String, byte[]> targetXAttrs = getXAttrs(targetFS, path);
    if (srcXAttrs != null && !srcXAttrs.equals(targetXAttrs)) {
      for (Entry<String, byte[]> entry : srcXAttrs.entrySet()) {
        String xattrName = entry.getKey();
        if (xattrName.startsWith(rawNS) || preserveXAttrs) {
          targetFS.setXAttr(path, xattrName, entry.getValue());
        }
      }
    }
  }

  if (attributes.contains(FileAttribute.REPLICATION) && !targetFileStatus.isDirectory() &&
      (srcFileStatus.getReplication() != targetFileStatus.getReplication())) {
    targetFS.setReplication(path, srcFileStatus.getReplication());
  }

  if (attributes.contains(FileAttribute.GROUP) &&
      !group.equals(srcFileStatus.getGroup())) {
    group = srcFileStatus.getGroup();
    chown = true;
  }

  if (attributes.contains(FileAttribute.USER) &&
      !user.equals(srcFileStatus.getOwner())) {
    user = srcFileStatus.getOwner();
    chown = true;
  }

  if (chown) {
    targetFS.setOwner(path, user, group);
  }
  
  if (attributes.contains(FileAttribute.TIMES)) {
    targetFS.setTimes(path, 
        srcFileStatus.getModificationTime(), 
        srcFileStatus.getAccessTime());
  }
}
 

@Test
public void testWriteWithGPGSymmetricEncryption() throws Exception {
  byte[] streamString = "testEncryptedContents".getBytes("UTF-8");

  FileStatus status = fs.getFileStatus(testTempPath);
  OwnerAndPermission ownerAndPermission =
      new OwnerAndPermission(status.getOwner(), status.getGroup(), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
  CopyableFile cf = CopyableFileUtils.getTestCopyableFile(ownerAndPermission);

  CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(new TestCopyableDataset(new Path("/source")));

  WorkUnitState state = TestUtils.createTestWorkUnitState();
  state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, new Path(testTempPath, "staging").toString());
  state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, new Path(testTempPath, "output").toString());
  state.setProp(ConfigurationKeys.WRITER_FILE_PATH, RandomStringUtils.randomAlphabetic(5));
  state.setProp("writer.encrypt." + EncryptionConfigParser.ENCRYPTION_ALGORITHM_KEY, "gpg");
  state.setProp("writer.encrypt." + EncryptionConfigParser.ENCRYPTION_KEYSTORE_PASSWORD_KEY, "testPassword");

  CopySource.serializeCopyEntity(state, cf);
  CopySource.serializeCopyableDataset(state, metadata);

  FileAwareInputStreamDataWriter dataWriter = new FileAwareInputStreamDataWriter(state, 1, 0);

  FileAwareInputStream fileAwareInputStream = FileAwareInputStream.builder().file(cf)
      .inputStream(StreamUtils.convertStream(new ByteArrayInputStream(streamString))).build();
  dataWriter.write(fileAwareInputStream);
  dataWriter.commit();

  Path writtenFilePath = new Path(new Path(state.getProp(ConfigurationKeys.WRITER_OUTPUT_DIR),
      cf.getDatasetAndPartition(metadata).identifier()), cf.getDestination());
  Assert.assertTrue(writtenFilePath.getName().endsWith("gpg"),
      "Expected encryption name to be appended to destination");
  byte[] encryptedContent = IOUtils.toByteArray(new FileInputStream(writtenFilePath.toString()));
  byte[] decryptedContent = new byte[streamString.length];
  IOUtils.readFully(GPGFileDecryptor.decryptFile(new FileInputStream(writtenFilePath.toString()), "testPassword"),
      decryptedContent);


  // encrypted string should not be the same as the plaintext
  Assert.assertNotEquals(encryptedContent, streamString);

  // decrypted string should be the same as the plaintext
  Assert.assertEquals(decryptedContent, streamString);

}