MINOR: Remove unused TopicAndPartition usage in tests (#5419)

Also replace `TopicAndPartition` with `TopicPartition` in `MetadataCache`.

Reviewers: Ismael Juma <ismael@juma.me.uk>

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

@@ -24,7 +24,6 @@ import scala.collection.{Seq, Set, mutable}
 import scala.collection.JavaConverters._
 import kafka.cluster.{Broker, EndPoint}
 import kafka.api._
-import kafka.common.TopicAndPartition
 import kafka.controller.StateChangeLogger
 import kafka.utils.CoreUtils._
 import kafka.utils.Logging
@@ -71,7 +70,7 @@ class MetadataCache(brokerId: Int) extends Logging {
                                    errorUnavailableListeners: Boolean): Option[Iterable[MetadataResponse.PartitionMetadata]] = {
     cache.get(topic).map { partitions =>
       partitions.map { case (partitionId, partitionState) =>
-        val topicPartition = TopicAndPartition(topic, partitionId)
+        val topicPartition = new TopicPartition(topic, partitionId)
         val leaderBrokerId = partitionState.basePartitionState.leader
         val maybeLeader = getAliveEndpoint(leaderBrokerId, listenerName)
         val replicas = partitionState.basePartitionState.replicas.asScala.map(_.toInt)

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -20,7 +20,6 @@ import java.util.{ArrayList, Collections, Properties}
 import java.time.Duration
 
 import kafka.admin.ConsumerGroupCommand.{ConsumerGroupCommandOptions, ConsumerGroupService}
-import kafka.common.TopicAndPartition
 import kafka.log.LogConfig
 import kafka.network.SocketServer
 import kafka.security.auth._
@@ -70,7 +69,6 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
   val tp = new TopicPartition(topic, part)
   val logDir = "logDir"
   val deleteRecordsPartition = new TopicPartition(deleteTopic, part)
-  val topicAndPartition = TopicAndPartition(topic, part)
   val group = "my-group"
   val topicResource = Resource(Topic, topic, LITERAL)
   val groupResource = Resource(Group, group, LITERAL)

core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala

@@ -22,7 +22,6 @@ import java.util.ArrayList
 import java.util.concurrent.ExecutionException
 
 import kafka.admin.AclCommand
-import kafka.common.TopicAndPartition
 import kafka.security.auth._
 import kafka.server._
 import kafka.utils._
@@ -73,7 +72,6 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
   val wildcard = "*"
   val part = 0
   val tp = new TopicPartition(topic, part)
-  val topicAndPartition = TopicAndPartition(topic, part)
   val clientPrincipal: String
   val kafkaPrincipal: String
 

core/src/test/scala/integration/kafka/api/EndToEndClusterIdTest.scala

@@ -21,7 +21,6 @@ import java.util.concurrent.ExecutionException
 import java.util.concurrent.atomic.AtomicReference
 import java.util.Properties
 
-import kafka.common.TopicAndPartition
 import kafka.integration.KafkaServerTestHarness
 import kafka.server._
 import kafka.utils._
@@ -100,7 +99,6 @@ class EndToEndClusterIdTest extends KafkaServerTestHarness {
   val topic = "e2etopic"
   val part = 0
   val tp = new TopicPartition(topic, part)
-  val topicAndPartition = TopicAndPartition(topic, part)
   this.serverConfig.setProperty(KafkaConfig.MetricReporterClassesProp, classOf[MockBrokerMetricsReporter].getName)
 
   override def generateConfigs = {