disable ls to read during restart

src/storage/slog_ckpt/ob_server_checkpoint_slog_handler.cpp

@@ -172,15 +172,17 @@ int ObServerCheckpointSlogHandler::try_write_checkpoint_for_compat()
     bool need_svr_ckpt = false;
     for (int64_t i = 0; OB_SUCC(ret) && i < tenant_metas.size(); ++i) {
       const ObTenantSuperBlock &super_block = tenant_metas.at(i).super_block_;
-      if (!super_block.is_old_version()) {
-        // nothing to do.
-      } else {
-        MTL_SWITCH(super_block.tenant_id_) {
-          if (OB_FAIL(MTL(ObTenantCheckpointSlogHandler*)->write_checkpoint(true/*is_force*/))) {
-            LOG_WARN("fail to write tenant slog checkpoint", K(ret));
-          } else {
-            need_svr_ckpt = true;
-          }
+      MTL_SWITCH(super_block.tenant_id_) {
+        ObTenantCheckpointSlogHandler *tenant_ckpt_handler = MTL(ObTenantCheckpointSlogHandler*);
+        if (!super_block.is_old_version()) {
+          // nothing to do.
+        } else if (OB_FAIL(tenant_ckpt_handler->write_checkpoint(true/*is_force*/))) {
+          LOG_WARN("fail to write tenant slog checkpoint", K(ret));
+        } else {
+          need_svr_ckpt = true;
+        }
+        if (OB_SUCC(ret) && OB_FAIL(tenant_ckpt_handler->enable_ls_read())) {
+          LOG_WARN("fail to enable ls to read", K(ret));
         }
       }
     }

src/storage/slog_ckpt/ob_tenant_checkpoint_slog_handler.cpp

@@ -661,6 +661,33 @@ int ObTenantCheckpointSlogHandler::replay_load_tablets()
   return ret;
 }
 
+int ObTenantCheckpointSlogHandler::enable_ls_read()
+{
+  int ret = OB_SUCCESS;
+  common::ObSharedGuard<ObLSIterator> ls_iter;
+  ObLS *ls = nullptr;
+
+  if (OB_FAIL(MTL(ObLSService *)->get_ls_iter(ls_iter, ObLSGetMod::STORAGE_MOD))) {
+    LOG_WARN("fail to get log stream iter", K(ret));
+  } else {
+    while (OB_SUCC(ret)) {
+      if (OB_FAIL(ls_iter->get_next(ls))) {
+        if (OB_ITER_END == ret) {
+          ret = OB_SUCCESS;
+          break;
+        } else {
+          LOG_WARN("fail to get next log stream", K(ret));
+        }
+      } else {
+        ObLSLockGuard lock_ls(ls);
+        ls->enable_to_read();
+      }
+    }
+  }
+
+  return ret;
+}
+
 int ObTenantCheckpointSlogHandler::report_slog(
     const ObTabletMapKey &tablet_key,
     const ObMetaDiskAddr &slog_addr)

src/storage/slog_ckpt/ob_tenant_checkpoint_slog_handler.h

@@ -100,6 +100,7 @@ public:
   void wait();
   void destroy();
 
+  int enable_ls_read();
   int report_slog(const ObTabletMapKey &tablet_key, const ObMetaDiskAddr &slog_addr);
   int check_slog(const ObTabletMapKey &tablet_key, bool &has_slog);
   int read_tablet_checkpoint_by_addr(

src/storage/tx_storage/ob_ls_service.cpp

@@ -838,6 +838,7 @@ int ObLSService::replay_create_ls_(const ObLSMeta &ls_meta)
                                    unused_allow_log_sync))) {
       LOG_WARN("enable ls palf failed", K(ret), K(ls_meta));
     } else {
+      ls->disable_to_read();
       LOG_INFO("success replay create ls", K(ret), K(ls_meta));
     }
   }