[8.17] Avoid hoarding cluster state references during rollover (elastic#124107) (elastic#124267)

# Backport

This will backport the following commits from `main` to `8.17`:  -
[Avoid hoarding cluster state references during rollover
(elastic#124107)](elastic#124107)

<!--- Backport version: 9.6.4 -->

### Questions ? Please refer to the [Backport tool
documentation](https://github.com/sorenlouv/backport)

Author: nielsbauman

docs/changelog/124107.yaml

@@ -0,0 +1,6 @@
+pr: 124107
+summary: Avoid hoarding cluster state references during rollover
+area: Indices APIs
+type: bug
+issues:
+ - 123893

docs/changelog/124267.yaml

@@ -0,0 +1,5 @@
+pr: 124267
+summary: "[8.17] Avoid hoarding cluster state references during rollover"
+area: Indices APIs
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/LazyRolloverAction.java

@@ -30,7 +30,6 @@
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.CancellableTask;
@@ -178,7 +177,7 @@ record LazyRolloverExecutor(
         @Override
         public ClusterState execute(BatchExecutionContext<LazyRolloverTask> batchExecutionContext) {
             final var listener = new AllocationActionMultiListener<RolloverResponse>(threadPool.getThreadContext());
-            final var results = new ArrayList<MetadataRolloverService.RolloverResult>(batchExecutionContext.taskContexts().size());
+            final var results = new ArrayList<String>(batchExecutionContext.taskContexts().size());
             var state = batchExecutionContext.initialState();
             Map<RolloverRequest, List<TaskContext<LazyRolloverTask>>> groupedRequests = new HashMap<>();
             for (final var taskContext : batchExecutionContext.taskContexts()) {
@@ -198,14 +197,7 @@ public ClusterState execute(BatchExecutionContext<LazyRolloverTask> batchExecuti
 
             if (state != batchExecutionContext.initialState()) {
                 var reason = new StringBuilder();
-                Strings.collectionToDelimitedStringWithLimit(
-                    (Iterable<String>) () -> Iterators.map(results.iterator(), t -> t.sourceIndexName() + "->" + t.rolloverIndexName()),
-                    ",",
-                    "lazy bulk rollover [",
-                    "]",
-                    1024,
-                    reason
-                );
+                Strings.collectionToDelimitedStringWithLimit(results, ",", "lazy bulk rollover [", "]", 1024, reason);
                 try (var ignored = batchExecutionContext.dropHeadersContext()) {
                     state = allocationService.reroute(state, reason.toString(), listener.reroute());
                 }
@@ -218,7 +210,7 @@ public ClusterState execute(BatchExecutionContext<LazyRolloverTask> batchExecuti
         public ClusterState executeTask(
             ClusterState currentState,
             RolloverRequest rolloverRequest,
-            List<MetadataRolloverService.RolloverResult> results,
+            ArrayList<String> results,
             List<TaskContext<LazyRolloverTask>> rolloverTaskContexts,
             AllocationActionMultiListener<RolloverResponse> allocationActionMultiListener
         ) throws Exception {
@@ -248,7 +240,7 @@ public ClusterState executeTask(
                 null,
                 rolloverRequest.targetsFailureStore()
             );
-            results.add(rolloverResult);
+            results.add(rolloverResult.sourceIndexName() + "->" + rolloverResult.rolloverIndexName());
             logger.trace("lazy rollover result [{}]", rolloverResult);
 
             final var rolloverIndexName = rolloverResult.rolloverIndexName();

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverAction.java

@@ -45,7 +45,6 @@
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Nullable;
@@ -455,7 +454,7 @@ record RolloverExecutor(
         @Override
         public ClusterState execute(BatchExecutionContext<RolloverTask> batchExecutionContext) {
             final var listener = new AllocationActionMultiListener<RolloverResponse>(threadPool.getThreadContext());
-            final var results = new ArrayList<MetadataRolloverService.RolloverResult>(batchExecutionContext.taskContexts().size());
+            final var results = new ArrayList<String>(batchExecutionContext.taskContexts().size());
             var state = batchExecutionContext.initialState();
             for (final var taskContext : batchExecutionContext.taskContexts()) {
                 try (var ignored = taskContext.captureResponseHeaders()) {
@@ -467,14 +466,7 @@ public ClusterState execute(BatchExecutionContext<RolloverTask> batchExecutionCo
 
             if (state != batchExecutionContext.initialState()) {
                 var reason = new StringBuilder();
-                Strings.collectionToDelimitedStringWithLimit(
-                    (Iterable<String>) () -> Iterators.map(results.iterator(), t -> t.sourceIndexName() + "->" + t.rolloverIndexName()),
-                    ",",
-                    "bulk rollover [",
-                    "]",
-                    1024,
-                    reason
-                );
+                Strings.collectionToDelimitedStringWithLimit(results, ",", "bulk rollover [", "]", 1024, reason);
                 try (var ignored = batchExecutionContext.dropHeadersContext()) {
                     state = allocationService.reroute(state, reason.toString(), listener.reroute());
                 }
@@ -486,7 +478,7 @@ public ClusterState execute(BatchExecutionContext<RolloverTask> batchExecutionCo
 
         public ClusterState executeTask(
             ClusterState currentState,
-            List<MetadataRolloverService.RolloverResult> results,
+            ArrayList<String> results,
             TaskContext<RolloverTask> rolloverTaskContext,
             AllocationActionMultiListener<RolloverResponse> allocationActionMultiListener
         ) throws Exception {
@@ -552,7 +544,7 @@ public ClusterState executeTask(
                     rolloverTask.autoShardingResult(),
                     rolloverRequest.targetsFailureStore()
                 );
-                results.add(rolloverResult);
+                results.add(rolloverResult.sourceIndexName() + "->" + rolloverResult.rolloverIndexName());
                 logger.trace("rollover result [{}]", rolloverResult);
 
                 final var rolloverIndexName = rolloverResult.rolloverIndexName();