divijvaidya commented on code in PR #13848:
URL: https://github.com/apache/kafka/pull/13848#discussion_r1231007534


##########
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 may be using the same for KRaft mode but do you agree that this approach 
of re-using existing topic artifacts is brittle? Even if this hasn't caused 
problems until today, it could cause problems in future. So, while we are 
fixing it, why not fix it in an ideal manner. The cost of checking the end 
offset is very small! We just have to ensure that it is zero.
   
   Alternatively, if you wish to keep the behaviour consistent with kraft 
version, could you please update the java doc for this method and mention that 
if a topic already exists with same partitions, we may retain the data.



-- 
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

Reply via email to