dajac commented on code in PR #13848:
URL: https://github.com/apache/kafka/pull/13848#discussion_r1230916047
##########
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:
We do the same when we create it with the admin client in KRaft mode and I
has been enough until today. The cluster is usually recreated for all tests so
it can't be leaked from a previous one. This error only happens when the first
creation fails for instance due to an network exception.
--
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
