HBASE-26579 Set storage policy of recovered edits when hbase.wal.stor…

…age.type is configured (#3948)

Signed-off-by: Duo Zhang <[email protected]>
Signed-off-by: Peter Somogyi <[email protected]>

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitUtil.java

@@ -150,6 +150,7 @@ public static void moveWAL(FileSystem fs, Path p, Path targetDir) throws IOExcep
    * named for the sequenceid in the passed <code>logEntry</code>: e.g.
    * /hbase/some_table/2323432434/recovered.edits/2332. This method also ensures existence of
    * RECOVERED_EDITS_DIR under the region creating it if necessary.
+   * And also set storage policy for RECOVERED_EDITS_DIR if WAL_STORAGE_POLICY is configured.
    * @param tableName the table name
    * @param encodedRegionName the encoded region name
    * @param seqId the sequence id which used to generate file name
@@ -184,6 +185,10 @@ static Path getRegionSplitEditsPath(TableName tableName, byte[] encodedRegionNam
 
     if (!walFS.exists(dir) && !walFS.mkdirs(dir)) {
       LOG.warn("mkdir failed on {}", dir);
+    } else {
+      String storagePolicy =
+        conf.get(HConstants.WAL_STORAGE_POLICY, HConstants.DEFAULT_WAL_STORAGE_POLICY);
+      CommonFSUtils.setStoragePolicy(walFS, dir, storagePolicy);
     }
     // Append fileBeingSplit to prevent name conflict since we may have duplicate wal entries now.
     // Append file name ends with RECOVERED_LOG_TMPFILE_SUFFIX to ensure
@@ -583,4 +588,4 @@ public static FileStatus[] getRecoveredHFiles(final FileSystem rootFS,
     Path dir = getRecoveredHFilesDir(regionDir, familyName);
     return CommonFSUtils.listStatus(rootFS, dir);
   }
-}
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplit.java

@@ -1174,6 +1174,18 @@ protected Writer createWriter(Path logfile)
     }
   }
 
+  @Test
+  public void testRecoveredEditsStoragePolicy() throws IOException {
+    conf.set(HConstants.WAL_STORAGE_POLICY, "ALL_SSD");
+    try {
+      Path path = createRecoveredEditsPathForRegion();
+      assertEquals("ALL_SSD", fs.getStoragePolicy(path.getParent()).getName());
+    } finally {
+      conf.unset(HConstants.WAL_STORAGE_POLICY);
+    }
+
+  }
+
   private Writer generateWALs(int leaveOpen) throws IOException {
     return generateWALs(NUM_WRITERS, ENTRIES, leaveOpen, 0);
   }