fix the failing unit tests

Author: kamalcph

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -23,7 +23,7 @@ import kafka.log.remote.RemoteLogManager
 import kafka.log.{LogManager, OffsetResultHolder, UnifiedLog}
 import kafka.server.HostedPartition.Online
 import kafka.server.QuotaFactory.QuotaManagers
-import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrUpdatesPerSecMetricName, IsrExpandsPerSecMetricName, IsrShrinksPerSecMetricName, LeaderCountMetricName, OfflineReplicaCountMetricName, PartitionCountMetricName, PartitionsWithLateTransactionsCountMetricName, ProducerIdCountMetricName, ReassigningPartitionsMetricName, UnderMinIsrPartitionCountMetricName, UnderReplicatedPartitionsMetricName, createLogReadResult}
+import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrUpdatesPerSecMetricName, IsrExpandsPerSecMetricName, IsrShrinksPerSecMetricName, LeaderCountMetricName, OfflineReplicaCountMetricName, PartitionCountMetricName, PartitionsWithLateTransactionsCountMetricName, ProducerIdCountMetricName, ReassigningPartitionsMetricName, UnderMinIsrPartitionCountMetricName, UnderReplicatedPartitionsMetricName, createLogReadResult, isListOffsetsTimestampUnsupported}
 import kafka.server.metadata.ZkMetadataCache
 import kafka.utils.Implicits._
 import kafka.utils._
@@ -226,6 +226,14 @@ object ReplicaManager {
 
   private[server] val MetricNames = GaugeMetricNames.union(MeterMetricNames)
 
+  private val timestampMinSupportedVersion: immutable.Map[Long, Short] = immutable.Map[Long, Short](
+    ListOffsetsRequest.EARLIEST_TIMESTAMP -> 1.toShort,
+    ListOffsetsRequest.LATEST_TIMESTAMP -> 1.toShort,
+    ListOffsetsRequest.MAX_TIMESTAMP -> 7.toShort,
+    ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP -> 8.toShort,
+    ListOffsetsRequest.LATEST_TIERED_TIMESTAMP -> 9.toShort
+  )
+
   def createLogReadResult(highWatermark: Long,
                           leaderLogStartOffset: Long,
                           leaderLogEndOffset: Long,
@@ -252,6 +260,11 @@ object ReplicaManager {
       lastStableOffset = None,
       exception = Some(e))
   }
+
+  private[server] def isListOffsetsTimestampUnsupported(timestamp: JLong, version: Short): Boolean = {
+    timestamp < 0 &&
+      (!timestampMinSupportedVersion.contains(timestamp) || version < timestampMinSupportedVersion(timestamp))
+  }
 }
 
 class ReplicaManager(val config: KafkaConfig,
@@ -1462,13 +1475,6 @@ class ReplicaManager(val config: KafkaConfig,
                   version: Short,
                   buildErrorResponse: (Errors, ListOffsetsPartition) => ListOffsetsPartitionResponse,
                   responseCallback: List[ListOffsetsTopicResponse] => Unit): Unit = {
-    val timestampMinSupportedVersion = immutable.Map[Long, Short](
-      ListOffsetsRequest.EARLIEST_TIMESTAMP -> 1.toShort,
-      ListOffsetsRequest.LATEST_TIMESTAMP -> 1.toShort,
-      ListOffsetsRequest.MAX_TIMESTAMP -> 7.toShort,
-      ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP -> 8.toShort,
-      ListOffsetsRequest.LATEST_TIERED_TIMESTAMP -> 9.toShort
-    )
     val statusByPartition = mutable.Map[TopicPartition, ListOffsetsPartitionStatus]()
     topics.foreach { topic =>
       topic.partitions.asScala.foreach { partition =>
@@ -1477,9 +1483,8 @@ class ReplicaManager(val config: KafkaConfig,
           debug(s"OffsetRequest with correlation id $correlationId from client $clientId on partition $topicPartition " +
             s"failed because the partition is duplicated in the request.")
           statusByPartition += topicPartition -> ListOffsetsPartitionStatus(Some(buildErrorResponse(Errors.INVALID_REQUEST, partition)))
-        } else if (partition.timestamp() < 0 &&
-          (!timestampMinSupportedVersion.contains(partition.timestamp()) || version < timestampMinSupportedVersion(partition.timestamp()))) {
-          buildErrorResponse(Errors.UNSUPPORTED_VERSION, partition)
+        } else if (isListOffsetsTimestampUnsupported(partition.timestamp(), version)) {
+          statusByPartition += topicPartition -> ListOffsetsPartitionStatus(Some(buildErrorResponse(Errors.UNSUPPORTED_VERSION, partition)))
         } else {
           try {
             val fetchOnlyFromLeader = replicaId != ListOffsetsRequest.DEBUGGING_REPLICA_ID

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -10165,6 +10165,31 @@ class KafkaApisTest extends Logging {
         .setPartitionIndex(tp.partition)
         .setTimestamp(timestamp)).asJava)).asJava
 
+    when(replicaManager.fetchOffset(
+      ArgumentMatchers.any[Seq[ListOffsetsTopic]](),
+      ArgumentMatchers.eq(Set.empty[TopicPartition]),
+      ArgumentMatchers.eq(IsolationLevel.READ_UNCOMMITTED),
+      ArgumentMatchers.eq(ListOffsetsRequest.CONSUMER_REPLICA_ID),
+      ArgumentMatchers.eq[String](clientId),
+      ArgumentMatchers.anyInt(), // correlationId
+      ArgumentMatchers.anyShort(), // version
+      ArgumentMatchers.any[(Errors, ListOffsetsPartition) => ListOffsetsPartitionResponse](),
+      ArgumentMatchers.any[List[ListOffsetsTopicResponse] => Unit]()
+    )).thenAnswer(ans => {
+      val version = ans.getArgument[Short](6)
+      val callback = ans.getArgument[List[ListOffsetsTopicResponse] => Unit](8)
+      val errorCode = if (ReplicaManager.isListOffsetsTimestampUnsupported(timestamp, version))
+        Errors.UNSUPPORTED_VERSION.code()
+      else
+        Errors.INVALID_REQUEST.code()
+      val partitionResponse = new ListOffsetsPartitionResponse()
+        .setErrorCode(errorCode)
+        .setOffset(ListOffsetsResponse.UNKNOWN_OFFSET)
+        .setTimestamp(ListOffsetsResponse.UNKNOWN_TIMESTAMP)
+        .setPartitionIndex(tp.partition())
+      callback(List(new ListOffsetsTopicResponse().setName(tp.topic()).setPartitions(List(partitionResponse).asJava)))
+    })
+
     val data = new ListOffsetsRequestData().setTopics(targetTimes).setReplicaId(ListOffsetsRequest.CONSUMER_REPLICA_ID)
     val listOffsetRequest = ListOffsetsRequest.parse(MessageUtil.toByteBuffer(data, version), version)
     val request = buildRequest(listOffsetRequest)