KAFKA-15738: Adding KRaft support in ConsumerWithLegacyMessageFormatIntegrationTest (#15171)

adixitconfluent · web-flow · commit 8cdf1abb0b89 · 2024-01-12T13:41:04.000+05:30
Reviewers: Manikumar Reddy &lt;manikumar.reddy@gmail.com&gt;
diff --git a/core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala
@@ -17,11 +17,13 @@
 package kafka.api
 
 import kafka.server.KafkaConfig
+import kafka.utils.TestInfoUtils
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.config.TopicConfig
 import org.junit.jupiter.api.Assertions.{assertEquals, assertNull, assertThrows}
-import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
 
 import java.util
 import java.util.{Collections, Optional, Properties}
@@ -32,12 +34,15 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
 
   override protected def brokerPropertyOverrides(properties: Properties): Unit = {
     // legacy message formats are only supported with IBP < 3.0
-    properties.put(KafkaConfig.InterBrokerProtocolVersionProp, "2.8")
+    // KRaft mode is not supported for inter.broker.protocol.version = 2.8, The minimum version required is 3.0-IV1"
+    if(!isKRaftTest())
+      properties.put(KafkaConfig.InterBrokerProtocolVersionProp, "2.8")
   }
 
   @nowarn("cat=deprecation")
-  @Test
-  def testOffsetsForTimes(): Unit = {
+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+  @ValueSource(strings = Array("zk", "kraft"))
+  def testOffsetsForTimes(quorum: String): Unit = {
     val numParts = 2
     val topic1 = "part-test-topic-1"
     val topic2 = "part-test-topic-2"
@@ -86,8 +91,22 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
     assertEquals(20, timestampTopic1P1.timestamp)
     assertEquals(Optional.of(0), timestampTopic1P1.leaderEpoch)
 
-    assertNull(timestampOffsets.get(new TopicPartition(topic2, 0)), "null should be returned when message format is 0.9.0")
-    assertNull(timestampOffsets.get(new TopicPartition(topic2, 1)), "null should be returned when message format is 0.9.0")
+    if(!isKRaftTest()) {
+      assertNull(timestampOffsets.get(new TopicPartition(topic2, 0)), "null should be returned when message format is 0.9.0")
+      assertNull(timestampOffsets.get(new TopicPartition(topic2, 1)), "null should be returned when message format is 0.9.0")
+    }
+    else {
+      // legacy message formats are supported for IBP version < 3.0 and KRaft runs on minimum version 3.0-IV1
+      val timestampTopic2P0 = timestampOffsets.get(new TopicPartition(topic2, 0))
+      assertEquals(40, timestampTopic2P0.offset)
+      assertEquals(40, timestampTopic2P0.timestamp)
+      assertEquals(Optional.of(0), timestampTopic2P0.leaderEpoch)
+
+      val timestampTopic2P1 = timestampOffsets.get(new TopicPartition(topic2, 1))
+      assertEquals(60, timestampTopic2P1.offset)
+      assertEquals(60, timestampTopic2P1.timestamp)
+      assertEquals(Optional.of(0), timestampTopic2P1.leaderEpoch)
+    }
 
     val timestampTopic3P0 = timestampOffsets.get(new TopicPartition(topic3, 0))
     assertEquals(80, timestampTopic3P0.offset)
@@ -98,8 +117,9 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
   }
 
   @nowarn("cat=deprecation")
-  @Test
-  def testEarliestOrLatestOffsets(): Unit = {
+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+  @ValueSource(strings = Array("zk", "kraft"))
+  def testEarliestOrLatestOffsets(quorum: String): Unit = {
     val topic0 = "topicWithNewMessageFormat"
     val topic1 = "topicWithOldMessageFormat"
     val prop = new Properties()
