[GOBBLIN-2095] Set owner and group when pre-creating directories (#3981)

* Set owner and group when pre-creating directories
* Fix tests

gobblin-data-management/src/main/java/org/apache/gobblin/util/commit/CreateAndSetDirectoryPermissionCommitStep.java

@@ -71,6 +71,8 @@ public void execute() throws IOException {
         if (!fs.exists(path)) {
           log.info("Creating path {} with permission {}", path, entry.getValue().getFsPermission());
           fs.mkdirs(path, entry.getValue().getFsPermission());
+          // Set owner and group since created directories will have owner set to the job runner (instead of the data owner)
+          fs.setOwner(path, entry.getValue().getOwner(), entry.getValue().getGroup());
         } else {
           log.info("Setting permission {} on existing path {}", entry.getValue().getFsPermission(), path);
           fs.setPermission(path, entry.getValue().getFsPermission());

gobblin-data-management/src/test/java/org/apache/gobblin/util/commit/CreateAndSetDirectoryPermissionCommitStepTest.java

@@ -32,6 +32,8 @@
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
+import com.google.common.io.Files;
+
 import org.apache.gobblin.data.management.copy.OwnerAndPermission;
 
 
@@ -40,7 +42,7 @@
  */
 @Test(groups = { "gobblin.commit" })
 public class CreateAndSetDirectoryPermissionCommitStepTest {
-  private static final String ROOT_DIR = "set-permission-commit-step-test";
+  private static final Path ROOT_DIR = new Path(Files.createTempDir().getPath(),"set-permission-commit-step-test");
 
   private FileSystem fs;
   private CreateAndSetDirectoryPermissionCommitStep step;
@@ -50,11 +52,14 @@ public class CreateAndSetDirectoryPermissionCommitStepTest {
   @BeforeClass
   public void setUp() throws IOException {
     this.fs = FileSystem.getLocal(new Configuration());
-    this.fs.delete(new Path(ROOT_DIR), true);
+    this.fs.delete(ROOT_DIR, true);
+    this.fs.mkdirs(ROOT_DIR);
 
     dir1 = new Path(ROOT_DIR, "dir1");
+    String owner = this.fs.getFileStatus(ROOT_DIR).getOwner();
+    String group = this.fs.getFileStatus(ROOT_DIR).getGroup();
 
-    OwnerAndPermission ownerAndPermission = new OwnerAndPermission("owner", "group", permission);
+    OwnerAndPermission ownerAndPermission = new OwnerAndPermission(owner, group, permission);
     Map<String, OwnerAndPermission> pathAndPermissions = new HashMap<>();
     pathAndPermissions.put(dir1.toString(), ownerAndPermission);
 
@@ -63,13 +68,15 @@ public void setUp() throws IOException {
 
   @AfterClass
   public void tearDown() throws IOException {
-    this.fs.delete(new Path(ROOT_DIR), true);
+    this.fs.delete(ROOT_DIR, true);
   }
 
   @Test
   public void testExecute() throws IOException {
     this.step.execute();
     Assert.assertEquals(this.fs.exists(dir1), true);
     Assert.assertEquals(this.fs.getFileStatus(dir1).getPermission(), permission);
+    Assert.assertEquals(this.fs.getFileStatus(dir1).getOwner(), this.fs.getFileStatus(ROOT_DIR).getOwner());
+    Assert.assertEquals(this.fs.getFileStatus(dir1).getGroup(), this.fs.getFileStatus(ROOT_DIR).getGroup());
   }
 }