[incubator-gobblin] branch master updated: [GOBBLIN-1374] Generate a unique staging path in FS SpecProducer for writing job specs[]

[lesun]

This is an automated email from the ASF dual-hosted git repository.

lesun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new 5b5b1f9  [GOBBLIN-1374] Generate a unique staging path in FS 
SpecProducer for writing job specs[]
5b5b1f9 is described below

commit 5b5b1f98b0b232bd5305fa572224d30511f14e46
Author: suvasude <suvas...@linkedin.biz>
AuthorDate: Wed Jan 27 12:08:15 2021 -0800

    [GOBBLIN-1374] Generate a unique staging path in FS SpecProducer for 
writing job specs[]
    
    Closes #3216 from sv2000/specProducerUniqueStaging
---
 .../main/java/org/apache/gobblin/runtime/api/FsSpecProducer.java   | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git 
a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FsSpecProducer.java
 
b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FsSpecProducer.java
index 4471ec9..e8264eb 100644
--- 
a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FsSpecProducer.java
+++ 
b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/FsSpecProducer.java
@@ -21,6 +21,7 @@ import java.io.OutputStream;
 import java.net.URI;
 import java.util.List;
 import java.util.Properties;
+import java.util.UUID;
 import java.util.concurrent.Future;
 
 import org.apache.avro.file.DataFileWriter;
@@ -140,7 +141,7 @@ public class FsSpecProducer implements SpecProducer<Spec> {
     Path jobSpecPath = new Path(this.specConsumerPath, jobSpec.getUri());
 
     //Write the new JobSpec to a temporary path first.
-    Path tmpDir = new Path(this.specConsumerPath, "_tmp");
+    Path tmpDir = new Path(this.specConsumerPath, 
UUID.randomUUID().toString());
     if (!fs.exists(tmpDir)) {
       fs.mkdirs(tmpDir);
     }
@@ -155,6 +156,10 @@ public class FsSpecProducer implements SpecProducer<Spec> {
 
     //Rename the JobSpec from temporary to final location.
     HadoopUtils.renamePath(fs, tmpJobSpecPath, jobSpecPath, true);
+
+    //Delete the temporary path once the jobspec has been moved to its final 
publish location.
+    log.info("Deleting {}", tmpJobSpecPath.getParent().toString());
+    fs.delete(tmpJobSpecPath.getParent(), true);
   }
 
 }
\ No newline at end of file