github-actions[bot] commented on code in PR #67464:
URL: https://github.com/apache/doris/pull/67464#discussion_r3922178494
##########
be/src/io/cache/peer_file_cache_reader.cpp:
##########
@@ -212,11 +260,15 @@ Status PeerFileCacheReader::fetch_blocks(const
std::vector<FileBlockSPtr>& block
}
}
std::string brpc_addr = get_host_port(realhost, port);
+ if (!peer_connection_circuit_allows(brpc_addr)) {
+ return Status::RpcError<false>("Peer connection circuit is open for
{}", brpc_addr);
+ }
Review Comment:
`peer_connection_circuit_allows()` returns a generic THRIFT_RPC_ERROR here,
but both peer callers treat every such status as an actual candidate RPC
failure (`cached_remote_file_reader.cpp:435-439` and `:643-650`). Once this
address circuit is open, three reads for a tablet can evict its candidate
without issuing any RPC; if the peer recovers after the 30-second cooldown,
that tablet has no candidate left to probe until FE refresh. Please return a
distinct circuit-open classification (or have callers skip
candidate-failure/all-miss accounting) so deliberate breaker rejections do not
poison per-tablet recovery state.
##########
be/src/io/cache/peer_file_cache_reader.cpp:
##########
@@ -45,6 +50,49 @@ namespace doris::io {
namespace {
+struct PeerConnectionHealth {
+ int32_t consecutive_failures = 0;
+ std::chrono::steady_clock::time_point circuit_open_until;
+ bool probe_in_flight = false;
+};
+
+// Entries are removed after a successful connection. Add expiry cleanup if
permanently
+// unavailable peer addresses accumulate.
+bthread::Mutex peer_connection_health_mutex;
+std::unordered_map<std::string, PeerConnectionHealth> peer_connection_health;
+
+bool peer_connection_circuit_allows(const std::string& address) {
+ std::unique_lock<bthread::Mutex> lock(peer_connection_health_mutex);
+ auto it = peer_connection_health.find(address);
+ if (it == peer_connection_health.end() ||
+ it->second.circuit_open_until == std::chrono::steady_clock::time_point
{}) {
+ return true;
+ }
+ auto& health = it->second;
+ if (std::chrono::steady_clock::now() < health.circuit_open_until ||
health.probe_in_flight) {
+ return false;
+ }
+ health.probe_in_flight = true;
+ return true;
+}
+
+void record_peer_connection_failure(const std::string& address) {
+ std::unique_lock<bthread::Mutex> lock(peer_connection_health_mutex);
+ auto& health = peer_connection_health[address];
+ health.probe_in_flight = false;
+ ++health.consecutive_failures;
+ if (health.consecutive_failures >= std::max(1,
config::cache_peer_read_failure_threshold)) {
+ health.circuit_open_until =
Review Comment:
These configs are mutable (`DEFINE_mInt32`), but the circuit state is not
reconciled when they change. For example, after 3 failures open a circuit,
raising the threshold to 10 causes the first failed half-open probe to reach
count 4 (<10) and leave the old deadline expired; every later call is then
allowed to probe immediately until count 10, bypassing the promised cooldown.
Changing the cooldown similarly leaves already-computed deadlines stale. Please
normalize the state on config changes or always re-open after a failed
half-open probe using the current cooldown.
##########
be/src/io/cache/peer_file_cache_reader.cpp:
##########
@@ -45,6 +50,49 @@ namespace doris::io {
namespace {
+struct PeerConnectionHealth {
+ int32_t consecutive_failures = 0;
+ std::chrono::steady_clock::time_point circuit_open_until;
+ bool probe_in_flight = false;
+};
+
+// Entries are removed after a successful connection. Add expiry cleanup if
permanently
+// unavailable peer addresses accumulate.
+bthread::Mutex peer_connection_health_mutex;
+std::unordered_map<std::string, PeerConnectionHealth> peer_connection_health;
+
Review Comment:
The health map is process-wide and its only erase path is a later successful
RPC. Every permanently dead or retired host:port inserted by
`record_peer_connection_failure()` therefore keeps its string and state until
BE shutdown; candidate expiry does not touch this map. In a long-lived cloud BE
with endpoint churn this grows without bound (and is not MemTracker-accounted).
Please add TTL/periodic cleanup or a bounded eviction policy.
##########
be/src/io/cache/peer_file_cache_reader.cpp:
##########
@@ -244,8 +296,10 @@ Status PeerFileCacheReader::fetch_blocks(const
std::vector<FileBlockSPtr>& block
peer_cache_reader_read_counter << 1;
brpc_stub->fetch_peer_data(&cntl, &req, &resp, nullptr);
if (cntl.Failed()) {
+ record_peer_connection_failure(brpc_addr);
return Status::RpcError<false>(cntl.ErrorText());
}
+ record_peer_connection_success(brpc_addr);
if (resp.has_status()) {
Review Comment:
`record_peer_connection_success()` runs before `resp.status` is checked. A
peer that accepts the RPC but returns INTERNAL_ERROR (the existing
`MockPeerCacheService::fail_status` path) therefore erases the address failure
count, so repeated non-OK peer RPCs never open this BE-wide breaker despite the
PR contract covering ‘connection or RPC failures.’ Please record failure for
non-OK statuses (while treating an expected NOT_FOUND cache miss as healthy if
intended) and clear health only after a fully successful read.
--
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]
