KAFKA-7940: wait until we've got the expected partition size

[showuon]

The test will create 99 partitions in a topic, and expect we can get the partition info after 15 seconds. If we can't get the partition info within 15 secs, we'll get the error:

org.scalatest.exceptions.TestFailedException: Partition [group1_largeTopic,69] metadata not propagated after 15000 ms

Obviously, 15 secs is not enough to complete the 99 partitions creation under slow system. So, fix it by explicitly wait until we've got the expected partition size before retrieving each partition info.

Committer Checklist (excluded from commit message)

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

[showuon]

@rajinisivaram , please help review this PR. Thanks.

[showuon]

@rajinisivaram , please help review this PR. Thanks.

[showuon]

@rajinisivaram @junrao , please help review this PR. Thanks.

[showuon]

@rajinisivaram @junrao , please help review this PR. Thanks.

[showuon]

@omkreddy @hachikuji , could you take a look for this PR?
The failed test cases keep increasing with the same root cause while we try to create topic with many partitions, it'll fail that to propagate the metadata in time.

https://ci-builds.apache.org/job/Kafka/job/kafka-trunk-jdk11/439/testReport/junit/kafka.server/MultipleListenersWithDefaultJaasContextTest/testProduceConsume__/
https://ci-builds.apache.org/job/Kafka/job/kafka-trunk-jdk11/437/testReport/kafka.api/PlaintextConsumerTest/testMultiConsumerStickyAssignment__/

[hachikuji]

@showuon Thanks for the patch and sorry for the delay. I left a couple comments.

[showuon]

@hachikuji , thanks for the comments. I've updated:

1. rename the method:
   waitUntilMetadataIsPropagatedWithExpectedSize -> waitForAllPartitionMetadata
waitUntilMetadataIsPropagated -> waitForPartitionMetadata
2. return the partition metadata for waitForAllPartitionMetadata and waitForPartitionMetadata.
3. skip the calls to waitUntilMetadataIsPropagated and waitUntilLeaderIsElectedOrChanged, get the info from the returned partition metadata
4. Use server.metadataCache directly instead of server.dataPlaneRequestProcessor.metadataCache

Thank you.

[showuon]

Don't replace them with waitForAllPartitionsMetadata because I'm afraid it'll break the original testing purposes. And same as other places.

[hachikuji]

LGTM. Thanks for the patch!