Merge pull request #316 from pcadabam/master

Fix group in FsWriter
apache · Sep 1, 2015 · 292c270 · 292c270
2 parents 6632b5b + 8c4d434
commit 292c270
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 2 deletions.
diff --git a/gobblin-azkaban/src/main/java/gobblin/azkaban/AzkabanJobLauncher.java b/gobblin-azkaban/src/main/java/gobblin/azkaban/AzkabanJobLauncher.java
@@ -19,7 +19,6 @@
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.log4j.Logger;
-
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 

diff --git a/gobblin-core/src/main/java/gobblin/writer/FsDataWriter.java b/gobblin-core/src/main/java/gobblin/writer/FsDataWriter.java
@@ -122,9 +122,14 @@ public FsDataWriter(State properties, String fileName, int numBranches, int bran
     this.stagingFileOutputStream = this.closer.register(this.fs.create(this.stagingFile, this.filePermission, true,
         this.bufferSize, this.replicationFactor, this.blockSize, null));
 
-    this.group = Optional.fromNullable(properties.getProp(ConfigurationKeys.WRITER_GROUP_NAME));
+    this.group =
+        Optional.fromNullable(properties.getProp(ForkOperatorUtils.getPropertyNameForBranch(
+            ConfigurationKeys.WRITER_GROUP_NAME, numBranches, branchId)));
+
     if (this.group.isPresent()) {
       HadoopUtils.setGroup(this.fs, this.stagingFile, this.group.get());
+    } else {
+      LOG.warn("No group found for " + this.stagingFile);
     }
 
     // Create the parent directory of the output file if it does not exist