MINOR: Fix flaky test in ReplicaManagerTest (#21244)

Refer to
#20082 (comment).

Refactored the test to fix a race condition caused by dynamic Mockito
stubbing during test execution.

The previous implementation used `doReturn(false)` and `reset()` on a
spy object while a background thread was running, causing a
`ClassCastException`.

This patch replaces that logic with a thread-safe `AtomicBoolean` and
`doAnswer` approach to toggle the mock's behavior safely.

## Test Command
```
N=100; I=0; while [ $I -lt $N ] && ./gradlew cleanTest core:test --tests
ReplicaManagerTest -PmaxParallelForks=4 \
; do (( I=$I+1 )); echo "Completed run: $I"; sleep 1; done
```

## Test Result
```
BUILD SUCCESSFUL in 12s
151 actionable tasks: 2 executed, 149 up-to-date
Consider enabling configuration cache to speed up this build:
https://docs.gradle.org/9.2.1/userguide/configuration_cache_enabling.html
Completed run: 100
```

Reviewers: Gaurav Narula <[email protected]>, Chia-Ping Tsai
 <[email protected]>, PoAn Yang <[email protected]>

Author: Parkerhiphop

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

@@ -94,7 +94,7 @@ import java.io.{ByteArrayInputStream, File}
 import java.net.InetAddress
 import java.nio.file.{Files, Paths}
 import java.util
-import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong, AtomicReference}
 import java.util.concurrent.{Callable, CompletableFuture, ConcurrentHashMap, CountDownLatch, Future, TimeUnit}
 import java.util.function.{BiConsumer, Consumer}
 import java.util.stream.IntStream
@@ -5514,6 +5514,17 @@ class ReplicaManagerTest {
 
     try {
       val spiedPartition = spy(Partition(tpId, time, replicaManager))
+
+      // Prevent promotion of future replica
+      val blockPromotion = new AtomicBoolean(true)
+      doAnswer { invocation =>
+        if (blockPromotion.compareAndSet(true, false)) {
+          false
+        } else {
+          invocation.callRealMethod()
+        }
+      }.when(spiedPartition).maybeReplaceCurrentWithFutureReplica()
+
       replicaManager.addOnlinePartition(tp, spiedPartition)
 
       val leaderDelta = topicsCreateDelta(localId, isStartIdLeader = true, partitions = List(0, 1), List.empty, topic, topicIds(topic))
@@ -5526,9 +5537,6 @@ class ReplicaManagerTest {
       val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == firstLogDir).head
       replicaManager.alterReplicaLogDirs(Map(tp -> newReplicaFolder.getAbsolutePath))
 
-      // Prevent promotion of future replica
-      doReturn(false).when(spiedPartition).maybeReplaceCurrentWithFutureReplica()
-
       // Make sure the future log is created with the correct topic ID.
       val futureLog = replicaManager.futureLocalLogOrException(tp)
       assertEquals(Optional.of(topicId), futureLog.topicId)
@@ -5537,8 +5545,6 @@ class ReplicaManagerTest {
       val finalReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(it => it == firstLogDir || it == newReplicaFolder).head
       replicaManager.alterReplicaLogDirs(Map(tp -> finalReplicaFolder.getAbsolutePath))
 
-      reset(spiedPartition)
-
       TestUtils.waitUntilTrue(() => {
         replicaManager.replicaAlterLogDirsManager.shutdownIdleFetcherThreads()
         replicaManager.replicaAlterLogDirsManager.fetcherThreadMap.isEmpty