fix

Author: mymeiyi

be/src/runtime/group_commit_mgr.cpp

@@ -260,6 +260,13 @@ Status GroupCommitTable::get_first_block_load_queue(
     DCHECK(table_id == _table_id);
     std::unique_lock l(_lock);
     auto try_to_get_matched_queue = [&]() -> Status {
+        DBUG_EXECUTE_IF("GroupCommitTable.get_first_block_load_queue.can_not_get_a_block_queue", {
+            if (dp->param<int64_t>("table_id", -1) == table_id) {
+                LOG(INFO) << "debug promise set: can not get a block queue for table=" << table_id;
+                return Status::InternalError<false>("can not get a block queue for table_id: " +
+                                                    std::to_string(table_id));
+            }
+        };);
         for (const auto& [_, inner_block_queue] : _load_block_queues) {
             if (inner_block_queue->contain_load_id(load_id)) {
                 load_block_queue = inner_block_queue;

fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java

@@ -72,10 +72,13 @@
 import org.apache.thrift.TException;
 import org.apache.thrift.TSerializer;
 
+import java.net.HttpURLConnection;
+import java.net.URL;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -328,4 +331,34 @@ private void setReturnInfo(ConnectContext ctx, boolean reuse, PGroupCommitInsert
         // update it, so that user can get loaded rows in fe.audit.log
         ctx.updateReturnRows((int) loadedRows);
     }
+
+    public static void removeBeDebugPoint() {
+        try {
+            for (Entry<Long, Backend> entry : Env.getCurrentSystemInfo().getAllBackendsByAllCluster().entrySet()) {
+                Backend backend = entry.getValue();
+                if (!backend.isAlive()) {
+                    continue;
+                }
+                String host = backend.getHost();
+                int port = backend.getBePort();
+
+                String urlStr = String.format(
+                        "http://%s:%d/api/debug_point/remove/GroupCommitTable.get_first_block_load_queue.can_not_get_a_block_queue",
+                        host, port);
+                URL url = new URL(urlStr);
+                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+                try {
+                    conn.setRequestMethod("POST");
+                    int responseCode = conn.getResponseCode();
+                    LOG.warn("remove debug point for backend {}:{}, response code: {}", host, port, responseCode);
+                } catch (Exception e) {
+                    LOG.warn("failed to remove debug point for backend {}:{}", host, port, e);
+                } finally {
+                    conn.disconnect();
+                }
+            }
+        } catch (Exception e) {
+            LOG.warn("failed to remove debug point for backends", e);
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -2443,23 +2443,33 @@ private void handleInsertStmt() throws Exception {
                 PGroupCommitInsertResponse response = groupCommitPlanner.executeGroupCommitInsert(context, rows);
                 TStatusCode code = TStatusCode.findByValue(response.getStatus().getStatusCode());
                 ProtocolStringList errorMsgsList = response.getStatus().getErrorMsgsList();
-                if (code == TStatusCode.DATA_QUALITY_ERROR && !errorMsgsList.isEmpty() && errorMsgsList.get(0)
-                        .contains("schema version not match")) {
+                boolean schemaVersionNotMatch = code == TStatusCode.DATA_QUALITY_ERROR && !errorMsgsList.isEmpty()
+                        && errorMsgsList.get(0).contains("schema version not match");
+                boolean canNotGetBlockQueue = code != TStatusCode.OK && !errorMsgsList.isEmpty() && errorMsgsList.get(0)
+                        .contains("can not get a block queue");
+                if (schemaVersionNotMatch || canNotGetBlockQueue) {
                     LOG.info("group commit insert failed. stmt: {}, query_id: {}, db_id: {}, table_id: {}"
                                     + ", schema version: {}, backend_id: {}, status: {}, retry: {}",
                             insertStmt.getOrigStmt().originStmt, DebugUtil.printId(context.queryId()), dbId, tableId,
                             nativeInsertStmt.getBaseSchemaVersion(), groupCommitPlanner.getBackendId(),
                             response.getStatus(), i);
                     if (i < maxRetry) {
-                        List<TableIf> tables = Lists.newArrayList(insertStmt.getTargetTable());
-                        tables.sort((Comparator.comparing(TableIf::getId)));
-                        MetaLockUtils.readLockTables(tables);
-                        try {
-                            insertStmt.reset();
-                            analyzer = new Analyzer(context.getEnv(), context);
-                            analyzeAndGenerateQueryPlan(context.getSessionVariable().toThrift());
-                        } finally {
-                            MetaLockUtils.readUnlockTables(tables);
+                        if (schemaVersionNotMatch) {
+                            List<TableIf> tables = Lists.newArrayList(insertStmt.getTargetTable());
+                            tables.sort((Comparator.comparing(TableIf::getId)));
+                            MetaLockUtils.readLockTables(tables);
+                            try {
+                                insertStmt.reset();
+                                analyzer = new Analyzer(context.getEnv(), context);
+                                analyzeAndGenerateQueryPlan(context.getSessionVariable().toThrift());
+                            } finally {
+                                MetaLockUtils.readUnlockTables(tables);
+                            }
+                        }
+                        long debugTableId = DebugPointUtil.getDebugParamOrDefault("GroupCommitInsert.retry", 0);
+                        if (debugTableId == tableId) {
+                            // Thread.sleep(10000);
+                            GroupCommitPlanner.removeBeDebugPoint();
                         }
                         continue;
                     } else {