KAFKA-18172 Move RemoteIndexCacheTest to the storage module (#19469)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

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

@@ -22,6 +22,7 @@ import kafka.utils.{CoreUtils, TestUtils}
 import org.apache.kafka.common.metrics.{KafkaMetric, MetricsContext, MetricsReporter}
 import org.apache.kafka.server.config.ServerConfigs
 import org.apache.kafka.server.metrics.MetricConfigs
+import org.apache.kafka.test.{TestUtils => JTestUtils}
 import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.params.ParameterizedTest
@@ -85,7 +86,7 @@ class KafkaMetricsReporterTest extends QuorumTestHarness {
     assertNotNull(KafkaMetricsReporterTest.MockMetricsReporter.JMXPREFIX.get())
 
     broker.shutdown()
-    TestUtils.assertNoNonDaemonThreads(this.getClass.getName)
+    JTestUtils.assertNoLeakedThreadsWithNameAndDaemonStatus(this.getClass.getName, true)
   }
 
   @AfterEach

core/src/test/scala/unit/kafka/utils/TestUtils.scala

@@ -943,16 +943,6 @@ object TestUtils extends Logging {
     } finally outputStream.close()
   }
 
-  // Note: Call this method in the test itself, rather than the @AfterEach method.
-  // Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
-  def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
-    val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
-      !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
-    }
-    val threadCount = nonDaemonThreads.size
-    assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
-  }
-
   /**
    * Create new LogManager instance with default configuration for testing
    */

storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java

@@ -149,9 +149,7 @@ public class RemoteIndexCache implements Closeable {
     private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
         return Caffeine.newBuilder()
                 .maximumWeight(maxSize)
-                .weigher((Uuid key, Entry entry) -> {
-                    return (int) entry.entrySizeBytes;
-                })
+                .weigher((Uuid key, Entry entry) -> (int) entry.entrySizeBytes)
                 // This listener is invoked each time an entry is being automatically removed due to eviction. The cache will invoke this listener
                 // during the atomic operation to remove the entry (refer: https://github.com/ben-manes/caffeine/wiki/Removal),
                 // hence, care must be taken to ensure that this operation is not expensive. Note that this listener is not invoked when