Copilot commented on code in PR #65241: URL: https://github.com/apache/doris/pull/65241#discussion_r3526858286
########## regression-test/suites/load_p0/test_stream_load_close_wait_hang.groovy: ########## @@ -0,0 +1,140 @@ +// 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.apache.doris.regression.suite.ClusterOptions + +suite("test_stream_load_close_wait_hang", "docker") { + def options = new ClusterOptions() + options.feNum = 1 + options.beNum = 3 + options.cloudMode = false + + // Enable single replica load in FE BE to trigger write_single_replica mode + options.feConfigs += [ + 'enable_single_replica_load=true' + ] + + options.beConfigs += [ + 'enable_debug_points=true', + 'enable_single_replica_load=true' + ] + + docker(options) { + def tableName = "test_close_wait_hang_table" + + // Create table with 3 replicas and RANDOM distribution + // - replication_num=3: triggers write_single_replica mode (1 master + 2 slaves) + // - RANDOM distribution: required by load_to_single_tablet parameter + sql """ DROP TABLE IF EXISTS ${tableName} """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + name VARCHAR(100) + ) + DUPLICATE KEY(id) + DISTRIBUTED BY RANDOM BUCKETS 1 + PROPERTIES ( + "replication_num" = "3" + ) + """ + + // Prepare test data - write to local file + def dataFile = new File("/tmp/test_stream_load_close_wait_hang_data.csv") Review Comment: The test writes to a fixed filename under /tmp. Regression suites can run in parallel, and this can cause data races/cross-test interference (file content overwritten or deleted by another run). Use a unique temp file instead. ########## regression-test/suites/load_p0/test_stream_load_close_wait_hang.groovy: ########## @@ -0,0 +1,140 @@ +// 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.apache.doris.regression.suite.ClusterOptions + +suite("test_stream_load_close_wait_hang", "docker") { + def options = new ClusterOptions() + options.feNum = 1 + options.beNum = 3 + options.cloudMode = false + + // Enable single replica load in FE BE to trigger write_single_replica mode + options.feConfigs += [ + 'enable_single_replica_load=true' + ] + + options.beConfigs += [ + 'enable_debug_points=true', + 'enable_single_replica_load=true' + ] + + docker(options) { + def tableName = "test_close_wait_hang_table" + + // Create table with 3 replicas and RANDOM distribution + // - replication_num=3: triggers write_single_replica mode (1 master + 2 slaves) + // - RANDOM distribution: required by load_to_single_tablet parameter + sql """ DROP TABLE IF EXISTS ${tableName} """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + name VARCHAR(100) + ) + DUPLICATE KEY(id) + DISTRIBUTED BY RANDOM BUCKETS 1 + PROPERTIES ( + "replication_num" = "3" + ) + """ + + // Prepare test data - write to local file + def dataFile = new File("/tmp/test_stream_load_close_wait_hang_data.csv") + dataFile.withWriter { writer -> + for (int i = 1; i <= 5000; i++) { + writer.writeLine("${i}\ttest${i}") + } + } + log.info("Test data written to ${dataFile.absolutePath}") + + // Enable debug point to simulate "slave node not found" scenario + // Effect: Sets node = nullptr to trigger the check + // Expected behavior: Code should call cancel() and clear_in_flight() before returning + try { + GetDebugPoint().enableDebugPointForAllBEs("VNodeChannel.try_send_pending_block.slave_node_not_found") + log.info("Debug point enabled: simulating slave node not found scenario") + } catch (Exception e) { + log.warn("Failed to enable debug point: ${e.message}") + } Review Comment: The test swallows failures when enabling the debug point. If enableDebugPointForAllBEs() fails (e.g., debug points disabled/misconfigured), the test may proceed without exercising the hang scenario and still pass. ########## regression-test/suites/load_p0/test_stream_load_close_wait_hang.groovy: ########## @@ -0,0 +1,140 @@ +// 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.apache.doris.regression.suite.ClusterOptions + +suite("test_stream_load_close_wait_hang", "docker") { + def options = new ClusterOptions() + options.feNum = 1 + options.beNum = 3 + options.cloudMode = false + + // Enable single replica load in FE BE to trigger write_single_replica mode + options.feConfigs += [ + 'enable_single_replica_load=true' + ] + + options.beConfigs += [ + 'enable_debug_points=true', + 'enable_single_replica_load=true' + ] + + docker(options) { + def tableName = "test_close_wait_hang_table" + + // Create table with 3 replicas and RANDOM distribution + // - replication_num=3: triggers write_single_replica mode (1 master + 2 slaves) + // - RANDOM distribution: required by load_to_single_tablet parameter + sql """ DROP TABLE IF EXISTS ${tableName} """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + name VARCHAR(100) + ) + DUPLICATE KEY(id) + DISTRIBUTED BY RANDOM BUCKETS 1 + PROPERTIES ( + "replication_num" = "3" + ) + """ + + // Prepare test data - write to local file + def dataFile = new File("/tmp/test_stream_load_close_wait_hang_data.csv") + dataFile.withWriter { writer -> + for (int i = 1; i <= 5000; i++) { + writer.writeLine("${i}\ttest${i}") + } + } + log.info("Test data written to ${dataFile.absolutePath}") + + // Enable debug point to simulate "slave node not found" scenario + // Effect: Sets node = nullptr to trigger the check + // Expected behavior: Code should call cancel() and clear_in_flight() before returning + try { + GetDebugPoint().enableDebugPointForAllBEs("VNodeChannel.try_send_pending_block.slave_node_not_found") + log.info("Debug point enabled: simulating slave node not found scenario") + } catch (Exception e) { + log.warn("Failed to enable debug point: ${e.message}") + } + + def testStartTime = System.currentTimeMillis() + def timed_out = false + def load_exception = null + + // Execute stream load in separate thread to detect hangs + // If cancel() is NOT called when slave node is missing, close_wait() will hang forever + def load_thread = Thread.start { + try { + streamLoad { + table "${tableName}" + + set 'column_separator', '\t' + set 'format', 'csv' + + set 'load_to_single_tablet', 'true' + + time 60000 + + file dataFile.absolutePath + + check { result, exception, startTime, endTime -> + if (exception != null) { + load_exception = exception + log.error("Stream load failed with exception: ${exception.message}") + } + } Review Comment: The streamLoad check callback does not assert any expected outcome. Since providing a check closure bypasses StreamLoadAction’s default success/failure validation, this test can pass even when the load unexpectedly succeeds or fails for the wrong reason. Assert an expected failure status (and optionally the error message). ########## be/src/exec/sink/writer/vtablet_writer.cpp: ########## @@ -1008,7 +1008,16 @@ void VNodeChannel::try_send_pending_block(RuntimeState* state) { PSlaveTabletNodes slave_tablet_nodes; for (auto node_id : _slave_tablet_node.second) { const auto* node = _parent->_nodes_info->find_node(node_id); + DBUG_EXECUTE_IF("VNodeChannel.try_send_pending_block.slave_node_not_found", { + LOG(WARNING) << "trigger " + "VNodeChannel.try_send_pending_block.slave_node_not_found " + "debug point will set node to nullptr"; + node = nullptr; + }); if (node == nullptr) { + LOG(WARNING) << "slave node not found, node_id=" << node_id; + cancel(fmt::format("slave node not found, node_id={}", node_id)); + _send_block_callback->clear_in_flight(); Review Comment: This new cancel() reason drops the channel context, while other cancel paths in the same method include channel_info() for easier diagnosis. Consider including channel_info() in the message for consistency and debugging. ########## be/src/exec/sink/writer/vtablet_writer.cpp: ########## @@ -1052,6 +1061,7 @@ void VNodeChannel::try_send_pending_block(RuntimeState* state) { if (!status.ok()) { LOG(WARNING) << "failed to get ip from host " << _node_info.host << ": " << status.to_string(); + cancel(fmt::format("failed to get ip from host {}", _node_info.host)); _send_block_callback->clear_in_flight(); return; Review Comment: This cancel() reason omits both channel context and the underlying DNS error details. Other cancel paths in this method include channel_info() and the original error string; adding them will make failures actionable in logs. ########## regression-test/suites/load_p0/test_stream_load_close_wait_hang.groovy: ########## @@ -0,0 +1,140 @@ +// 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.apache.doris.regression.suite.ClusterOptions + +suite("test_stream_load_close_wait_hang", "docker") { + def options = new ClusterOptions() + options.feNum = 1 + options.beNum = 3 + options.cloudMode = false + + // Enable single replica load in FE BE to trigger write_single_replica mode + options.feConfigs += [ + 'enable_single_replica_load=true' + ] + + options.beConfigs += [ + 'enable_debug_points=true', + 'enable_single_replica_load=true' + ] + + docker(options) { + def tableName = "test_close_wait_hang_table" + + // Create table with 3 replicas and RANDOM distribution + // - replication_num=3: triggers write_single_replica mode (1 master + 2 slaves) + // - RANDOM distribution: required by load_to_single_tablet parameter + sql """ DROP TABLE IF EXISTS ${tableName} """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + name VARCHAR(100) + ) + DUPLICATE KEY(id) + DISTRIBUTED BY RANDOM BUCKETS 1 + PROPERTIES ( + "replication_num" = "3" + ) + """ + + // Prepare test data - write to local file + def dataFile = new File("/tmp/test_stream_load_close_wait_hang_data.csv") + dataFile.withWriter { writer -> + for (int i = 1; i <= 5000; i++) { + writer.writeLine("${i}\ttest${i}") + } + } + log.info("Test data written to ${dataFile.absolutePath}") + + // Enable debug point to simulate "slave node not found" scenario + // Effect: Sets node = nullptr to trigger the check + // Expected behavior: Code should call cancel() and clear_in_flight() before returning + try { + GetDebugPoint().enableDebugPointForAllBEs("VNodeChannel.try_send_pending_block.slave_node_not_found") + log.info("Debug point enabled: simulating slave node not found scenario") + } catch (Exception e) { + log.warn("Failed to enable debug point: ${e.message}") + } + + def testStartTime = System.currentTimeMillis() + def timed_out = false + def load_exception = null + + // Execute stream load in separate thread to detect hangs + // If cancel() is NOT called when slave node is missing, close_wait() will hang forever + def load_thread = Thread.start { + try { + streamLoad { + table "${tableName}" + + set 'column_separator', '\t' + set 'format', 'csv' + + set 'load_to_single_tablet', 'true' + + time 60000 + + file dataFile.absolutePath + + check { result, exception, startTime, endTime -> + if (exception != null) { + load_exception = exception + log.error("Stream load failed with exception: ${exception.message}") + } + } + } + } catch (Exception e) { + load_exception = e + log.error("Stream load threw exception: ${e.message}") + } + } + + // Check if stream load completes within timeout + // Expected: Should fail quickly (within 65s) because cancel() is called + // Bug symptom: Hangs forever if cancel() is not called + load_thread.join(65000) + + if (load_thread.isAlive()) { + timed_out = true + log.error("Stream load thread hung for too many seconds!") + + load_thread.interrupt() + load_thread.join(5000) + } + + def elapsed = System.currentTimeMillis() - testStartTime + log.info("Stream load completed in ${elapsed}ms") + + try { + GetDebugPoint().disableDebugPointForAllBEs("VNodeChannel.try_send_pending_block.slave_node_not_found") + } catch (Exception e) { + log.warn("Failed to disable debug point: ${e.message}") + } + + if (timed_out) { + throw new Exception("Stream load time out") + } Review Comment: load_exception is captured from the load thread but never acted on. If streamLoad throws (network error, unexpected server error, etc.), the test may still pass as long as it doesn't time out. -- 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]
