diff --git a/core/src/main/scala/kafka/utils/CoreUtils.scala b/core/src/main/scala/kafka/utils/CoreUtils.scala
index 3aca6a5d34ee3..221399cf7ff6a 100755
--- a/core/src/main/scala/kafka/utils/CoreUtils.scala
+++ b/core/src/main/scala/kafka/utils/CoreUtils.scala
@@ -19,7 +19,6 @@ package kafka.utils
 
 import java.io._
 import java.nio._
-import java.nio.channels._
 import java.util.concurrent.locks.{Lock, ReadWriteLock}
 import java.lang.management._
 import java.util.{Base64, Properties, UUID}
@@ -50,12 +49,6 @@ import scala.annotation.nowarn
 object CoreUtils {
   private val logger = Logger(getClass)
 
-  /**
-   * Return the smallest element in `iterable` if it is not empty. Otherwise return `ifEmpty`.
-   */
-  def min[A, B >: A](iterable: Iterable[A], ifEmpty: A)(implicit cmp: Ordering[B]): A =
-    if (iterable.isEmpty) ifEmpty else iterable.min(cmp)
-
   /**
     * Do the given action and log any exceptions thrown without rethrowing them.
     *
@@ -134,30 +127,6 @@ object CoreUtils {
     }
   }
 
-  /**
-   * Unregister the mbean with the given name, if there is one registered
-   * @param name The mbean name to unregister
-   */
-  def unregisterMBean(name: String): Unit = {
-    val mbs = ManagementFactory.getPlatformMBeanServer()
-    mbs synchronized {
-      val objName = new ObjectName(name)
-      if (mbs.isRegistered(objName))
-        mbs.unregisterMBean(objName)
-    }
-  }
-
-  /**
-   * Read some bytes into the provided buffer, and return the number of bytes read. If the
-   * channel has been closed or we get -1 on the read for any reason, throw an EOFException
-   */
-  def read(channel: ReadableByteChannel, buffer: ByteBuffer): Int = {
-    channel.read(buffer) match {
-      case -1 => throw new EOFException("Received -1 when reading from channel, socket has likely been closed.")
-      case n => n
-    }
-  }
-
   /**
    * This method gets comma separated values which contains key,value pairs and returns a map of
    * key value pairs. the format of allCSVal is key1:val1, key2:val2 ....
@@ -195,14 +164,6 @@ object CoreUtils {
     constructor.newInstance(args: _*)
   }
 
-  /**
-   * Create a circular (looping) iterator over a collection.
-   * @param coll An iterable over the underlying collection.
-   * @return A circular iterator over the collection.
-   */
-  def circularIterator[T](coll: Iterable[T]) =
-    for (_ <- Iterator.continually(1); t <- coll) yield t
-
   /**
    * Execute the given function inside the lock
    */
diff --git a/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
index 88faa24983a94..832810d82f6b7 100755
--- a/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
@@ -23,7 +23,6 @@ import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.locks.ReentrantLock
 import java.nio.ByteBuffer
 import java.util.regex.Pattern
-
 import org.junit.jupiter.api.Assertions._
 import kafka.utils.CoreUtils.inLock
 import org.apache.kafka.common.KafkaException
@@ -34,7 +33,7 @@ import org.slf4j.event.Level
 import scala.jdk.CollectionConverters._
 import scala.collection.mutable
 import scala.concurrent.duration.Duration
-import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutorService, Future}
 
 class CoreUtilsTest extends Logging {
 
@@ -60,9 +59,9 @@ class CoreUtilsTest extends Logging {
 
     CoreUtils.tryAll(Seq(
       () => recordingFunction(Right("valid-0")),
-      () => recordingFunction(Left(new TestException("exception-1"))),
+      () => recordingFunction(Left(TestException("exception-1"))),
       () => recordingFunction(Right("valid-2")),
-      () => recordingFunction(Left(new TestException("exception-3")))
+      () => recordingFunction(Left(TestException("exception-3")))
     ))
     var expected = Map(
       "valid-0" -> Right("valid-0"),
@@ -85,8 +84,8 @@ class CoreUtilsTest extends Logging {
 
     recorded.clear()
     CoreUtils.tryAll(Seq(
-      () => recordingFunction(Left(new TestException("exception-0"))),
-      () => recordingFunction(Left(new TestException("exception-1")))
+      () => recordingFunction(Left(TestException("exception-0"))),
+      () => recordingFunction(Left(TestException("exception-1")))
     ))
     expected = Map(
       "exception-0" -> Left(TestException("exception-0")),
@@ -95,25 +94,6 @@ class CoreUtilsTest extends Logging {
     assertEquals(expected, recorded)
   }
 
-  @Test
-  def testCircularIterator(): Unit = {
-    val l = List(1, 2)
-    val itl = CoreUtils.circularIterator(l)
-    assertEquals(1, itl.next())
-    assertEquals(2, itl.next())
-    assertEquals(1, itl.next())
-    assertEquals(2, itl.next())
-    assertFalse(itl.isEmpty)
-
-    val s = Set(1, 2)
-    val its = CoreUtils.circularIterator(s)
-    assertEquals(1, its.next())
-    assertEquals(2, its.next())
-    assertEquals(1, its.next())
-    assertEquals(2, its.next())
-    assertEquals(1, its.next())
-  }
-
   @Test
   def testReadBytes(): Unit = {
     for(testCase <- List("", "a", "abcd")) {
@@ -221,9 +201,9 @@ class CoreUtilsTest extends Logging {
     val nThreads = 5
     val createdCount = new AtomicInteger
     val map = new ConcurrentHashMap[Int, AtomicInteger]().asScala
-    implicit val executionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))
+    implicit val executionContext: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))
     try {
-      Await.result(Future.traverse(1 to count) { i =>
+      Await.result(Future.traverse(1 to count) { _ =>
         Future {
           CoreUtils.atomicGetOrUpdate(map, 0, {
             createdCount.incrementAndGet