Kafka 7382 - Guarantee atleast one replica of partition be alive during create topic. Kafka 7465 - Fix to add rack unaware partitions using --disable-rack-aware.

[sumannewton]

Fixed KAFKA-7382

Validate that the replica assignment provided during topic creation doesn't include brokers(broker ids) that are not registered to the kafka cluster

Reuse the method AdminZkClient.validateReplicaAssignment() by making it public. (Hope changing the modifier shouldn't be a problem)

• Test for creating a topic with replica assignment with valid brokers.
• Test for creating a topic with replica assignment with invalid brokers.

Fixed KAFKA-7465

Fix to add rack unaware partitions using --disable-rack-aware in kafka-topics.sh command.
Maintain consistency in AdminZkClient createTopic/addPartitions methods. Accept rackAwareMode in both-the methods unlike accepting brokers in addPartitions and accepting rackAwareMode in createTopic.

• Test to verify create/add partitions with rack aware disabled.

@rajinisivaram @cmccabe @mjsax @vvcephei @hachikuji Review and merge.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[omkreddy]

For consistency, we need to handle this in AdminManager also
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/AdminManager.scala#L95

maybe we can push this validation to createOrUpdateTopicPartitionAssignmentPathInZK method, which will be used both AdminManager, TopicCommand

[sumannewton]

Moving this piece of code inside createOrUpdateTopicPartitionAssignmentPathInZK is leading to ~30 test fails. Fixing them. Hope that should be fine.

[sumannewton]

@omkreddy Handling the code in createOrUpdateTopicPartitionAssignmentPathInZK is little tricky and will lead to serious issues:

• Topic Creation - Works fine.
• Add Partitions - Fails. Let's say a topic is created with 1 partition(replicas being 0,1) and let's say broker 0 goes down after creation. Now if I have to add partitions, alterTopic() internally calls createOrUpdateTopicPartitionAssignmentPathInZK and since the existing partition 0 doesn't have all the brokers up, it fails the add partitions functionality. So it's better we handle it outside the createOrUpdateTopicPartitionAssignmentPathInZK.
• Topic Deletion - Works fine.

[omkreddy]

@sumannewton may be we can call validateReplicaAssigment, only if update flag is false here: https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/zk/AdminZkClient.scala#L96

We should try to avoid code duplication between AdminManager, TopicCommand

[sumannewton]

Agreed @omkreddy . Even I had the same solution.

[sumannewton]

@omkreddy Also replacing

if (!replicas.toSet.subsetOf(availableBrokerIds))
        throw new BrokerNotAvailableException(
          s"Some brokers specified for partition id $partitionId are not available. " +
            s"Specified brokers: ${replicas.mkString(", ")}, " +
            s"available brokers: ${availableBrokerIds.mkString(", ")}.")

with

if ((replicas.toSet -- availableBrokerIds).size == replicas.size)
        throw new BrokerNotAvailableException(
          s"All brokers specified for partition id $partitionId are not available. " +
            s"Specified brokers: ${replicas.mkString(", ")}, " +
            s"available brokers: ${availableBrokerIds.mkString(", ")}.")

makes sense.
Throw error only if any partition has all unavailable replicas. With one available replica also, topic creation should happen. When those unavailable brokers come up, then they act as the replica.

[sumannewton]

@omkreddy @cmccabe @rajinisivaram @vvcephei @mjsax Addressed review comments.

[sumannewton]

Any update on this?

[omkreddy]

@sumannewton currently replica assignment validation is different for topic creation and partition addition. With current patch, we are trying to unify both code paths. This will change the addPartitons behavior also. I am fine with the change, but we need to get committer approval for the approach. Let us wait for others comments.

[sumannewton]

retest this please

[sumannewton]

@cmccabe @mjsax @vvcephei @ijuma Please review.

[omkreddy]

@sumannewton #5812 PR did some cleanups to AdminZkClient. Can you rebase the PR? Also can you raise separate PRs for Kafka-7382 and KAFKA-7465.

[sumannewton]

@omkreddy #5812 PR isn't merged yet. Once that is done, then I can rebase PR. KAFKA-7465 needs fixes of KAFKA-7382 for tests to run. If not, they will require another PR to merge the conflicts and make the tests run pass.