[BugFix] Fix miss update job state when replaying restore log may cause restored table lost after FE restart (backport #59056) (backport #68396) (#68412)

Signed-off-by: srlch <111035020+srlch@users.noreply.github.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Co-authored-by: srlch <111035020+srlch@users.noreply.github.com>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/backup/BackupHandler.java

@@ -641,10 +641,6 @@ public void replayAddJob(AbstractJob job) {
             // for example: In restore job, PENDING will transfer to SNAPSHOTING, not DOWNLOAD.
             job.replayRun();
         }
-        if (isJobExpired(job, System.currentTimeMillis())) {
-            LOG.warn("skip expired job {}", job);
-            return;
-        }
         dbIdToBackupOrRestoreJob.put(job.getDbId(), job);
         mvRestoreContext.addIntoMvBaseTableBackupInfo(job);
     }

fe/fe-core/src/test/java/com/starrocks/backup/RestoreJobTest.java

@@ -927,4 +927,31 @@ public synchronized QueryStatement init() throws UserException {
         Assert.assertEquals(Status.OK, job.getStatus());
         Assert.assertEquals(RestoreJobState.FINISHED, job.getState());
     }
-}
+
+    @Test
+    public void testReplayAddExpiredJob() {
+        RestoreJob job1 = new RestoreJob(label, "2018-01-01 01:01:01", db.getId() + 999, db.getFullName() + "xxx",
+                new BackupJobInfo(), false, 3, 100000,
+                globalStateMgr, repo.getId(), backupMeta, new MvRestoreContext());
+
+        BackupJobInfo jobInfo = new BackupJobInfo();
+        BackupTableInfo tblInfo = new BackupTableInfo();
+        tblInfo.id = CatalogMocker.TEST_TBL2_ID;
+        tblInfo.name = CatalogMocker.TEST_TBL2_NAME;
+        jobInfo.tables.put(tblInfo.name, tblInfo);
+        RestoreJob job3 = new RestoreJob(label, "2018-01-01 01:01:01", db.getId() + 999, db.getFullName() + "xxx",
+                jobInfo, false, 3, 100000,
+                globalStateMgr, repo.getId(), backupMeta, new MvRestoreContext());
+
+        BackupHandler localBackupHandler = new BackupHandler();
+        job1.setState(RestoreJob.RestoreJobState.PENDING);
+        localBackupHandler.replayAddJob(job1);
+        Assert.assertTrue(localBackupHandler.getJob(db.getId() + 999).isPending());
+        int oldVal = Config.history_job_keep_max_second;
+        Config.history_job_keep_max_second = 0;
+        job3.setState(RestoreJob.RestoreJobState.FINISHED);
+        localBackupHandler.replayAddJob(job3);
+        Config.history_job_keep_max_second = oldVal;
+        Assert.assertTrue(localBackupHandler.getJob(db.getId() + 999).isDone());
+    }
+}