put new status when create object
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a8bbfa93 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a8bbfa93 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a8bbfa93 Branch: refs/heads/s3_create Commit: a8bbfa93ea9fceb4cb182e76b704c62241b501ff Parents: b7e876a Author: Lei Xu <l...@cloudera.com> Authored: Fri Jul 8 14:21:40 2016 -0700 Committer: Lei Xu <l...@cloudera.com> Committed: Fri Jul 8 14:21:40 2016 -0700 ---------------------------------------------------------------------- .../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 5 +++-- .../org/apache/hadoop/fs/s3a/S3AOutputStream.java | 18 +++++++++++++++--- 2 files changed, 18 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/a8bbfa93/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java index 6ac1043..eaa15ca 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java @@ -497,7 +497,7 @@ public class S3AFileSystem extends FileSystem { /* Turns a path (relative or otherwise) into an S3 key */ - private String pathToKey(Path path) { + String pathToKey(Path path) { if (!path.isAbsolute()) { path = new Path(workingDir, path); } @@ -591,7 +591,8 @@ public class S3AFileSystem extends FileSystem { return new FSDataOutputStream( new S3AOutputStream(getConf(), this, - key, + f, + this.metadataStore, progress ), null); http://git-wip-us.apache.org/repos/asf/hadoop/blob/a8bbfa93/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java index 23ba682..ee285b9 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AOutputStream.java @@ -25,6 +25,7 @@ import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.LocalDirAllocator; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.util.Progressable; import org.slf4j.Logger; @@ -48,21 +49,25 @@ public class S3AOutputStream extends OutputStream { private OutputStream backupStream; private File backupFile; private boolean closed; + private Path path; private String key; private Progressable progress; private long partSize; private long partSizeThreshold; private S3AFileSystem fs; + private final MetadataStore metadataStore; private LocalDirAllocator lDirAlloc; public static final Logger LOG = S3AFileSystem.LOG; public S3AOutputStream(Configuration conf, - S3AFileSystem fs, String key, Progressable progress) + S3AFileSystem fs, Path f, MetadataStore store, Progressable progress) throws IOException { - this.key = key; + this.path = f; + this.key = fs.pathToKey(f); this.progress = progress; this.fs = fs; + this.metadataStore = store; partSize = fs.getPartitionSize(); partSizeThreshold = fs.getMultiPartThreshold(); @@ -100,7 +105,6 @@ public class S3AOutputStream extends OutputStream { LOG.debug("Minimum upload part size: {} threshold {}" , partSize, partSizeThreshold); - try { final ObjectMetadata om = fs.newObjectMetadata(); Upload upload = fs.putObject( @@ -116,6 +120,14 @@ public class S3AOutputStream extends OutputStream { listener.uploadCompleted(); // This will delete unnecessary fake parent directories fs.finishedWrite(key); + if (metadataStore != null) { + // Every newly write keeps a record in the metadata store. + Path p = path.makeQualified(fs.getUri(), fs.getWorkingDirectory()); + S3AFileStatus status = new S3AFileStatus(backupFile.length(), + om.getLastModified().getTime(), p, fs.getDefaultBlockSize(p)); + CachedFileStatus cfs = new CachedFileStatus(status); + metadataStore.putNew(cfs); + } } catch (InterruptedException e) { throw (InterruptedIOException) new InterruptedIOException(e.toString()) .initCause(e); --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org
