[HUDI-9737] Add checksum checks while recovering from backup file for HoodieTableConfig (#13740)

Author: Rajeev-01

hudi-common/src/main/java/org/apache/hudi/common/util/ConfigUtils.java

@@ -36,6 +36,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -685,11 +686,37 @@ public static TypedProperties fetchConfigs(
 
   public static void recoverIfNeeded(HoodieStorage storage, StoragePath cfgPath,
                                      StoragePath backupCfgPath) throws IOException {
+    boolean needCopy = false;
     if (!storage.exists(cfgPath)) {
-      // copy over from backup
-      try (InputStream in = storage.open(backupCfgPath);
-           OutputStream out = storage.create(cfgPath, false)) {
-        FileIOUtils.copy(in, out);
+      needCopy = true;
+    } else {
+      TypedProperties props = new TypedProperties();
+      try (InputStream in = storage.open(cfgPath)) {
+        props.load(in);
+        if (!props.containsKey(TABLE_CHECKSUM.key()) || !HoodieTableConfig.validateChecksum(props)) {
+          // the cfg file is invalid
+          storage.deleteFile(cfgPath);
+          needCopy = true;
+        }
+      }
+    }
+    if (needCopy && storage.exists(backupCfgPath)) {
+      byte[] bytes = FileIOUtils.readAsByteArray(storage.open(backupCfgPath));
+      // check whether existing backup file is valid or not
+      try (InputStream backupStream = new ByteArrayInputStream(bytes)) {
+        TypedProperties backupProps = new TypedProperties();
+        backupProps.load(backupStream);
+        if (!backupProps.containsKey(TABLE_CHECKSUM.key()) || !HoodieTableConfig.validateChecksum(backupProps)) {
+          // need to delete the backup as anyway reads will also fail
+          // subsequent writes will recover and update
+          storage.deleteFile(backupCfgPath);
+          LOG.warn("Invalid properties file {}: {}", backupCfgPath, backupProps);
+          throw new IOException("Corrupted backup file");
+        }
+        // copy over from backup
+        try (OutputStream out = storage.create(cfgPath, false)) {
+          out.write(bytes);
+        }
       }
     }
     // regardless, we don't need the backup anymore.

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java

@@ -240,6 +240,32 @@ void testUpdateRecovery(boolean shouldPropsFileExist) throws IOException {
     assertEquals(7, config.getProps().size());
   }
 
+  @Test
+  void testUpdateRecoveryWithInvalidProps() throws IOException {
+    // 1. Actual props is invalid, populate it from backup if they are valid
+    HoodieTableConfig config = new HoodieTableConfig(storage, metaPath);
+    try (OutputStream out = storage.create(backupCfgPath)) {
+      config.getProps().store(out, "");
+    }
+    storage.deleteFile(cfgPath);
+    // create the empty file
+    storage.create(cfgPath);
+    recoverIfNeeded(storage, cfgPath, backupCfgPath);
+    assertTrue(storage.exists(cfgPath));
+    assertFalse(storage.exists(backupCfgPath));
+    config = new HoodieTableConfig(storage, metaPath);
+    assertEquals(7, config.getProps().size());
+
+    // 2. Backup properties file is also invalid
+    storage.deleteFile(cfgPath);
+    // create the empty files
+    storage.create(cfgPath);
+    storage.create(backupCfgPath);
+    assertThrows(IOException.class, () -> recoverIfNeeded(storage, cfgPath, backupCfgPath));
+    assertFalse(storage.exists(backupCfgPath));
+    assertFalse(storage.exists(cfgPath));
+  }
+
   @Test
   void testReadRetry() throws IOException {
     // When both the hoodie.properties and hoodie.properties.backup do not exist then the read fails