TEZ-4413: Fix permission may not be set after crash.

Author: ayushtkn

tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java

@@ -18,6 +18,7 @@
 
 package org.apache.tez.dag.history.logging.proto;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -71,10 +72,19 @@ public DatePartitionedLogger(Parser<T> parser, Path baseDir, Configuration conf,
 
   private void createDirIfNotExists(Path path) throws IOException {
     FileSystem fileSystem = path.getFileSystem(conf);
+    FileStatus fileStatus = null;
     try {
-      if (!fileSystem.exists(path)) {
+      fileStatus = fileSystem.getFileStatus(path);
+    } catch (FileNotFoundException fnf) {
+      // ignore
+    }
+    try {
+      if (fileStatus == null) {
         fileSystem.mkdirs(path);
         fileSystem.setPermission(path, DIR_PERMISSION);
+      } else if (!fileStatus.getPermission().equals(DIR_PERMISSION)) {
+        LOG.info("Permission on path {} is {}, setting it to {}", path, fileStatus.getPermission(), DIR_PERMISSION);
+        fileSystem.setPermission(path, DIR_PERMISSION);
       }
     } catch (IOException e) {
       // Ignore this exception, if there is a problem it'll fail when trying to read or write.

tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestProtoHistoryLoggingService.java

@@ -32,7 +32,10 @@
 
 import com.google.protobuf.CodedInputStream;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.util.Time;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -267,6 +270,23 @@ public void testServiceSplitEvents() throws Exception {
     scanner.close();
   }
 
+  @Test
+  public void testDirPermissions() throws IOException {
+    Path basePath = new Path(tempFolder.newFolder().getAbsolutePath());
+    Configuration conf = new Configuration();
+    FileSystem fs = basePath.getFileSystem(conf);
+    FsPermission expectedPermissions = FsPermission.createImmutable((short) 01777);
+
+    // Check the directory already exists and doesn't have the expected permissions.
+    Assert.assertTrue(fs.exists(basePath));
+    Assert.assertNotEquals(expectedPermissions, fs.getFileStatus(basePath).getPermission());
+
+    new DatePartitionedLogger<>(HistoryEventProto.PARSER, basePath, conf, new FixedClock(Time.now()));
+
+    // Check the permissions they should be same as the expected permissions
+    Assert.assertEquals(expectedPermissions, fs.getFileStatus(basePath).getPermission());
+  }
+
   private List<DAGHistoryEvent> makeHistoryEvents(TezDAGID dagId,
       ProtoHistoryLoggingService service) {
     List<DAGHistoryEvent> historyEvents = new ArrayList<>();