producer_state_manager: make period configurable for testing

src/v/cluster/producer_state_manager.cc

@@ -41,7 +41,7 @@ producer_state_manager::producer_state_manager(
 
 ss::future<> producer_state_manager::start() {
     _reaper.set_callback([this] { evict_excess_producers(); });
-    _reaper.arm(period);
+    _reaper.arm(_reaper_period);
     vlog(clusterlog.info, "Started producer state manager");
     return ss::now();
 }
@@ -69,6 +69,12 @@ void producer_state_manager::setup_metrics() {
          sm::description("Number of evicted producers so far."))});
 }
 
+void producer_state_manager::rearm_eviction_timer_for_testing(
+  std::chrono::milliseconds new_period) {
+    _reaper_period = new_period;
+    _reaper.rearm(ss::lowres_clock::now() + _reaper_period);
+}
+
 void producer_state_manager::register_producer(
   producer_state& state, std::optional<model::vcluster_id> vcluster) {
     vlog(
@@ -97,7 +103,7 @@ void producer_state_manager::evict_excess_producers() {
     _cache.evict_older_than<ss::lowres_system_clock>(
       ss::lowres_system_clock::now() - _producer_expiration_ms());
     if (!_gate.is_closed()) {
-        _reaper.arm(period);
+        _reaper.arm(_reaper_period);
     }
 }
 

src/v/cluster/producer_state_manager.h

@@ -45,8 +45,10 @@ class producer_state_manager {
      */
     void touch(producer_state&, std::optional<model::vcluster_id>);
 
+    void rearm_eviction_timer_for_testing(std::chrono::milliseconds);
+
 private:
-    static constexpr std::chrono::seconds period{5};
+    std::chrono::milliseconds _reaper_period{5000};
     /**
      *  Constant to be used when a partition has no vcluster_id assigned.
      */
@@ -83,7 +85,7 @@ class producer_state_manager {
     config::binding<size_t> _virtual_cluster_min_producer_ids;
     // cache of all producers on this shard
     cache_t _cache;
-    ss::timer<ss::steady_clock_type> _reaper;
+    ss::timer<ss::lowres_clock> _reaper;
     ss::gate _gate;
     metrics::internal_metric_groups _metrics;