Fix create_partitions sending empty assignments array instead of null

kafka/admin/_partitions.py

@@ -54,12 +54,17 @@ def _process_create_partitions_input(topic_partitions):
                         assignments=[_Assignment(broker_ids=broker_ids)
                                      for broker_ids in count['assignments']]))
             else:
+                # Deprecated dict[topic, NewPartitions]
+                if count.new_assignments is not None:
+                    new_assignments = [_Assignment(broker_ids=broker_ids)
+                                       for broker_ids in count.new_assignments]
+                else:
+                    new_assignments = None
                 topics.append(
                     _Topic(
                         name=topic,
                         count=count.total_count,
-                        assignments=[_Assignment(broker_ids=broker_ids)
-                                     for broker_ids in count.new_assignments]))
+                        assignments=new_assignments))
         return topics
 
     def create_partitions(self, topic_partitions, timeout_ms=None, validate_only=False, raise_errors=True):

kafka/admin/_topics.py

@@ -12,7 +12,7 @@
 
 import kafka.errors as Errors
 from kafka.errors import IncompatibleBrokerVersion
-from kafka.protocol.admin import CreateTopicsRequest, DeleteTopicsRequest, CreatePartitionsRequest
+from kafka.protocol.admin import CreateTopicsRequest, DeleteTopicsRequest
 from ._configs import ConfigResourceType
 
 if TYPE_CHECKING:

kafka/protocol/admin/topics.py

@@ -17,7 +17,12 @@ def encode(self, version=None, header=False, framed=False):
 
 class DeleteTopicsResponse(ApiMessage): pass
 
-class CreatePartitionsRequest(ApiMessage): pass
+class CreatePartitionsRequest(ApiMessage):
+    @classmethod
+    def json_patch(cls, json):
+        json['fields'][0]['fields'][2]['default'] = 'null'
+        return json
+
 class CreatePartitionsResponse(ApiMessage): pass
 
 class AlterPartitionReassignmentsRequest(ApiMessage): pass

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'