Repository: redis/lettuce
Pull request #3040 (2024-11-01 08:26:10)

Add defensive copy for Futures allOf() method (#2943)
Backport #2943 to the 6.4.x branch

---------

<!--
Thank you for proposing a pull request. This template will guide you through the essential steps necessary for a pull request.
-->
Make sure that:

- [x] You have read the [contribution guidelines](https://github.com/lettuce-io/lettuce-core/blob/main/.github/CONTRIBUTING.md).
- [x] You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request.
- [x] You applied code formatting rules using the `mvn formatter:format` target. Don’t submit any formatting related changes.
- [x] You submit test cases (unit or integration tests) that back your changes.

 <!--
Great! Live long and prosper.
-->



Patch:
diff --git a/src/main/java/io/lettuce/core/internal/Futures.java b/src/main/java/io/lettuce/core/internal/Futures.java
index 8c3fcf69f..f8276f8b3 100644
--- a/src/main/java/io/lettuce/core/internal/Futures.java
+++ b/src/main/java/io/lettuce/core/internal/Futures.java
@@ -12,6 +12,7 @@ import io.netty.channel.ChannelFuture;
  * without further notice.
  *
  * @author Mark Paluch
+ * @author jinkshower
  * @since 5.1
  */
 public abstract class Futures {
@@ -21,7 +22,7 @@ public abstract class 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 +33,11 @@ public abstract class Futures {
 
         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();
         }
 


Test patch:
diff --git a/src/test/java/io/lettuce/core/internal/FuturesUnitTests.java b/src/test/java/io/lettuce/core/internal/FuturesUnitTests.java
index 1e3ec4178..d7c1868d4 100644
--- a/src/test/java/io/lettuce/core/internal/FuturesUnitTests.java
+++ b/src/test/java/io/lettuce/core/internal/FuturesUnitTests.java
@@ -4,7 +4,13 @@ import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -16,6 +22,7 @@ import io.lettuce.core.RedisCommandInterruptedException;
  * Unit tests for {@link Futures}.
  *
  * @author Mark Paluch
+ * @author Tihomir Mateev
  */
 class FuturesUnitTests {
 
@@ -56,4 +63,33 @@ class FuturesUnitTests {
         assertThat(Thread.currentThread().isInterrupted()).isTrue();
     }
 
+    @Test
+    void allOfShouldNotThrow() throws InterruptedException {
+        int threadCount = 100;
+        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+        List<Throwable> issues = new ArrayList<>();
+        List<CompletableFuture<Void>> futures = Collections.synchronizedList(new ArrayList<>());
+        // Submit multiple threads to perform concurrent operations
+        CountDownLatch latch = new CountDownLatch(threadCount);
+        for (int i = 0; i < threadCount; i++) {
+            executorService.submit(() -> {
+                try {
+                    for (int y = 0; y < 1000; y++) {
+                        futures.add(new CompletableFuture<>());
+                    }
+
+                    Futures.allOf(futures);
+                } catch (Exception e) {
+                    issues.add(e);
+                } finally {
+                    latch.countDown();
+                }
+            });
+        }
+
+        // wait for all threads to complete
+        latch.await();
+        assertThat(issues).doesNotHaveAnyElementsOfTypes(ArrayIndexOutOfBoundsException.class);
+    }
+
 }