Add defensive copy

Closes #2935

src/main/java/io/lettuce/core/internal/Futures.java

@@ -21,7 +21,7 @@ private Futures() {
     }
 
     /**
-     * Create a composite {@link CompletableFuture} is composed from the given {@code stages}.
+     * Create a composite {@link CompletableFuture} that is composed of the given {@code stages}.
      *
      * @param stages must not be {@code null}.
      * @return the composed {@link CompletableFuture}.
@@ -32,10 +32,11 @@ public static CompletableFuture<Void> allOf(Collection<? extends CompletionStage
 
         LettuceAssert.notNull(stages, "Futures must not be null");
 
-        CompletableFuture[] futures = new CompletableFuture[stages.size()];
+        CompletionStage[] copies = stages.toArray(new CompletionStage[0]);
+        CompletableFuture[] futures = new CompletableFuture[copies.length];
 
         int index = 0;
-        for (CompletionStage<?> stage : stages) {
+        for (CompletionStage<?> stage : copies) {
             futures[index++] = stage.toCompletableFuture();
         }
 

src/test/java/io/lettuce/core/internal/FuturesUnitTests.java

@@ -3,8 +3,12 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -56,4 +60,30 @@ void awaitAllShouldSetInterruptedBit() {
         assertThat(Thread.currentThread().isInterrupted()).isTrue();
     }
 
+    @Test
+    void allOfShouldNotThrow() throws InterruptedException {
+        List<CompletionStage<?>> stages = new ArrayList<>();
+
+        for (int i = 0; i < 50; i++) {
+            stages.add(new CompletableFuture<>());
+        }
+
+        Thread thread1 = new Thread(() -> assertDoesNotThrow(() -> {
+            for (int i = 0; i < 10; i++) {
+                Futures.allOf(stages);
+            }
+        }));
+
+        Thread thread2 = new Thread(() -> {
+            for (int i = 0; i < 10; i++) {
+                stages.remove(0);
+            }
+        });
+
+        thread2.start();
+        thread1.start();
+
+        thread2.join();
+        thread1.join();
+    }
 }