github-actions[bot] commented on code in PR #63722: URL: https://github.com/apache/doris/pull/63722#discussion_r3489862256
########## regression-test/suites/load_p0/stream_load/test_group_commit_stream_load_high_concurrency_async.groovy: ########## @@ -0,0 +1,138 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.awaitility.Awaitility + +import static java.util.concurrent.TimeUnit.SECONDS + +import java.util.Collections +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger + +suite("test_group_commit_stream_load_high_concurrency_async", "p0") { + def tableName = "test_group_commit_stream_load_high_concurrency_async" + int concurrentClients = 20 + int loadsPerClient = 20 + int expectedRows = concurrentClients * loadsPerClient + def errors = Collections.synchronizedList(new ArrayList<String>()) + def stopRequested = new AtomicBoolean(false) + def successLoads = new AtomicInteger(0) + def getProperty = { property, userName -> + def result = sql_return_maparray """SHOW PROPERTY FOR '${userName}'""" + result.find { + it.Key == property as String + } + } + def originMaxUserConnections = getProperty("max_user_connections", "root").Value as long + + def waitRowCount = { expected -> + Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({ + def result = sql "select count(*) from ${tableName}" + logger.info("table: ${tableName}, rowCount: ${result}, expected: ${expected}") + return result[0][0] == expected + }) + } + + def checkStreamLoadResult = { loadId, result, exception -> + if (exception != null) { + stopRequested.set(true) + errors.add("load ${loadId} exception: ${exception.getMessage()}") + return + } + def json = parseJson(result) + if (!"success".equalsIgnoreCase(json.Status?.toString())) { + stopRequested.set(true) + errors.add("load ${loadId} status=${json.Status}, msg=${json.Message}") + return + } + if (json.GroupCommit != true) { + stopRequested.set(true) + errors.add("load ${loadId} is not group commit: ${result}") + return + } + if (json.NumberTotalRows != 1 || json.NumberLoadedRows != 1 || + json.NumberFilteredRows != 0 || json.NumberUnselectedRows != 0) { + stopRequested.set(true) + errors.add("load ${loadId} unexpected counters: ${result}") + return + } + successLoads.incrementAndGet() + } + + try { + sql """SET PROPERTY FOR 'root' 'max_user_connections' = '1024'""" Review Comment: This test changes the cluster-wide `root` `max_user_connections` property from a normal `p0` suite and restores it only in `finally`. That leaks global auth/session state into any other normal suite running at the same time, and it can temporarily lower a valid preconfigured root limit because FE accepts values up to 10000. The test currently starts only 20 clients, below the default per-user limit of 100, so this mutation does not appear necessary. Please avoid changing `root` here (or use a dedicated temporary user), or make the suite non-concurrent and set at least `max(origin, required)` before restoring. ########## be/src/load/group_commit/group_commit_mgr.cpp: ########## @@ -277,31 +285,131 @@ Status GroupCommitTable::get_first_block_load_queue( return Status::OK(); } create_plan_dep->block(); + _create_plan_be_exe_version = be_exe_version; + if (_create_plan_deps.empty()) { + _create_plan_start_time_ms = MonotonicMillis(); + } _create_plan_deps.emplace(load_id, std::make_tuple(create_plan_dep, put_block_dep, base_schema_version, index_size)); - if (!_is_creating_plan_fragment) { - _is_creating_plan_fragment = true; - RETURN_IF_ERROR( - _thread_pool->submit_func([&, be_exe_version, mem_tracker, dep = create_plan_dep] { - Defer defer {[&, dep = dep]() { - std::unique_lock l(_lock); - for (auto it : _create_plan_deps) { - std::get<0>(it.second)->set_ready(); + [[maybe_unused]] auto submit_st = _submit_create_group_commit_load(); + return try_to_get_matched_queue(); +} + +Status GroupCommitTable::submit_create_group_commit_load() { + std::unique_lock l(_lock); Review Comment: The timeout check is placed after the `_is_creating_plan_fragment` fast return, so it is never evaluated while the create-plan task is queued or still running. That means loads already blocked on `create_plan_dep` can wait past `group_commit_create_plan_timeout_ms`; for example, if this mutable config is set below the 10s `streamLoadPut` RPC timeout in `_create_group_commit_load()`, the pending loads will not be released at the configured deadline. Please check and release timed-out `_create_plan_deps` before this early return, or add a timed wakeup path that enforces the timeout while the create-plan task is in flight. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
