MINOR: Remove unused imports, exceptions, and values (#6117)

1. Remove unthrown exceptions from MemoryRecordsBuilderTest
2. Remove unused imports from ReplicaFetcherThread, ZooKeeperClient, ApiVersionTest, PartitionTest
3. Remove unused value from PartitionTest

clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsBuilderTest.java

@@ -597,7 +597,7 @@ public class MemoryRecordsBuilderTest {
     }
 
     @Test
-    public void shouldThrowIllegalStateExceptionOnBuildWhenAborted() throws Exception {
+    public void shouldThrowIllegalStateExceptionOnBuildWhenAborted() {
         expectExceptionWithZStd(compressionType, RecordBatch.MAGIC_VALUE_V0);
 
         ByteBuffer buffer = ByteBuffer.allocate(128);
@@ -616,7 +616,7 @@ public class MemoryRecordsBuilderTest {
     }
 
     @Test
-    public void shouldResetBufferToInitialPositionOnAbort() throws Exception {
+    public void shouldResetBufferToInitialPositionOnAbort() {
         expectExceptionWithZStd(compressionType, RecordBatch.MAGIC_VALUE_V0);
 
         ByteBuffer buffer = ByteBuffer.allocate(128);
@@ -631,7 +631,7 @@ public class MemoryRecordsBuilderTest {
     }
 
     @Test
-    public void shouldThrowIllegalStateExceptionOnCloseWhenAborted() throws Exception {
+    public void shouldThrowIllegalStateExceptionOnCloseWhenAborted() {
         expectExceptionWithZStd(compressionType, RecordBatch.MAGIC_VALUE_V0);
 
         ByteBuffer buffer = ByteBuffer.allocate(128);
@@ -650,7 +650,7 @@ public class MemoryRecordsBuilderTest {
     }
 
     @Test
-    public void shouldThrowIllegalStateExceptionOnAppendWhenAborted() throws Exception {
+    public void shouldThrowIllegalStateExceptionOnAppendWhenAborted() {
         expectExceptionWithZStd(compressionType, RecordBatch.MAGIC_VALUE_V0);
 
         ByteBuffer buffer = ByteBuffer.allocate(128);

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

@@ -19,7 +19,6 @@ package kafka.server
 
 import java.util.Optional
 
-import kafka.api
 import kafka.api._
 import kafka.cluster.BrokerEndPoint
 import kafka.log.LogAppendInfo

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

@@ -29,7 +29,6 @@ import kafka.utils.{KafkaScheduler, Logging}
 import org.apache.kafka.common.utils.Time
 import org.apache.zookeeper.AsyncCallback._
 import org.apache.zookeeper.KeeperException.Code
-import org.apache.zookeeper.OpResult.{CreateResult, SetDataResult}
 import org.apache.zookeeper.Watcher.Event.{EventType, KeeperState}
 import org.apache.zookeeper.ZooKeeper.States
 import org.apache.zookeeper.data.{ACL, Stat}

core/src/test/scala/unit/kafka/api/ApiVersionTest.scala

@@ -17,13 +17,10 @@
 
 package kafka.api
 
-import org.apache.commons.collections.CollectionUtils
 import org.apache.kafka.common.record.RecordVersion
 import org.junit.Test
 import org.junit.Assert._
 
-import scala.collection.JavaConverters
-
 class ApiVersionTest {
 
   @Test

core/src/test/scala/unit/kafka/cluster/PartitionTest.scala

@@ -29,7 +29,7 @@ import kafka.server._
 import kafka.utils.{CoreUtils, MockScheduler, MockTime, TestUtils}
 import kafka.zk.KafkaZkClient
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.errors.{ApiException, LeaderNotAvailableException, OffsetNotAvailableException, ReplicaNotAvailableException}
+import org.apache.kafka.common.errors.{ApiException, OffsetNotAvailableException, ReplicaNotAvailableException}
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.record.FileRecords.TimestampAndOffset
@@ -405,9 +405,6 @@ class PartitionTest {
     val batch2 = TestUtils.records(records = List(new SimpleRecord("k3".getBytes, "v1".getBytes),
       new SimpleRecord(20,"k4".getBytes, "v2".getBytes),
       new SimpleRecord(21,"k5".getBytes, "v3".getBytes)))
-    val batch3 = TestUtils.records(records = List(
-      new SimpleRecord(30,"k6".getBytes, "v1".getBytes),
-      new SimpleRecord(31,"k7".getBytes, "v2".getBytes)))
 
     val partition = Partition(topicPartition, time, replicaManager)
     assertTrue("Expected first makeLeader() to return 'leader changed'",
@@ -421,7 +418,7 @@ class PartitionTest {
     val follower2Replica = partition.getReplica(follower2).get
 
     // append records with initial leader epoch
-    val lastOffsetOfFirstBatch = partition.appendRecordsToLeader(batch1, isFromClient = true).lastOffset
+    partition.appendRecordsToLeader(batch1, isFromClient = true)
     partition.appendRecordsToLeader(batch2, isFromClient = true)
     assertEquals("Expected leader's HW not move", leaderReplica.logStartOffset, leaderReplica.highWatermark.messageOffset)