kafka_topics fails to increase partition count if topic list include both - topics to change and desired ones

[mrandreyeff]

Expected Behavior

Partition count changed, no error is thrown

Actual Behavior

Partition count changed, error is thrown:

Unable to initialize Kafka manager: KafkaManagerError: Error while updating topic 'topic_b' partitions. Error key is INVALID_PARTITIONS, Number of partitions is invalid.. Request was CreatePartitionsRequest_v0(topic_partitions=[(topic='topic_b', new_partitions=(count=<built-in method count of tuple object at 0x7fcc18234200>, assignment=[])), (topic='topic_a', new_partitions=(count=<built-in method count of tuple object at 0x7fcc180bdd40>, assignment=[[202, 203, 204]]))], timeout=15000, validate_only=False).

Tasks to Reproduce the Problem

  - name: "manage topics - init"
    kafka_topics:
      api_version: "2.6.0"
      topics:
        - name: topic_a
           partitions: 1
           replica_factor: 3
        - name: topic_b
           partitions: 1
           replica_factor: 3
      zookeeper: "localhost:2181"
      bootstrap_servers: "localhost:9092"

  - name: "manage topics - error"
    kafka_topics:
      api_version: "2.6.0"
      topics:
        - name: topic_a
           partitions: 2
           replica_factor: 3
        - name: topic_b
           partitions: 1
           replica_factor: 3
      zookeeper: "localhost:2181"
      bootstrap_servers: "localhost:9092"

Code with problem

            for _new_partition in range(partitions - old_partition):
                assignment = []
                for _replica in range(total_replica):
                    assignment.append(next(brokers_iterator))
                assignments.append(assignment)
            topics_assignments.append(
                (topic_name, (partitions, assignments))
            )

For the good topics (which do not require any change), the code adds record as well with empty assignments, e.g. from an example above:

(topic='topic_b', new_partitions=(count=<built-in method count of tuple object at 0x7fcc18234200>, assignment=[]

The proposed solution is to add an if partitions > old_partition: for that block.