From 5c3acb650b0effc833a84b368eb4a99aea60a075 Mon Sep 17 00:00:00 2001
From: ebrard <emmanuel.brard@getyourguide.com>
Date: Tue, 4 Aug 2026 08:58:54 +0200
Subject: [PATCH] Fix create_partitions sending empty assignments array instead
 of null

_process_create_partitions_input always built the per-topic assignments
field as a list. For the int total-count form (and a bare NewPartitions
with no manual assignments) this produced an empty list [], which
serialises to a present-but-empty array. The broker reads that as a
manual replica assignment and rejects the request with
InvalidReplicationAssignmentError.

Pass assignments=None when the caller supplies no manual assignments so
it serialises to null and the broker auto-assigns replicas. Add unit
coverage for the auto-assign and manual-assignment paths.

Fixes #3146
---
 kafka/admin/_partitions.py      | 27 +++++++++++++-------------
 test/admin/test_admin_topics.py | 34 +++++++++++++++++++++++++++++++++
 2 files changed, 48 insertions(+), 13 deletions(-)

diff --git a/kafka/admin/_partitions.py b/kafka/admin/_partitions.py
index f49c65b21..efe8b1140 100644
--- a/kafka/admin/_partitions.py
+++ b/kafka/admin/_partitions.py
@@ -45,21 +45,22 @@ def _process_create_partitions_input(topic_partitions):
         topics = []
         for topic, count in topic_partitions.items():
             if isinstance(count, int):
-                topics.append(_Topic(name=topic, count=count))
+                total_count, new_assignments = count, None
             elif isinstance(count, dict):
-                topics.append(
-                    _Topic(
-                        name=topic,
-                        count=count['count'],
-                        assignments=[_Assignment(broker_ids=broker_ids)
-                                     for broker_ids in count['assignments']]))
+                total_count, new_assignments = count['count'], count.get('assignments')
             else:
-                topics.append(
-                    _Topic(
-                        name=topic,
-                        count=count.total_count,
-                        assignments=[_Assignment(broker_ids=broker_ids)
-                                     for broker_ids in count.new_assignments]))
+                total_count, new_assignments = count.total_count, count.new_assignments
+            # new_assignments is None -> no manual assignment; leave the request's
+            # assignments field as None so it serialises to null (the broker then
+            # auto-assigns replicas). Passing an empty list instead of None makes
+            # the broker treat it as a manual assignment and reject the request
+            # with InvalidReplicationAssignmentError. See issue #3146.
+            if new_assignments is None:
+                assignments = None
+            else:
+                assignments = [_Assignment(broker_ids=broker_ids)
+                               for broker_ids in new_assignments]
+            topics.append(_Topic(name=topic, count=total_count, assignments=assignments))
         return topics
 
     def create_partitions(self, topic_partitions, timeout_ms=None, validate_only=False, raise_errors=True):
diff --git a/test/admin/test_admin_topics.py b/test/admin/test_admin_topics.py
index 9b5270e5d..fa2d9b490 100644
--- a/test/admin/test_admin_topics.py
+++ b/test/admin/test_admin_topics.py
@@ -22,6 +22,40 @@ def test_new_partitions():
     assert good_partitions.new_assignments == [[1, 2, 3]]
 
 
+def test_process_create_partitions_input_auto_assign_is_null():
+    # When the caller does not supply manual replica assignments, the request's
+    # `assignments` field must be None so it serialises to null (broker
+    # auto-assigns). An empty list [] serialises to a present-but-empty array,
+    # which the broker reads as "manual assignment" and rejects with
+    # InvalidReplicationAssignmentError. See issue #3146.
+    proc = KafkaAdminClient._process_create_partitions_input
+
+    # int total-count -> auto-assign
+    topics = proc({'foo': 6})
+    assert len(topics) == 1
+    assert topics[0].name == 'foo'
+    assert topics[0].count == 6
+    assert topics[0].assignments is None
+
+    # bare NewPartitions with no manual assignments -> auto-assign
+    topics = proc({'foo': NewPartitions(6)})
+    assert topics[0].assignments is None
+
+
+def test_process_create_partitions_input_manual_assignments():
+    # Explicit manual assignments must be preserved, for both the dict form
+    # and the (deprecated) NewPartitions form.
+    proc = KafkaAdminClient._process_create_partitions_input
+
+    topics = proc({'foo': {'count': 7, 'assignments': [[1, 2, 3]]}})
+    assert topics[0].count == 7
+    assert [a.broker_ids for a in topics[0].assignments] == [[1, 2, 3]]
+
+    topics = proc({'foo': NewPartitions(7, [[1, 2, 3]])})
+    assert topics[0].count == 7
+    assert [a.broker_ids for a in topics[0].assignments] == [[1, 2, 3]]
+
+
 def test_new_topic():
     good_topic = NewTopic('foo')
     assert good_topic.name == 'foo'