dajac commented on code in PR #13848:
URL: https://github.com/apache/kafka/pull/13848#discussion_r1231061065
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -611,11 +611,33 @@ object TestUtils extends Logging {
*/
def createOffsetsTopic(zkClient: KafkaZkClient, servers: Seq[KafkaBroker]):
Unit = {
val server = servers.head
- createTopic(zkClient, Topic.GROUP_METADATA_TOPIC_NAME,
- server.config.getInt(KafkaConfig.OffsetsTopicPartitionsProp),
-
server.config.getShort(KafkaConfig.OffsetsTopicReplicationFactorProp).toInt,
- servers,
- server.groupCoordinator.groupMetadataTopicConfigs)
+ val numPartitions = server.config.offsetsTopicPartitions
+ val replicationFactor = server.config.offsetsTopicReplicationFactor.toInt
+
+ try {
+ createTopic(
+ zkClient,
+ Topic.GROUP_METADATA_TOPIC_NAME,
+ numPartitions,
+ replicationFactor,
+ servers,
+ server.groupCoordinator.groupMetadataTopicConfigs
+ )
+ } catch {
+ case ex: TopicExistsException =>
+ val allPartitionsMetadata = waitForAllPartitionsMetadata(
+ servers,
+ Topic.GROUP_METADATA_TOPIC_NAME,
+ numPartitions
+ )
+
+ // If the topic already exists, we ensure that it has the required
+ // number of partitions and replication factor. If it has not, the
+ // exception is thrown further.
+ if (allPartitionsMetadata.size != numPartitions ||
allPartitionsMetadata.head._2.replicas.size != replicationFactor) {
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
