test(gax): isolate local executors inside loops in ScheduledRetryingExecutorTest

Addressed PR comments by moving the creation and shutdown of local
executors inside the test loops for testSuccessWithFailuresPeekAttempt
and testCancelOuterFutureAfterStart to ensure complete isolation
between iterations and avoid potential flakiness.

TAG=agy
CONV=71af906a-b7dc-4dda-a2fa-ebe24b5d8b91
BUG=13535

Author: lqiu96

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java

@@ -103,10 +103,9 @@ void testSuccessWithFailuresPeekAttempt() throws Exception {
             .setTotalTimeoutDuration(java.time.Duration.ofMillis(10000L))
             .setMaxAttempts(100)
             .build();
-    ScheduledExecutorService localExecutor = Executors.newSingleThreadScheduledExecutor();
-    try {
-      for (int executionsCount = 0; executionsCount < EXECUTIONS_COUNT; executionsCount++) {
-
+    for (int executionsCount = 0; executionsCount < EXECUTIONS_COUNT; executionsCount++) {
+      ScheduledExecutorService localExecutor = Executors.newSingleThreadScheduledExecutor();
+      try {
         FailingCallable callable = new FailingCallable(15, "request", "SUCCESS", tracer);
 
         RetryingExecutorWithContext<String> executor =
@@ -151,9 +150,9 @@ void testSuccessWithFailuresPeekAttempt() throws Exception {
         assertFutureSuccess(future);
         assertEquals(15, future.getAttemptSettings().getAttemptCount());
         assertTrue(failedAttempts.get() > 0);
+      } finally {
+        localExecutor.shutdownNow();
       }
-    } finally {
-      localExecutor.shutdownNow();
     }
   }
 
@@ -284,9 +283,9 @@ void testCancelOuterFutureAfterStart() throws Exception {
             // tiny RRD value is small, but not impossible.
             .setJittered(false)
             .build();
-    ScheduledExecutorService localExecutor = Executors.newSingleThreadScheduledExecutor();
-    try {
-      for (int executionsCount = 0; executionsCount < EXECUTIONS_COUNT; executionsCount++) {
+    for (int executionsCount = 0; executionsCount < EXECUTIONS_COUNT; executionsCount++) {
+      ScheduledExecutorService localExecutor = Executors.newSingleThreadScheduledExecutor();
+      try {
         FailingCallable callable = new FailingCallable(4, "request", "SUCCESS", tracer);
         RetryingExecutorWithContext<String> executor =
             getRetryingExecutor(
@@ -317,9 +316,9 @@ void testCancelOuterFutureAfterStart() throws Exception {
         // i.e. The future from executor.submit() has been run by the ScheduledExecutor
         assertTrue(future.getAttemptSettings().getAttemptCount() > 0);
         assertTrue(future.getAttemptSettings().getAttemptCount() < 4);
+      } finally {
+        localExecutor.shutdownNow();
       }
-    } finally {
-      localExecutor.shutdownNow();
     }
   }