Merge pull request #15071 from ztlpn/fix-whole-topic-enable

Fix enabling whole topic no-op check
redpanda-data · Nov 22, 2023 · b58d8bb · b58d8bb
2 parents 622e724 + 37dd96c
commit b58d8bb
Show file tree

Hide file tree

Showing 6 changed files with 33 additions and 11 deletions.
diff --git a/src/v/cluster/topic_table.cc b/src/v/cluster/topic_table.cc
@@ -644,12 +644,12 @@ topic_table::apply(set_topic_partitions_disabled_cmd cmd, model::offset o) {
             disabled_set.remove(*cmd.value.partition_id, assignments);
         }
 
-        if (disabled_set.is_empty()) {
+        if (disabled_set.is_fully_enabled()) {
             _disabled_partitions.erase(disabled_it);
         }
     } else {
         if (cmd.value.disabled) {
-            _disabled_partitions[cmd.value.ns_tp].set_topic_disabled();
+            _disabled_partitions[cmd.value.ns_tp].set_fully_disabled();
         } else {
             _disabled_partitions.erase(cmd.value.ns_tp);
         }

diff --git a/src/v/cluster/topic_table.h b/src/v/cluster/topic_table.h
@@ -518,12 +518,20 @@ class topic_table {
         return &it->second;
     }
 
-    bool is_disabled(model::topic_namespace_view ns_tp) const {
+    bool is_fully_disabled(model::topic_namespace_view ns_tp) const {
         auto it = _disabled_partitions.find(ns_tp);
         if (it == _disabled_partitions.end()) {
             return false;
         }
-        return it->second.is_topic_disabled();
+        return it->second.is_fully_disabled();
+    }
+
+    bool is_fully_enabled(model::topic_namespace_view ns_tp) const {
+        auto it = _disabled_partitions.find(ns_tp);
+        if (it == _disabled_partitions.end()) {
+            return true;
+        }
+        return it->second.is_fully_enabled();
     }
 
     bool is_disabled(

diff --git a/src/v/cluster/topics_frontend.cc b/src/v/cluster/topics_frontend.cc
@@ -1040,7 +1040,11 @@ ss::future<std::error_code> topics_frontend::set_topic_partitions_disabled(
         if (!_topics.local().contains(ns_tp)) {
             co_return errc::topic_not_exists;
         }
-        if (_topics.local().is_disabled(ns_tp) == disabled) {
+        if (disabled && _topics.local().is_fully_disabled(ns_tp)) {
+            // no-op
+            co_return errc::success;
+        }
+        if (!disabled && _topics.local().is_fully_enabled(ns_tp)) {
             // no-op
             co_return errc::success;
         }

diff --git a/src/v/cluster/types.h b/src/v/cluster/types.h
@@ -4418,12 +4418,12 @@ struct topic_disabled_partitions_set
     bool is_disabled(model::partition_id id) const {
         return !partitions || partitions->contains(id);
     }
-    bool is_topic_disabled() const { return !partitions.has_value(); }
-    bool is_empty() const { return partitions && partitions->empty(); }
+    bool is_fully_disabled() const { return !partitions.has_value(); }
+    bool is_fully_enabled() const { return partitions && partitions->empty(); }
 
     void add(model::partition_id id);
     void remove(model::partition_id id, const assignments_set& all_partitions);
-    void set_topic_disabled() { partitions = std::nullopt; }
+    void set_fully_disabled() { partitions = std::nullopt; }
 };
 
 } // namespace cluster

diff --git a/src/v/redpanda/admin/server.cc b/src/v/redpanda/admin/server.cc
@@ -4530,13 +4530,13 @@ fragmented_vector<cluster_partition_info> topic2cluster_partitions(
         // fast exits
         if (
           disabled_filter.value()
-          && (!disabled_set || disabled_set->is_empty())) {
+          && (!disabled_set || disabled_set->is_fully_enabled())) {
             return ret;
         }
 
         if (
           !disabled_filter.value() && disabled_set
-          && disabled_set->is_topic_disabled()) {
+          && disabled_set->is_fully_disabled()) {
             return ret;
         }
     }

diff --git a/tests/rptest/tests/recovery_mode_test.py b/tests/rptest/tests/recovery_mode_test.py
@@ -242,7 +242,7 @@ def test_apis(self):
         rpk = RpkTool(self.redpanda)
         admin = Admin(self.redpanda)
 
-        topics = ["mytopic1", "mytopic2", "mytopic3"]
+        topics = ["mytopic1", "mytopic2", "mytopic3", "mytopic4"]
         for topic in topics:
             rpk.create_topic(topic, partitions=3, replicas=3)
 
@@ -263,6 +263,13 @@ def test_apis(self):
                                       partition=1,
                                       value=False)
 
+        admin.set_partitions_disabled(ns="kafka",
+                                      topic="mytopic4",
+                                      partition=0)
+        admin.set_partitions_disabled(ns="kafka",
+                                      topic="mytopic4",
+                                      value=False)
+
         self.sync()
 
         def pi(topic_partition, disabled=False):
@@ -279,6 +286,9 @@ def pi(topic_partition, disabled=False):
             pi('mytopic3/0', True),
             pi('mytopic3/1', False),
             pi('mytopic3/2', True),
+            pi('mytopic4/0', False),
+            pi('mytopic4/1', False),
+            pi('mytopic4/2', False),
         ]
 
         def filtered(topic, partition, disabled):