MINOR: Fix flaky test in ReplicaManagerTest

[Parkerhiphop]

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 gaurav_narula2@apple.com, Chia-Ping Tsai
chia7712@gmail.com, PoAn Yang payang@apache.org

[chia7712]

will the following changes be sufficient?

      val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == firstLogDir).head
      // Prevent promotion of future replica
      doReturn(false).when(spiedPartition).maybeReplaceCurrentWithFutureReplica()

      replicaManager.alterReplicaLogDirs(Map(tp -> newReplicaFolder.getAbsolutePath))

[gaurav-narula]

I think the spied call might still not be visible to the ReplicaAlterLogDirsThread. IIUC, the visibility across threads can only be enforced through some synchronization.

[Parkerhiphop]

Thanks @gaurav-narula for the explanation on visibility!

[gaurav-narula]

Thanks for the change @Parkerhiphop. You might want to initialise this as new AtomicBoolean(true) and avoid invoking blockPromotion.set(true) explicitly

[Parkerhiphop]

@gaurav-narula Thanks! That's a great refinement! It makes the code much cleaner.

I have updated the code to new AtomicBoolean(true) and removed the explicit set(true) call.

The logic remains the same (blocking the first attempt via CAS), and I verified it with the loop test again (same result as before)

[gaurav-narula]

Thanks for the PR. Left some comments

[gaurav-narula]

I think this is racey. What if the test thread is interrupted and ReplicaAlterDirsThread invokes spiedPartition.maybeReplaceCurrentWithFutureReplica() before this line executes?

Perhaps consider a CAS within doAnswer instead?

[chia7712]

+1 to use CAS if (blockPromotion.compareAndSet(true, false))

[Parkerhiphop]

Thanks for the CAS suggestion.

I have addressed the race condition and simplified the logic as suggested:

1. Execution Order: Moved blockPromotion.set(true) beforealterReplicaLogDirs to ensure the flag is set before the background thread starts.
2. CAS: Adopted compareAndSet(true, false) within doAnswer to automatically unblock the promotion after the first attempt.

To verify the stability, I ran the following test command with no failures (also updated in the PR description):

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

[gaurav-narula]

I think the spied call might still not be visible to the ReplicaAlterLogDirsThread. IIUC, the visibility across threads can only be enforced through some synchronization.

[Parkerhiphop]

Apologies for the force push.
I messed up the git merge locally which caused unrelated commits from trunk to appear in this PR.
I have cleaned up the commit history and re-synced with trunk.
The actual code changes remain the same.

[chia7712]

the fatal error is caused by ReconfigurableQuorumIntegrationTest.testRemoveAndAddSameController. open https://issues.apache.org/jira/browse/KAFKA-20044