locator, replica: Add a way to wait for table's effective_replication_map change

locator/abstract_replication_strategy.cc

@@ -494,6 +494,7 @@ effective_replication_map::effective_replication_map(replication_strategy_ptr rs
         : _rs(std::move(rs))
         , _tmptr(std::move(tmptr))
         , _replication_factor(replication_factor)
+        , _validity_abort_source(std::make_unique<abort_source>())
 { }
 
 vnode_effective_replication_map::factory_key vnode_effective_replication_map::make_factory_key(const replication_strategy_ptr& rs, const token_metadata_ptr& tmptr) {

locator/abstract_replication_strategy.hh

@@ -173,8 +173,10 @@ protected:
     replication_strategy_ptr _rs;
     token_metadata_ptr _tmptr;
     size_t _replication_factor;
+    std::unique_ptr<abort_source> _validity_abort_source;
 public:
     effective_replication_map(replication_strategy_ptr, token_metadata_ptr, size_t replication_factor) noexcept;
+    effective_replication_map(effective_replication_map&&) noexcept = default;
     virtual ~effective_replication_map() = default;
 
     const abstract_replication_strategy& get_replication_strategy() const noexcept { return *_rs; }
@@ -183,6 +185,16 @@ public:
     const topology& get_topology() const noexcept { return _tmptr->get_topology(); }
     size_t get_replication_factor() const noexcept { return _replication_factor; }
 
+    void invalidate() const noexcept {
+        _validity_abort_source->request_abort();
+    }
+
+    /// Returns a reference to abort_source which is aborted when this effective replication map
+    /// is no longer the latest table's effective replication map.
+    abort_source& get_validity_abort_source() const {
+        return *_validity_abort_source;
+    }
+
     /// Returns addresses of replicas for a given token.
     /// Does not include pending replicas except for a pending replica which
     /// has the same address as one of the old replicas. This can be the case during "nodetool replace"

replica/table.cc

@@ -1674,6 +1674,9 @@ table::table(schema_ptr schema, config config, lw_shared_ptr<const storage_optio
 }
 
 void table::update_effective_replication_map(locator::effective_replication_map_ptr erm) {
+    if (_erm) {
+        _erm->invalidate();
+    }
     _erm = std::move(erm);
 }