divijvaidya commented on code in PR #13848:
URL: https://github.com/apache/kafka/pull/13848#discussion_r1230801748
##########
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:
Is this sufficient validation? The topics might be leaked artifacts from
other tests / retries and might contain different data than what is expected.
So, we should probably validate that all partitions are empty as well.
As an alternative solution, can we delete the topic if it already exists
here and re-create it?
--
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
