[fix](group commit)Fix the issue of duplicate addition of wal path when encouter exception

be/src/common/config.cpp

@@ -1112,7 +1112,7 @@ DEFINE_Int32(grace_shutdown_wait_seconds, "120");
 DEFINE_Int16(bitmap_serialize_version, "1");
 
 // group commit insert config
-DEFINE_String(group_commit_replay_wal_dir, "./wal");
+DEFINE_String(group_commit_wal_path, "./wal");
 DEFINE_Int32(group_commit_replay_wal_retry_num, "10");
 DEFINE_Int32(group_commit_replay_wal_retry_interval_seconds, "5");
 DEFINE_Int32(group_commit_relay_wal_threads, "10");

be/src/common/config.h

@@ -1185,7 +1185,7 @@ DECLARE_Int32(grace_shutdown_wait_seconds);
 DECLARE_Int16(bitmap_serialize_version);
 
 // group commit insert config
-DECLARE_String(group_commit_replay_wal_dir);
+DECLARE_String(group_commit_wal_path);
 DECLARE_Int32(group_commit_replay_wal_retry_num);
 DECLARE_Int32(group_commit_replay_wal_retry_interval_seconds);
 DECLARE_mInt32(group_commit_relay_wal_threads);

be/src/olap/wal_manager.cpp

@@ -39,7 +39,7 @@
 namespace doris {
 WalManager::WalManager(ExecEnv* exec_env, const std::string& wal_dir_list)
         : _exec_env(exec_env), _stop_background_threads_latch(1), _stop(false) {
-    doris::vectorized::WalReader::string_split(wal_dir_list, ",", _wal_dirs);
+    doris::vectorized::WalReader::string_split(wal_dir_list, ";", _wal_dirs);
     _all_wal_disk_bytes = std::make_shared<std::atomic_size_t>(0);
     _cv = std::make_shared<std::condition_variable>();
     static_cast<void>(ThreadPoolBuilder("GroupCommitReplayWalThreadPool")

be/src/olap/wal_table.cpp

@@ -70,7 +70,8 @@ Status WalTable::replay_wals() {
         for (auto& [wal, info] : _replay_wal_map) {
             auto& [retry_num, start_ts, replaying] = info;
             if (replaying) {
-                continue;
+                LOG(INFO) << wal << " is replaying, skip this round";
+                return Status::OK();
             }
             if (retry_num >= config::group_commit_replay_wal_retry_num) {
                 LOG(WARNING) << "All replay wal failed, db=" << _db_id << ", table=" << _table_id

be/src/runtime/exec_env_init.cpp

@@ -222,7 +222,7 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
     _memtable_memory_limiter = std::make_unique<MemTableMemoryLimiter>();
     _load_stream_stub_pool = std::make_unique<stream_load::LoadStreamStubPool>();
     _delta_writer_v2_pool = std::make_unique<vectorized::DeltaWriterV2Pool>();
-    _wal_manager = WalManager::create_shared(this, config::group_commit_replay_wal_dir);
+    _wal_manager = WalManager::create_shared(this, config::group_commit_wal_path);
 
     _backend_client_cache->init_metrics("backend");
     _frontend_client_cache->init_metrics("frontend");

be/src/runtime/group_commit_mgr.cpp

@@ -333,7 +333,6 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
                      << ", executor status=" << status.to_string()
                      << ", request commit status=" << st.to_string();
         if (!prepare_failed) {
-            RETURN_IF_ERROR(_exec_env->wal_mgr()->add_wal_path(_db_id, table_id, txn_id, label));
             std::string wal_path;
             RETURN_IF_ERROR(_exec_env->wal_mgr()->get_wal_path(txn_id, wal_path));
             RETURN_IF_ERROR(_exec_env->wal_mgr()->add_recover_wal(
@@ -350,7 +349,6 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
     // TODO handle execute and commit error
     if (!prepare_failed && !result_status.ok() &&
         !(result_status.is<ErrorCode::PUBLISH_TIMEOUT>())) {
-        RETURN_IF_ERROR(_exec_env->wal_mgr()->add_wal_path(_db_id, table_id, txn_id, label));
         std::string wal_path;
         RETURN_IF_ERROR(_exec_env->wal_mgr()->get_wal_path(txn_id, wal_path));
         RETURN_IF_ERROR(_exec_env->wal_mgr()->add_recover_wal(std::to_string(db_id),

be/src/vec/exec/format/wal/wal_reader.cpp

@@ -67,7 +67,7 @@ Status WalReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
     }
     block->swap(dst_block);
     *read_rows = block->rows();
-    LOG(INFO) << "read block rows:" << *read_rows;
+    VLOG_DEBUG << "read block rows:" << *read_rows;
     return Status::OK();
 }
 

fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java

@@ -1174,7 +1174,7 @@ public static int analyzeGroupCommitIntervalMs(Map<String, String> properties) t
             try {
                 groupCommitIntervalMs = Integer.parseInt(groupIntervalCommitMsStr);
             } catch (Exception e) {
-                throw new AnalysisException("schema version format error");
+                throw new AnalysisException("parse group_commit_interval_ms format error");
             }
 
             properties.remove(PROPERTIES_GROUP_COMMIT_INTERVAL_MS);