Fix create_partitions sending empty assignments array instead of null (#3146)

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):

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'