fix server side error msg

Author: TaiJuWu

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

@@ -1488,7 +1488,7 @@ class ReplicaManager(val config: KafkaConfig,
           debug(s"OffsetRequest with correlation id $correlationId from client $clientId on partition ${topicIdPartition} " +
             s"failed because the provided topic ID ${requestTopicIdOpt.get} does not match the current topic ID $cachedTopicIdOpt.")
           statusByPartition += topicIdPartition ->
-            ListOffsetsPartitionStatus.builder().responseOpt(Optional.of(buildErrorResponse(Errors.UNKNOWN_TOPIC_ID, partition))).build()
+            ListOffsetsPartitionStatus.builder().responseOpt(Optional.of(buildErrorResponse(Errors.INCONSISTENT_TOPIC_ID, partition))).build()
         } else if (isListOffsetsTimestampUnsupported(partition.timestamp(), version)) {
           statusByPartition += topicIdPartition ->
             ListOffsetsPartitionStatus.builder().responseOpt(Optional.of(buildErrorResponse(Errors.UNSUPPORTED_VERSION, partition))).build()

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

@@ -4082,7 +4082,7 @@ class ReplicaManagerTest {
   }
 
   @Test
-  def testFetchOffsetReturnsUnknownTopicIdOnMismatch(): Unit = {
+  def testFetchOffsetReturnsInconsistentTopicIdOnMismatch(): Unit = {
     val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time))
     try {
       setupMetadataCacheWithTopicIds(topicIds, replicaManager.metadataCache)
@@ -4118,7 +4118,7 @@ class ReplicaManagerTest {
 
       assertNotNull(capturedResponses)
       val partitionResponse = capturedResponses.iterator().next().partitions().iterator().next()
-      assertEquals(Errors.UNKNOWN_TOPIC_ID.code, partitionResponse.errorCode)
+      assertEquals(Errors.INCONSISTENT_TOPIC_ID.code, partitionResponse.errorCode)
     } finally {
       replicaManager.shutdown(checkpointHW = false)
     }