tests/AlterParts: test CreatePartitions API

Signed-off-by: NyaliaLui <nyalia@redpanda.com>

tests/rptest/tests/partition_reassignments_test.py

@@ -19,7 +19,7 @@
 from rptest.services.verifiable_producer import VerifiableProducer
 from rptest.services.admin import Admin
 from rptest.tests.redpanda_test import RedpandaTest
-from rptest.clients.rpk import RpkTool
+from rptest.clients.rpk import RpkTool, RpkException
 
 from rptest.clients.types import TopicSpec
 from rptest.clients.kafka_cli_tools import KafkaCliTools
@@ -495,6 +495,64 @@ def test_disable_alter_reassignment_api(self):
         except subprocess.CalledProcessError as e:
             assert "AlterPartitionReassignment API is disabled. See" in e.output
 
+    @cluster(num_nodes=6)
+    def test_add_partitions_with_inprogress_reassignments(self):
+        all_topic_names = [self.topics[0].name, self.topics[1].name]
+        initial_assignments, all_node_idx, producers = self.initial_setup_steps(
+            producer_config={
+                "topics": all_topic_names,
+                "throughput": 1024
+            })
+
+        self.wait_producers(producers, num_messages=100000)
+
+        reassignments = {}
+        for assignment in initial_assignments:
+            if assignment.name not in reassignments:
+                reassignments[assignment.name] = {}
+
+            for partition in assignment.partitions:
+                assert partition.id not in reassignments[assignment.name]
+                assert len(partition.replicas) == self.REPLICAS_COUNT
+                missing_node_idx = self.get_missing_node_idx(
+                    all_node_idx, partition.replicas)
+                # Trigger replica add and removal by replacing one of the replicas
+                partition.replicas[2] = missing_node_idx
+                reassignments[assignment.name][
+                    partition.id] = partition.replicas
+
+        self.logger.debug(
+            f"Replacing replicas. New assignments: {reassignments}")
+        kcl = KCL(self.redpanda)
+        alter_partition_reassignments_with_kcl(kcl, reassignments)
+
+        all_partition_idx = [p for p in range(self.PARTITION_COUNT)]
+
+        self.logger.debug("Expect in-progress reassignments")
+        responses = kcl.list_partition_reassignments()
+        self.logger.debug(responses)
+
+        all_node_idx_set = set(all_node_idx)
+        for res in responses:
+            assert res.topic in all_topic_names, 'Unexpected topic'
+            assert res.partition in all_partition_idx, 'Unexpected partition id'
+            assert set(
+                res.replicas).issubset(all_node_idx_set), 'Unexpected replica'
+            assert len(res.adding_replicas) > 0 or len(
+                res.removing_replicas
+            ) > 0, 'Expected in-progress reassignments'
+
+        rpk = RpkTool(self.redpanda)
+        try:
+            out = rpk.add_partitions(all_topic_names[0], 3)
+            raise RuntimeError(f'Expected failure: output: {out}')
+        except RpkException as ex:
+            if 'REASSIGNMENT_IN_PROGRESS: A partition reassignment is in progress.' in str(
+                    ex):
+                pass
+            else:
+                raise
+
 
 class PartitionReassignmentsACLsTest(RedpandaTest):
     topics = [TopicSpec()]