[BugFix] fix pk concurrent apply issue (backport #51225)

be/src/storage/base_tablet.h

@@ -115,9 +115,9 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
         return flag;
     }
 
-protected:
     virtual void on_shutdown() {}
 
+protected:
     void _gen_tablet_path();
 
     TabletState _state;

be/src/storage/tablet.h

@@ -331,7 +331,6 @@ class Tablet : public BaseTablet {
     // set true when start to drop tablet. only set in `TabletManager::drop_tablet` right now
     void set_is_dropping(bool is_dropping) { _is_dropping = is_dropping; }
 
-protected:
     void on_shutdown() override;
 
 private:

be/src/storage/tablet_manager.cpp

@@ -437,6 +437,16 @@ Status TabletManager::drop_tablet(TTabletId tablet_id, TabletDropFlag flag) {
         _add_shutdown_tablet_unlocked(tablet_id, std::move(drop_info));
     } else {
         DCHECK_EQ(kKeepMetaAndFiles, flag);
+        {
+            // If the tablet is the primary key table, there might still be ongoing apply tasks.
+            // We should stop the background apply tasks before deleting it; otherwise, if a new tablet is created,
+            // there could be a scenario where apply tasks are executed simultaneously.
+            // e.g.
+            // 1. drop and clone a new tablet with the same tablet_id.
+            // 2. compact rocksdb meta and reload tablet again.
+            std::unique_lock l(dropped_tablet->get_header_lock());
+            dropped_tablet->on_shutdown();
+        }
     }
     // erase tablet from tablet map
     {

be/src/storage/tablet_updates.h

@@ -494,6 +494,7 @@ class TabletUpdates {
             _last_compaction_time_ms = UnixMillis();
         }
     }
+    bool is_apply_stop() { return _apply_stopped.load(); }
 
     bool compaction_running() { return _compaction_running; }
 

be/test/storage/tablet_updates_test.cpp

@@ -3731,4 +3731,11 @@ TEST_F(TabletUpdatesTest, test_get_compaction_status) {
     test_horizontal_compaction(false, true);
 }
 
+TEST_F(TabletUpdatesTest, test_drop_tablet_with_keep_meta_and_files) {
+    _tablet = create_tablet(rand(), rand());
+    ASSERT_FALSE(_tablet->updates()->is_apply_stop());
+    StorageEngine::instance()->tablet_manager()->drop_tablet(_tablet->tablet_id(), kKeepMetaAndFiles);
+    ASSERT_TRUE(_tablet->updates()->is_apply_stop());
+}
+
 } // namespace starrocks