[CORE-8161] storage: add min.cleanable.dirty.ratio, schedule compaction by dirty_ratio

src/v/cloud_storage/tests/topic_manifest_test.cc

@@ -487,6 +487,7 @@ SEASTAR_THREAD_TEST_CASE(test_topic_manifest_serde_feature_table) {
       std::nullopt,
       std::nullopt,
       std::nullopt,
+      tristate<double>{},
     };
 
     auto random_initial_revision_id

src/v/cluster/metadata_cache.cc

@@ -329,6 +329,11 @@ metadata_cache::get_default_iceberg_target_lag_ms() const {
     return config::shard_local_cfg().iceberg_target_lag_ms();
 }
 
+std::optional<double>
+metadata_cache::get_default_min_cleanable_dirty_ratio() const {
+    return config::shard_local_cfg().min_cleanable_dirty_ratio();
+}
+
 topic_properties metadata_cache::get_default_properties() const {
     topic_properties tp;
     tp.compression = {get_default_compression()};
@@ -348,6 +353,8 @@ topic_properties metadata_cache::get_default_properties() const {
       get_default_retention_local_target_ms()};
     tp.delete_retention_ms = tristate<std::chrono::milliseconds>{
       get_default_delete_retention_ms()};
+    tp.min_cleanable_dirty_ratio = tristate<double>{
+      get_default_min_cleanable_dirty_ratio()};
 
     return tp;
 }

src/v/cluster/metadata_cache.h

@@ -205,6 +205,7 @@ class metadata_cache {
     std::optional<std::chrono::milliseconds>
     get_default_delete_retention_ms() const;
     std::chrono::milliseconds get_default_iceberg_target_lag_ms() const;
+    std::optional<double> get_default_min_cleanable_dirty_ratio() const;
 
     topic_properties get_default_properties() const;
     std::optional<partition_assignment>

src/v/cluster/topic_configuration.cc

@@ -59,6 +59,7 @@ storage::ntp_config topic_configuration::make_ntp_config(
             .iceberg_mode = properties.iceberg_mode,
             .cloud_topic_enabled = properties.cloud_topic_enabled,
             .tombstone_retention_ms = properties.delete_retention_ms,
+            .min_cleanable_dirty_ratio = properties.min_cleanable_dirty_ratio,
           });
     }
     return {

src/v/cluster/topic_properties.cc

@@ -45,7 +45,8 @@ std::ostream& operator<<(std::ostream& o, const topic_properties& properties) {
       "iceberg_delete: {}, "
       "iceberg_partition_spec: {}, "
       "iceberg_invalid_record_action: {}, "
-      "iceberg_target_lag_ms: {}",
+      "iceberg_target_lag_ms: {}, "
+      "min_cleanable_dirty_ratio: {}",
       properties.compression,
       properties.cleanup_policy_bitflags,
       properties.compaction_strategy,
@@ -85,7 +86,8 @@ std::ostream& operator<<(std::ostream& o, const topic_properties& properties) {
       properties.iceberg_delete,
       properties.iceberg_partition_spec,
       properties.iceberg_invalid_record_action,
-      properties.iceberg_target_lag_ms);
+      properties.iceberg_target_lag_ms,
+      properties.min_cleanable_dirty_ratio);
 
     if (config::shard_local_cfg().development_enable_cloud_topics()) {
         fmt::print(
@@ -133,7 +135,8 @@ bool topic_properties::has_overrides() const {
         || leaders_preference.has_value() || delete_retention_ms.is_engaged()
         || iceberg_delete.has_value() || iceberg_partition_spec.has_value()
         || iceberg_invalid_record_action.has_value()
-        || iceberg_target_lag_ms.has_value();
+        || iceberg_target_lag_ms.has_value()
+        || min_cleanable_dirty_ratio.is_engaged();
 
     if (config::shard_local_cfg().development_enable_cloud_topics()) {
         return overrides
@@ -176,6 +179,7 @@ topic_properties::get_ntp_cfg_overrides() const {
     ret.iceberg_mode = iceberg_mode;
     ret.cloud_topic_enabled = cloud_topic_enabled;
     ret.tombstone_retention_ms = delete_retention_ms;
+    ret.min_cleanable_dirty_ratio = min_cleanable_dirty_ratio;
     return ret;
 }
 
@@ -272,6 +276,7 @@ adl<cluster::topic_properties>::from(iobuf_parser& parser) {
       std::nullopt,
       std::nullopt,
       std::nullopt,
+      tristate<double>{std::nullopt},
     };
 }
 

src/v/cluster/topic_properties.h

@@ -33,7 +33,7 @@ namespace cluster {
  */
 struct topic_properties
   : serde::
-      envelope<topic_properties, serde::version<11>, serde::compat_version<0>> {
+      envelope<topic_properties, serde::version<12>, serde::compat_version<0>> {
     topic_properties() noexcept = default;
     topic_properties(
       std::optional<model::compression> compression,
@@ -80,7 +80,8 @@ struct topic_properties
       std::optional<ss::sstring> iceberg_partition_spec,
       std::optional<model::iceberg_invalid_record_action>
         iceberg_invalid_record_action,
-      std::optional<std::chrono::milliseconds> iceberg_target_lag_ms)
+      std::optional<std::chrono::milliseconds> iceberg_target_lag_ms,
+      tristate<double> min_cleanable_dirty_ratio)
       : compression(compression)
       , cleanup_policy_bitflags(cleanup_policy_bitflags)
       , compaction_strategy(compaction_strategy)
@@ -125,7 +126,8 @@ struct topic_properties
       , iceberg_delete(iceberg_delete)
       , iceberg_partition_spec(std::move(iceberg_partition_spec))
       , iceberg_invalid_record_action(iceberg_invalid_record_action)
-      , iceberg_target_lag_ms(iceberg_target_lag_ms) {}
+      , iceberg_target_lag_ms(iceberg_target_lag_ms)
+      , min_cleanable_dirty_ratio(min_cleanable_dirty_ratio) {}
 
     std::optional<model::compression> compression;
     std::optional<model::cleanup_policy_bitflags> cleanup_policy_bitflags;
@@ -210,6 +212,8 @@ struct topic_properties
 
     std::optional<std::chrono::milliseconds> iceberg_target_lag_ms{};
 
+    tristate<double> min_cleanable_dirty_ratio{std::nullopt};
+
     bool is_compacted() const;
     bool has_overrides() const;
     bool requires_remote_erase() const;
@@ -259,7 +263,8 @@ struct topic_properties
           iceberg_delete,
           iceberg_partition_spec,
           iceberg_invalid_record_action,
-          iceberg_target_lag_ms);
+          iceberg_target_lag_ms,
+          min_cleanable_dirty_ratio);
     }
 
     friend bool operator==(const topic_properties&, const topic_properties&)

src/v/cluster/topic_table.cc

@@ -1110,6 +1110,10 @@ topic_properties topic_table::update_topic_properties(
     incremental_update(
       updated_properties.iceberg_target_lag_ms,
       overrides.iceberg_target_lag_ms);
+    incremental_update(
+      updated_properties.min_cleanable_dirty_ratio,
+      overrides.min_cleanable_dirty_ratio);
+
     return updated_properties;
 }
 

src/v/cluster/types.h

@@ -568,7 +568,7 @@ struct property_update<tristate<T>>
 struct incremental_topic_updates
   : serde::envelope<
       incremental_topic_updates,
-      serde::version<8>,
+      serde::version<9>,
       serde::compat_version<0>> {
     static constexpr int8_t version_with_data_policy = -1;
     static constexpr int8_t version_with_shadow_indexing = -3;
@@ -644,6 +644,7 @@ struct incremental_topic_updates
     property_update<std::optional<ss::sstring>> iceberg_partition_spec;
     property_update<std::optional<model::iceberg_invalid_record_action>>
       iceberg_invalid_record_action;
+    property_update<tristate<double>> min_cleanable_dirty_ratio;
 
     property_update<std::optional<std::chrono::milliseconds>>
       iceberg_target_lag_ms;
@@ -689,7 +690,8 @@ struct incremental_topic_updates
           iceberg_delete,
           iceberg_partition_spec,
           iceberg_invalid_record_action,
-          iceberg_target_lag_ms);
+          iceberg_target_lag_ms,
+          min_cleanable_dirty_ratio);
     }
 
     friend std::ostream&

src/v/compat/cluster_compat.h

@@ -363,6 +363,7 @@ struct compat_check<cluster::topic_properties> {
           "iceberg_invalid_record_action",
           obj.iceberg_invalid_record_action);
         json_write(iceberg_target_lag_ms);
+        json_write(min_cleanable_dirty_ratio);
     }
 
     static cluster::topic_properties from_json(json::Value& rd) {
@@ -401,6 +402,7 @@ struct compat_check<cluster::topic_properties> {
         json_read(remote_topic_namespace_override);
         json_read(iceberg_invalid_record_action);
         json_read(iceberg_target_lag_ms);
+        json_read(min_cleanable_dirty_ratio);
         return obj;
     }
 
@@ -434,6 +436,7 @@ struct compat_check<cluster::topic_properties> {
         obj.remote_topic_namespace_override = std::nullopt;
         obj.iceberg_invalid_record_action = std::nullopt;
         obj.iceberg_target_lag_ms = std::nullopt;
+        obj.min_cleanable_dirty_ratio = tristate<double>{std::nullopt};
 
         if (reply != obj) {
             throw compat_error(fmt::format(
@@ -520,6 +523,8 @@ struct compat_check<cluster::topic_configuration> {
         obj.properties.mpx_virtual_cluster_id = std::nullopt;
 
         obj.properties.iceberg_invalid_record_action = std::nullopt;
+        obj.properties.min_cleanable_dirty_ratio = tristate<double>{
+          std::nullopt};
 
         obj.properties.iceberg_target_lag_ms = std::nullopt;
 

src/v/compat/cluster_generator.h

@@ -663,7 +663,9 @@ struct instance_generator<cluster::topic_properties> {
                 {model::iceberg_invalid_record_action::drop,
                  model::iceberg_invalid_record_action::dlq_table});
           }),
-          tests::random_optional([] { return tests::random_duration_ms(); })};
+          tests::random_optional([] { return tests::random_duration_ms(); }),
+          tristate<double>{disable_tristate},
+        };
     }
 
     static std::vector<cluster::topic_properties> limits() { return {}; }

src/v/compat/cluster_json.h

@@ -631,6 +631,7 @@ inline void rjson_serialize(
     write_exceptional_member_type(
       w, "iceberg_invalid_record_action", tps.iceberg_invalid_record_action);
     write_member(w, "iceberg_target_lag_ms", tps.iceberg_target_lag_ms);
+    write_member(w, "min_cleanable_dirty_ratio", tps.min_cleanable_dirty_ratio);
     w.EndObject();
 }
 
@@ -706,6 +707,7 @@ inline void read_value(const json::Value& rd, cluster::topic_properties& obj) {
     read_member(
       rd, "iceberg_invalid_record_action", obj.iceberg_invalid_record_action);
     read_member(rd, "iceberg_target_lag_ms", obj.iceberg_target_lag_ms);
+    read_member(rd, "min_cleanable_dirty_ratio", obj.min_cleanable_dirty_ratio);
 }
 
 inline void rjson_serialize(

src/v/compat/json.h

@@ -35,6 +35,10 @@ inline const char* to_str(const rapidjson::Type t) {
     return str[t];
 }
 
+inline void read_value(const json::Value& v, double& target) {
+    target = v.GetDouble();
+}
+
 inline void read_value(const json::Value& v, int64_t& target) {
     target = v.GetInt64();
 }

src/v/config/configuration.cc

@@ -993,6 +993,19 @@ configuration::configuration()
       {.needs_restart = needs_restart::no, .visibility = visibility::user},
       std::nullopt,
       validate_tombstone_retention_ms)
+  , min_cleanable_dirty_ratio(
+      *this,
+      "min_cleanable_dirty_ratio",
+      "The minimum ratio between the number of bytes in \"dirty\" segments and "
+      "the total number of bytes in closed segments that must be reached "
+      "before a partition's log is eligible for compaction in a compact topic. "
+      "The topic property `min.cleanable.dirty.ratio` overrides the value of "
+      "`min_cleanable_dirty_ratio` at the topic level.",
+      {.needs_restart = needs_restart::no,
+       .example = "0.5",
+       .visibility = visibility::user},
+      std::nullopt,
+      {.min = 0.0, .max = 1.0})
   , log_disable_housekeeping_for_tests(
       *this,
       "log_disable_housekeeping_for_tests",

src/v/config/configuration.h

@@ -252,6 +252,8 @@ struct configuration final : public config_store {
     property<std::chrono::milliseconds> log_compaction_interval_ms;
     // same as delete.retention.ms in kafka
     property<std::optional<std::chrono::milliseconds>> tombstone_retention_ms;
+    bounded_property<std::optional<double>, numeric_bounds>
+      min_cleanable_dirty_ratio;
     property<bool> log_disable_housekeeping_for_tests;
     property<bool> log_compaction_use_sliding_window;
     // same as retention.size in kafka - TODO: size not implemented

src/v/kafka/server/handlers/alter_configs.cc

@@ -83,7 +83,7 @@ create_topic_properties_update(
     std::apply(apply_op(op_t::none), update.custom_properties.serde_fields());
 
     static_assert(
-      std::tuple_size_v<decltype(update.properties.serde_fields())> == 35,
+      std::tuple_size_v<decltype(update.properties.serde_fields())> == 36,
       "If you added a property, please decide on it's default alter config "
       "policy, and handle the update in the loop below");
     static_assert(
@@ -399,6 +399,14 @@ create_topic_properties_update(
                   });
                 continue;
             }
+            if (cfg.name == topic_property_min_cleanable_dirty_ratio) {
+                parse_and_set_tristate(
+                  update.properties.min_cleanable_dirty_ratio,
+                  cfg.value,
+                  kafka::config_resource_operation::set,
+                  min_cleanable_dirty_ratio_validator{});
+                continue;
+            }
         } catch (const validation_error& e) {
             return make_error_alter_config_resource_response<
               alter_configs_resource_response>(

src/v/kafka/server/handlers/configs/config_response_utils.cc

@@ -1037,6 +1037,18 @@ config_response_container_t make_topic_configs(
           describe_as_string<std::chrono::milliseconds>);
     }
 
+    add_topic_config_if_requested(
+      config_keys,
+      result,
+      topic_property_min_cleanable_dirty_ratio,
+      metadata_cache.get_default_min_cleanable_dirty_ratio(),
+      topic_property_min_cleanable_dirty_ratio,
+      topic_properties.min_cleanable_dirty_ratio,
+      include_synonyms,
+      maybe_make_documentation(
+        include_documentation,
+        config::shard_local_cfg().min_cleanable_dirty_ratio.desc()));
+
     return result;
 }
 

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

@@ -38,6 +38,7 @@
 
 #include <algorithm>
 #include <optional>
+#include <type_traits>
 
 namespace kafka {
 template<typename T>
@@ -539,6 +540,25 @@ struct iceberg_target_lag_ms_validator {
     }
 };
 
+struct min_cleanable_dirty_ratio_validator {
+    std::optional<ss::sstring>
+    operator()(const ss::sstring&, const tristate<double>& value) {
+        double min = 0.0;
+        double max = 1.0;
+        if (value.has_optional_value()) {
+            if (value.value() < min || value.value() > max) {
+                return fmt::format(
+                  "min.cleanable.dirty.ratio {} is outside of allowed range "
+                  "[{}, {}]",
+                  value.value(),
+                  min,
+                  max);
+            }
+        }
+        return std::nullopt;
+    }
+};
+
 template<typename T, typename... ValidatorTypes>
 requires requires(
   model::topic_namespace_view tns,
@@ -795,7 +815,9 @@ void parse_and_set_tristate(
     }
     // set property value
     if (op == config_resource_operation::set) {
-        auto parsed = boost::lexical_cast<int64_t>(*value);
+        using config_t
+          = std::conditional_t<std::is_floating_point_v<T>, T, int64_t>;
+        auto parsed = boost::lexical_cast<config_t>(*value);
         if (parsed <= 0) {
             property.value = tristate<T>{};
         } else {

src/v/kafka/server/handlers/create_topics.cc

@@ -79,7 +79,8 @@ bool is_supported(std::string_view name) {
        topic_property_iceberg_delete,
        topic_property_iceberg_partition_spec,
        topic_property_iceberg_invalid_record_action,
-       topic_property_iceberg_target_lag_ms});
+       topic_property_iceberg_target_lag_ms,
+       topic_property_min_cleanable_dirty_ratio});
 
     if (std::any_of(
           supported_configs.begin(),

src/v/kafka/server/handlers/incremental_alter_configs.cc

@@ -397,7 +397,14 @@ create_topic_properties_update(
                   iceberg_target_lag_ms_validator{});
                 continue;
             }
-
+            if (cfg.name == topic_property_min_cleanable_dirty_ratio) {
+                parse_and_set_tristate(
+                  update.properties.min_cleanable_dirty_ratio,
+                  cfg.value,
+                  op,
+                  min_cleanable_dirty_ratio_validator{});
+                continue;
+            }
         } catch (const validation_error& e) {
             vlog(
               klog.debug,