Merge pull request redpanda-data#23069 from WillemKauf/const_ref_exce…

…ption_cleanup

`misc`: `const&` exception catch clean-up

src/v/cloud_storage/cache_service.cc

@@ -172,7 +172,7 @@ cache::delete_file_and_empty_parents(const std::string_view& key) {
             vlog(cst_log.trace, "Removing {}", normal_path);
             co_await ss::remove_file(normal_path.native());
             deletions++;
-        } catch (std::filesystem::filesystem_error& e) {
+        } catch (const std::filesystem::filesystem_error& e) {
             if (e.code() == std::errc::directory_not_empty) {
                 // we stop when we find a non-empty directory
                 co_return deletions > 1;
@@ -230,7 +230,7 @@ ss::future<> cache::clean_up_at_start() {
                 co_await delete_file_and_empty_parents(filepath_to_remove);
                 deleted_bytes += file_item.size;
                 deleted_count++;
-            } catch (std::exception& e) {
+            } catch (const std::exception& e) {
                 vlog(
                   cst_log.error,
                   "Startup cache cleanup couldn't delete {}: {}.",
@@ -243,7 +243,7 @@ ss::future<> cache::clean_up_at_start() {
     for (const auto& path : empty_dirs) {
         try {
             co_await ss::remove_file(path);
-        } catch (std::exception& e) {
+        } catch (const std::exception& e) {
             // Leaving an empty dir will not prevent progress, so tolerate
             // errors on deletion (could be e.g. a permissions error)
             vlog(
@@ -670,7 +670,7 @@ cache::remove_segment_full(const file_list_item& file_stat) {
                 result.deleted_count += 1;
                 _current_cache_size -= sz;
                 _current_cache_objects -= 1;
-            } catch (std::filesystem::filesystem_error& e) {
+            } catch (const std::filesystem::filesystem_error& e) {
                 if (e.code() != std::errc::no_such_file_or_directory) {
                     throw;
                 }
@@ -686,7 +686,7 @@ cache::remove_segment_full(const file_list_item& file_stat) {
                 result.deleted_count += 1;
                 _current_cache_size -= sz;
                 _current_cache_objects -= 1;
-            } catch (std::filesystem::filesystem_error& e) {
+            } catch (const std::filesystem::filesystem_error& e) {
                 if (e.code() != std::errc::no_such_file_or_directory) {
                     throw;
                 }
@@ -899,7 +899,7 @@ ss::future<std::optional<uint64_t>> cache::access_time_tracker_size() const {
     auto path = _cache_dir / access_time_tracker_file_name;
     try {
         co_return static_cast<uint64_t>(co_await ss::file_size(path.string()));
-    } catch (std::filesystem::filesystem_error& e) {
+    } catch (const std::filesystem::filesystem_error& e) {
         if (e.code() == std::errc::no_such_file_or_directory) {
             co_return std::nullopt;
         } else {
@@ -1161,7 +1161,7 @@ ss::future<std::optional<cache_item>> cache::_get(std::filesystem::path key) {
                 });
             });
         }
-    } catch (std::filesystem::filesystem_error& e) {
+    } catch (const std::filesystem::filesystem_error& e) {
         if (e.code() == std::errc::no_such_file_or_directory) {
             probe.miss_get();
             co_return std::nullopt;
@@ -1248,7 +1248,7 @@ ss::future<> cache::put(
             tmp_cache_file = co_await ss::open_file_dma(
               (dir_path / tmp_filename).native(), flags);
             break;
-        } catch (std::filesystem::filesystem_error& e) {
+        } catch (const std::filesystem::filesystem_error& e) {
             if (e.code() == std::errc::no_such_file_or_directory) {
                 vlog(
                   cst_log.debug,
@@ -1271,7 +1271,7 @@ ss::future<> cache::put(
         co_await ss::copy(data, out)
           .then([&out]() { return out.flush(); })
           .finally([&out]() { return out.close(); });
-    } catch (std::filesystem::filesystem_error& e) {
+    } catch (const std::filesystem::filesystem_error& e) {
         // For ENOSPC errors, delay handling so that we can do a trim
         if (e.code() == std::errc::no_space_on_device) {
             disk_full_error = std::current_exception();
@@ -1363,7 +1363,7 @@ ss::future<> cache::_invalidate(const std::filesystem::path& key) {
         _current_cache_objects -= 1;
         probe.set_size(_current_cache_size);
         probe.set_num_files(_current_cache_objects);
-    } catch (std::filesystem::filesystem_error& e) {
+    } catch (const std::filesystem::filesystem_error& e) {
         if (e.code() == std::errc::no_such_file_or_directory) {
             vlog(
               cst_log.debug,

src/v/cloud_storage/recursive_directory_walker.cc

@@ -121,7 +121,7 @@ ss::future<> walker_process_directory(
         if (unlikely(!seen_dentries) && target != start_dir) {
             empty_dirs.push_back(target);
         }
-    } catch (std::filesystem::filesystem_error& e) {
+    } catch (const std::filesystem::filesystem_error& e) {
         if (e.code() == std::errc::no_such_file_or_directory) {
             // skip this directory, move to the next one
         } else {

src/v/cluster/archival/ntp_archiver_service.cc

@@ -2272,7 +2272,7 @@ ss::future<> ntp_archiver::housekeeping() {
         // Shutdown-type exceptions are thrown, to promptly drop out
         // of the upload loop.
         throw;
-    } catch (std::exception& e) {
+    } catch (const std::exception& e) {
         // Unexpected exceptions are logged, and suppressed: we do not
         // want to stop who upload loop because of issues in housekeeping
         vlog(_rtclog.warn, "Error occurred during housekeeping: {}", e.what());

src/v/cluster/cloud_metadata/offsets_recoverer.cc

@@ -139,7 +139,7 @@ ss::future<offsets_recovery_reply> offsets_recoverer::recover(
         }
         try {
             co_await offsets_lookup.run_lookups(std::move(ntps), retry_node);
-        } catch (ss::timed_out_error&) {
+        } catch (const ss::timed_out_error&) {
             reply.ec = cluster::errc::timeout;
             co_return reply;
         }

src/v/cluster/cloud_metadata/uploader.cc

@@ -333,7 +333,7 @@ ss::future<> uploader::upload_until_abort() {
             bool shutdown = false;
             try {
                 co_await _leader_cond.wait();
-            } catch (ss::broken_condition_variable&) {
+            } catch (const ss::broken_condition_variable&) {
                 shutdown = true;
             }
             if (shutdown || _as.abort_requested()) {

src/v/cluster/cluster_recovery_manager.cc

@@ -68,7 +68,7 @@ cluster_recovery_manager::sync_leader(ss::abort_source& as) {
     try {
         co_await _controller_stm.local().wait(
           committed_offset, model::timeout_clock::time_point::max(), as);
-    } catch (ss::abort_requested_exception&) {
+    } catch (const ss::abort_requested_exception&) {
         co_return std::nullopt;
     }
     if (!_raft0->is_leader() || synced_term != _raft0->term()) {

src/v/cluster/config_manager.cc

@@ -611,9 +611,9 @@ ss::future<> config_manager::reconcile_status() {
             co_await _reconcile_wait.wait(
               [this]() { return should_send_status(); });
         }
-    } catch (ss::condition_variable_timed_out&) {
+    } catch (const ss::condition_variable_timed_out&) {
         // Wait complete - proceed around next loop of do_until
-    } catch (ss::broken_condition_variable&) {
+    } catch (const ss::broken_condition_variable&) {
         // Shutting down - nextiteration will drop out
     }
 }
@@ -677,7 +677,7 @@ apply_local(cluster_config_delta_cmd_data const& data, bool silent) {
 
             bool changed = property.set_value(val);
             result.restart |= (property.needs_restart() && changed);
-        } catch (YAML::ParserException&) {
+        } catch (const YAML::ParserException&) {
             if (!silent) {
                 vlog(
                   clusterlog.warn,
@@ -687,7 +687,7 @@ apply_local(cluster_config_delta_cmd_data const& data, bool silent) {
             }
             result.invalid.push_back(u.key);
             continue;
-        } catch (YAML::BadConversion&) {
+        } catch (const YAML::BadConversion&) {
             if (!silent) {
                 vlog(
                   clusterlog.warn,
@@ -697,7 +697,7 @@ apply_local(cluster_config_delta_cmd_data const& data, bool silent) {
             }
             result.invalid.push_back(u.key);
             continue;
-        } catch (std::bad_alloc&) {
+        } catch (const std::bad_alloc&) {
             // Don't include bad_alloc in the catch-all below
             throw;
         } catch (...) {

src/v/cluster/drain_manager.cc

@@ -260,7 +260,7 @@ ss::future<> drain_manager::do_drain() {
         if (failed > 0 && dur < transfer_throttle && !_restore_requested) {
             try {
                 co_await ss::sleep_abortable(transfer_throttle - dur, _abort);
-            } catch (ss::sleep_aborted&) {
+            } catch (const ss::sleep_aborted&) {
             }
         }
     }

src/v/cluster/feature_manager.cc

@@ -263,7 +263,7 @@ ss::future<> feature_manager::maybe_log_license_check_info() {
     }
     try {
         co_await ss::sleep_abortable(license_check_retry, _as.local());
-    } catch (ss::sleep_aborted) {
+    } catch (const ss::sleep_aborted&) {
         // Shutting down - next iteration will drop out
     }
 }
@@ -306,11 +306,11 @@ ss::future<> feature_manager::maybe_update_feature_table() {
             // some feature activations to do.
             co_await _update_wait.wait([this]() { return updates_pending(); });
         }
-    } catch (ss::condition_variable_timed_out&) {
+    } catch (const ss::condition_variable_timed_out&) {
         // Wait complete - proceed around next loop of do_until
-    } catch (ss::broken_condition_variable&) {
+    } catch (const ss::broken_condition_variable&) {
         // Shutting down - nextiteration will drop out
-    } catch (ss::sleep_aborted&) {
+    } catch (const ss::sleep_aborted&) {
         // Shutting down - next iteration will drop out
     }
 }

src/v/cluster/leader_router.h

@@ -267,7 +267,7 @@ leader_router<req_t, resp_t, handler_t>::find_shard_and_process(
         while (!_as.abort_requested() && !shard && 0 < retries--) {
             try {
                 co_await sleep_abortable(delay_ms, _as);
-            } catch (ss::sleep_aborted&) {
+            } catch (const ss::sleep_aborted&) {
                 break;
             }
             shard = _shard_table.local().shard_for(ntp);

src/v/cluster/partition_probe.cc

@@ -221,7 +221,7 @@ void replicated_partition_probe::setup_public_metrics(const model::ntp& ntp) {
 
               try {
                   return _partition.log()->from_log_offset(log_offset);
-              } catch (std::runtime_error& e) {
+              } catch (const std::runtime_error& e) {
                   // Offset translation will throw if nothing was committed
                   // to the partition or if the offset is outside the
                   // translation range for any other reason.

src/v/features/feature_table.cc

@@ -518,7 +518,7 @@ feature_table::await_feature_then(feature f, std::function<void(void)> fn) {
     try {
         co_await await_feature(f);
         fn();
-    } catch (ss::abort_requested_exception&) {
+    } catch (const ss::abort_requested_exception&) {
         // Shutting down
     } catch (...) {
         // Should never happen, abort is the only exception that await_feature

src/v/kafka/protocol/types.h

@@ -213,7 +213,7 @@ inline kafka::leader_epoch leader_epoch_from_term(model::term_id term) {
     try {
         return kafka::leader_epoch(
           boost::numeric_cast<kafka::leader_epoch::type>(term()));
-    } catch (boost::bad_numeric_cast&) {
+    } catch (const boost::bad_numeric_cast&) {
         return kafka::invalid_leader_epoch;
     }
 }

src/v/kafka/server/handlers/configs/config_utils.h

@@ -600,7 +600,7 @@ inline void parse_and_set_bool(
             property.value = string_switch<bool>(*value)
                                .match("true", true)
                                .match("false", false);
-        } catch (std::runtime_error) {
+        } catch (const std::runtime_error&) {
             // Our callers expect this exception type on malformed values
             throw boost::bad_lexical_cast();
         }

src/v/net/connection.cc

@@ -79,7 +79,7 @@ bool is_reconnect_error(const std::system_error& e) {
 std::optional<ss::sstring> is_disconnect_exception(std::exception_ptr e) {
     try {
         rethrow_exception(e);
-    } catch (std::system_error& e) {
+    } catch (const std::system_error& e) {
         if (is_reconnect_error(e)) {
             return e.code().message();
         }

src/v/net/server.cc

@@ -115,7 +115,7 @@ bool is_gate_closed_exception(std::exception_ptr e) {
         if (e) {
             rethrow_exception(e);
         }
-    } catch (ss::gate_closed_exception&) {
+    } catch (const ss::gate_closed_exception&) {
         return true;
     } catch (...) {
         return false;

src/v/raft/consensus.cc

@@ -1578,7 +1578,7 @@ consensus::do_start(std::optional<xshard_transfer_state> xst_state) {
           _term,
           _configuration_manager.get_latest());
 
-    } catch (ss::broken_semaphore&) {
+    } catch (const ss::broken_semaphore&) {
     }
 }
 

src/v/redpanda/admin/server.cc

@@ -650,7 +650,7 @@ void admin_server::log_exception(
 
     try {
         std::rethrow_exception(eptr);
-    } catch (ss::httpd::base_exception& ex) {
+    } catch (const ss::httpd::base_exception& ex) {
         const auto status = static_cast<http_status_ut>(ex.status());
         if (ex.status() == http_status::internal_server_error) {
             vlog(adminlog.error, "{}", log_ex(status));
@@ -1274,7 +1274,7 @@ void admin_server::register_config_routes() {
               try {
                   config::shard_local_cfg().to_json_single_key(
                     writer, config::redact_secrets::yes, key_str);
-              } catch (std::out_of_range&) {
+              } catch (const std::out_of_range&) {
                   throw ss::httpd::bad_param_exception(
                     fmt::format("Unknown property {{{}}}", key_str));
               }
@@ -1346,7 +1346,7 @@ void admin_server::register_config_routes() {
           ss::log_level cur_level;
           try {
               cur_level = ss::global_logger_registry().get_logger_level(name);
-          } catch (std::out_of_range&) {
+          } catch (const std::out_of_range&) {
               throw ss::httpd::bad_param_exception(fmt::format(
                 "Cannot set log level: unknown logger {{{}}}", name));
           }
@@ -1387,7 +1387,7 @@ void admin_server::register_config_routes() {
           ss::log_level cur_level;
           try {
               cur_level = ss::global_logger_registry().get_logger_level(name);
-          } catch (std::out_of_range&) {
+          } catch (const std::out_of_range&) {
               throw ss::httpd::bad_param_exception(fmt::format(
                 "Cannot set log level: unknown logger {{{}}}", name));
           }

src/v/redpanda/admin/server.h

@@ -154,7 +154,7 @@ class admin_server {
         auto auth_state = [this, &req]() -> request_auth_result {
             try {
                 return _auth.authenticate(req);
-            } catch (unauthorized_user_exception& e) {
+            } catch (const unauthorized_user_exception& e) {
                 audit_authn_failure(req, e.get_username(), e.what());
                 throw;
             } catch (const ss::httpd::base_exception& e) {

src/v/redpanda/admin/util.cc

@@ -18,7 +18,7 @@ void apply_validator(
   json::validator& validator, json::Document::ValueType const& doc) {
     try {
         json::validate(validator, doc);
-    } catch (json::json_validation_error& err) {
+    } catch (const json::json_validation_error& err) {
         throw ss::httpd::bad_request_exception(fmt::format(
           "JSON request body does not conform to schema: {}", err.what()));
     }

src/v/redpanda/monitor_unsafe_log_flag.cc

@@ -80,7 +80,7 @@ ss::future<> monitor_unsafe_log_flag::maybe_log_flag_nag() {
 
     try {
         co_await ss::sleep_abortable(nag_check_retry, _as);
-    } catch (ss::sleep_aborted&) {
+    } catch (const ss::sleep_aborted&) {
         // Shutting down - next iteration will drop out
     }
 }

src/v/storage/api.cc

@@ -51,7 +51,7 @@ ss::future<bool> api::wait_for_cluster_uuid() {
         vassert(
           _cluster_uuid.has_value(), "Expected cluster UUID after waiting");
         co_return true;
-    } catch (ss::broken_condition_variable&) {
+    } catch (const ss::broken_condition_variable&) {
         vlog(stlog.info, "Stopped waiting for cluster UUID");
     }
     co_return false;

src/v/storage/segment_utils.cc

@@ -78,10 +78,10 @@ maybe_disable_cow(const std::filesystem::path& path, ss::file& file) {
                 vlog(stlog.trace, "Disabled COW on BTRFS segment {}", path);
             }
         }
-    } catch (std::filesystem::filesystem_error& e) {
+    } catch (const std::filesystem::filesystem_error& e) {
         // This can happen if e.g. our file open races with an unlink
         vlog(stlog.info, "Filesystem error disabling COW on {}: {}", path, e);
-    } catch (std::system_error& e) {
+    } catch (const std::system_error& e) {
         // Non-fatal, user will just get degraded behaviour
         // when btrfs tries to COW on a journal.
         vlog(stlog.info, "System error disabling COW on {}: {}", path, e);
@@ -1017,7 +1017,7 @@ ss::future<std::vector<ss::rwlock::holder>> write_lock_segments(
             }
             held = co_await ss::when_all_succeed(held_f.begin(), held_f.end());
             break;
-        } catch (ss::semaphore_timed_out&) {
+        } catch (const ss::semaphore_timed_out&) {
             held.clear();
         }
         if (retries == 0) {

src/v/test_utils/async.h

@@ -33,7 +33,7 @@ using namespace std::chrono_literals;
     do {                                                                       \
         try {                                                                  \
             co_await tests::cooperative_spin_wait_with_timeout(__VA_ARGS__);   \
-        } catch (ss::timed_out_error&) {                                       \
+        } catch (const ss::timed_out_error&) {                                 \
             RPTEST_FAIL_CORO(                                                  \
               ssx::sformat("Timed out at {}:{}", __FILE__, __LINE__));         \
         }                                                                      \
@@ -44,7 +44,7 @@ using namespace std::chrono_literals;
     do {                                                                       \
         try {                                                                  \
             tests::cooperative_spin_wait_with_timeout(__VA_ARGS__).get();      \
-        } catch (ss::timed_out_error&) {                                       \
+        } catch (const ss::timed_out_error&) {                                 \
             RPTEST_FAIL(                                                       \
               ssx::sformat("Timed out at {}:{}", __FILE__, __LINE__));         \
         }                                                                      \