[Bug-fix] Ignore modified table log when table has been dropped

fe/fe-core/src/main/java/org/apache/doris/alter/AlterHandler.java

@@ -427,7 +427,7 @@ public void handleFinishAlterTask(AlterReplicaTask task) throws MetaNotFoundExce
             if (replica == null) {
                 throw new MetaNotFoundException("replica " + task.getNewReplicaId() + " does not exist");
             }
-            
+
             LOG.info("before handle alter task tablet {}, replica: {}, task version: {}-{}",
                     task.getSignature(), replica, task.getVersion(), task.getVersionHash());
             boolean versionChanged = false;
@@ -445,7 +445,7 @@ public void handleFinishAlterTask(AlterReplicaTask task) throws MetaNotFoundExce
                         replica.getLastSuccessVersion(), replica.getLastSuccessVersionHash());
                 Catalog.getCurrentCatalog().getEditLog().logUpdateReplica(info);
             }
-            
+
             LOG.info("after handle alter task tablet: {}, replica: {}", task.getSignature(), replica);
         } finally {
             tbl.writeUnlock();

fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java

@@ -706,7 +706,7 @@ private void replayCreateJob(SchemaChangeJobV2 replayedJob) {
 
         this.watershedTxnId = replayedJob.watershedTxnId;
         jobState = JobState.WAITING_TXN;
-        LOG.info("replay pending schema change job: {}", jobId);
+        LOG.info("replay pending schema change job: {}, table id: {}", jobId, tableId);
     }
 
     /**
@@ -735,7 +735,7 @@ private void replayPendingJob(SchemaChangeJobV2 replayedJob) {
         // should still be in WAITING_TXN state, so that the alter tasks will be resend again
         this.jobState = JobState.WAITING_TXN;
         this.watershedTxnId = replayedJob.watershedTxnId;
-        LOG.info("replay waiting txn schema change job: {}", jobId);
+        LOG.info("replay waiting txn schema change job: {} table id: {}", jobId, tableId);
     }
 
     /**

fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java

@@ -147,6 +147,7 @@
 import org.apache.doris.load.LoadChecker;
 import org.apache.doris.load.LoadErrorHub;
 import org.apache.doris.load.LoadJob;
+import org.apache.doris.load.StreamLoadRecordMgr;
 import org.apache.doris.load.loadv2.LoadEtlChecker;
 import org.apache.doris.load.loadv2.LoadJobScheduler;
 import org.apache.doris.load.loadv2.LoadLoadingChecker;
@@ -155,7 +156,6 @@
 import org.apache.doris.load.routineload.RoutineLoadManager;
 import org.apache.doris.load.routineload.RoutineLoadScheduler;
 import org.apache.doris.load.routineload.RoutineLoadTaskScheduler;
-import org.apache.doris.load.StreamLoadRecordMgr;
 import org.apache.doris.master.Checkpoint;
 import org.apache.doris.master.MetaHelper;
 import org.apache.doris.meta.MetaContext;
@@ -225,14 +225,10 @@
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Queues;
 import com.google.common.collect.Sets;
-import com.sleepycat.je.rep.InsufficientLogException;
-import com.sleepycat.je.rep.NetworkRestore;
-import com.sleepycat.je.rep.NetworkRestoreConfig;
 
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.codehaus.jackson.map.ObjectMapper;
 
 import java.io.BufferedInputStream;
 import java.io.BufferedReader;
@@ -262,6 +258,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
+import com.sleepycat.je.rep.InsufficientLogException;
+import com.sleepycat.je.rep.NetworkRestore;
+import com.sleepycat.je.rep.NetworkRestoreConfig;
+import org.codehaus.jackson.map.ObjectMapper;
+
 public class Catalog {
     private static final Logger LOG = LogManager.getLogger(Catalog.class);
     // 0 ~ 9999 used for qe
@@ -4523,6 +4524,13 @@ private void unprotectUpdateReplica(ReplicaPersistInfo info) {
     public void replayAddReplica(ReplicaPersistInfo info) {
         Database db = getDb(info.getDbId());
         OlapTable olapTable = (OlapTable) db.getTable(info.getTableId());
+        if (olapTable == null) {
+            /**
+             * Same as replayUpdateReplica()
+             */
+            LOG.warn("Olap table is null when the add replica log is replayed, {}", info);
+            return;
+        }
         olapTable.writeLock();
         try {
             unprotectAddReplica(info);
@@ -4534,6 +4542,19 @@ public void replayAddReplica(ReplicaPersistInfo info) {
     public void replayUpdateReplica(ReplicaPersistInfo info) {
         Database db = getDb(info.getDbId());
         OlapTable olapTable = (OlapTable) db.getTable(info.getTableId());
+        if (olapTable == null) {
+            /**
+             * In the following cases, doris may record metadata modification information for a table that no longer exists.
+             * 1. Thread 1: get TableA object
+             * 2. Thread 2: lock db and drop table and record edit log of the dropped TableA
+             * 3. Thread 1: lock table, modify table and record edit log of the modified TableA
+             * **The modified edit log is after the dropped edit log**
+             * Because the table has been dropped, the olapTable in here is null when the modified edit log is replayed.
+             * So in this case, we will ignore the edit log of the modified table after the table is dropped.
+             */
+            LOG.warn("Olap table is null when the update replica log is replayed, {}", info);
+            return;
+        }
         olapTable.writeLock();
         try {
             unprotectUpdateReplica(info);
@@ -4554,6 +4575,13 @@ public void unprotectDeleteReplica(ReplicaPersistInfo info) {
     public void replayDeleteReplica(ReplicaPersistInfo info) {
         Database db = getDb(info.getDbId());
         OlapTable tbl = (OlapTable) db.getTable(info.getTableId());
+        if (tbl == null) {
+            /**
+             * Same as replayUpdateReplica()
+             */
+            LOG.warn("Olap table is null when the delete replica log is replayed, {}", info);
+            return;
+        }
         tbl.writeLock();
         try {
             unprotectDeleteReplica(info);

fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java

@@ -433,7 +433,7 @@ public Table getTable(long tableId) {
      */
     public Table getTableOrThrowException(long tableId, TableType tableType) throws MetaNotFoundException {
         Table table = idToTable.get(tableId);
-        if(table == null) {
+        if (table == null) {
             throw new MetaNotFoundException("unknown table, tableId=" + tableId);
         }
         if (table.getType() != tableType) {

fe/fe-core/src/main/java/org/apache/doris/persist/ReplicaPersistInfo.java

@@ -392,7 +392,6 @@ public String toString() {
         sb.append("table id: ").append(tableId);
         sb.append(" partition id: ").append(partitionId);
         sb.append(" index id: ").append(indexId);
-        sb.append(" index id: ").append(indexId);
         sb.append(" tablet id: ").append(tabletId);
         sb.append(" backend id: ").append(backendId);
         sb.append(" replica id: ").append(replicaId);