diff --git a/kafka/admin/_partitions.py b/kafka/admin/_partitions.py index f49c65b21..638dcfa49 100644 --- a/kafka/admin/_partitions.py +++ b/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): diff --git a/kafka/admin/_topics.py b/kafka/admin/_topics.py index 771c037dc..709896b0c 100644 --- a/kafka/admin/_topics.py +++ b/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: diff --git a/kafka/protocol/admin/topics.py b/kafka/protocol/admin/topics.py index e5fb9d487..50864c3c8 100644 --- a/kafka/protocol/admin/topics.py +++ b/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 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'