WIP ducktape pinning test

tests/rptest/tests/leadership_transfer_test.py

@@ -16,6 +16,7 @@
 from rptest.services.redpanda import RESTART_LOG_ALLOW_LIST
 from ducktape.utils.util import wait_until
 from rptest.clients.kafka_cat import KafkaCat
+from rptest.clients.rpk import RpkTool
 from rptest.util import wait_until_result
 from rptest.clients.types import TopicSpec
 from rptest.services.admin import Admin
@@ -339,3 +340,59 @@ def all_partitions_present(num_nodes, per_node=None):
             expected_min = math.floor(expected_on_shard * 0.8)
             assert count >= expected_min, \
                 f"leader count on shard {s} ({count}) is < {expected_min}"
+
+
+class LeadershipPinningTest(RedpandaTest):
+    def __init__(self, test_context):
+        super(LeadershipPinningTest,
+              self).__init__(test_context=test_context,
+                             num_brokers=6,
+                             extra_rp_conf={
+                                 'enable_rack_awareness': True,
+                                 'leader_balancer_idle_timeout': 20000,
+                             })
+
+    def setUp(self):
+        pass
+
+    def _get_leaders_by_node(self):
+        kc = KafkaCat(self.redpanda)
+        md = kc.metadata()
+        leaders = []
+        for topic in md["topics"]:
+            leaders.extend(p["leader"] for p in topic["partitions"])
+        return dict(collections.Counter(leaders))
+
+    @cluster(num_nodes=6, log_allow_list=RESTART_LOG_ALLOW_LIST)
+    def test_leadership_pinning(self):
+        rack_layout = ['A', 'A', 'B', 'B', 'C', 'C']
+        for ix, node in enumerate(self.redpanda.nodes):
+            self.redpanda.set_extra_node_conf(node, {
+                'rack': rack_layout[ix],
+            })
+        self.redpanda.add_extra_rp_conf({'default_leaders_preference': "A"})
+        self.redpanda.start()
+
+        rpk = RpkTool(self.redpanda)
+
+        rpk.create_topic("foo", partitions=61, replicas=3)
+        rpk.create_topic("bar",
+                         partitions=20,
+                         replicas=3,
+                         config={"redpanda.leaders.preference": "B"})
+
+        rpk.alter_topic_config("bar", "redpanda.leaders.preference", "C")
+
+        for i in range(20):
+            time.sleep(5)
+            self.logger.warn(
+                f"FFF {sorted(self._get_leaders_by_node().items())}")
+
+        self.logger.warn(f"FFF ------------------")
+
+        rpk.alter_topic_config("bar", "redpanda.leaders.preference", "B")
+
+        for i in range(6):
+            time.sleep(5)
+            self.logger.warn(
+                f"FFF {sorted(self._get_leaders_by_node().items())}")