handle unable to load ft checkpoint

Summary:
- not being able to load ft checkpoint crashes the trainer
- avoid loading the ft checkpoint for now to continue training

Author: tushar00jain

torchtitan/components/checkpoint.py

@@ -667,20 +667,26 @@ def _ft_load(self) -> None:
         step = self._find_load_step(folder=self._ft_folder())
         if step == -1:
             return
-
-        begin = time.monotonic()
-        logger.info(f"Loading the FT checkpoint at step {step}.")
-        checkpoint_id = self._create_checkpoint_id(step, folder=self._ft_folder())
-        self.dcp_load(
-            self.ft_states,
-            checkpoint_id=checkpoint_id,
-            # FT checkpoints are always DCP because FT checkpoint currently only save/load dataloader.
-            from_hf=False,
-        )
-        GarbageCollection.collect("GC collection for checkpoint loading.")
-        logger.info(
-            f"Finished loading the ft checkpoint in {time.monotonic() - begin:.2f} seconds."
-        )
+        try:
+            begin = time.monotonic()
+            logger.info(f"Loading the FT checkpoint at step {step}.")
+            checkpoint_id = self._create_checkpoint_id(step, folder=self._ft_folder())
+            logger.info(f"Calling dcp_load for {checkpoint_id}")
+            self.dcp_load(
+                self.ft_states,
+                checkpoint_id=checkpoint_id,
+                # FT checkpoints are always DCP because FT checkpoint currently only save/load dataloader.
+                from_hf=False,
+            )
+            GarbageCollection.collect("GC collection for checkpoint loading.")
+            logger.info(
+                f"Finished loading the ft checkpoint in {time.monotonic() - begin:.2f} seconds."
+            )
+        except Exception as e:
+            # The checkpoint is corrupt. We'll replay all data here.
+            # TODO: We can try to load checkpoint from previous steps.
+            logger.error("Failed to load the FT checkpoint.")
+            return
 
     def _flattened_model_states_sd(
         self, state_dict: dict[str, Any] | None = None