Make some constant SubscribableListener instances cheaper

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -157,15 +157,15 @@ public TransportClusterStatsAction(
     protected SubscribableListener<AdditionalStats> createActionContext(Task task, ClusterStatsRequest request) {
         assert task instanceof CancellableTask;
         final var cancellableTask = (CancellableTask) task;
-        final var additionalStatsListener = new SubscribableListener<AdditionalStats>();
         if (request.isRemoteStats() == false) {
+            final var additionalStatsListener = new SubscribableListener<AdditionalStats>();
             final AdditionalStats additionalStats = new AdditionalStats();
             additionalStats.compute(cancellableTask, request, additionalStatsListener);
+            return additionalStatsListener;
         } else {
             // For remote stats request, we don't need to compute anything
-            additionalStatsListener.onResponse(null);
+            return SubscribableListener.nullSuccess();
         }
-        return additionalStatsListener;
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/support/SubscribableListener.java

@@ -110,20 +110,6 @@ public SubscribableListener() {
         this(EMPTY);
     }
 
-    /**
-     * Create a {@link SubscribableListener} which has already succeeded with the given result.
-     */
-    public static <T> SubscribableListener<T> newSucceeded(T result) {
-        return new SubscribableListener<>(new SuccessResult<>(result));
-    }
-
-    /**
-     * Create a {@link SubscribableListener} which has already failed with the given exception.
-     */
-    public static <T> SubscribableListener<T> newFailed(Exception exception) {
-        return new SubscribableListener<>(new FailureResult(exception, exception));
-    }
-
     /**
      * Create a {@link SubscribableListener}, fork a computation to complete it, and return the listener. If the forking itself throws an
      * exception then the exception is caught and fed to the returned listener.
@@ -586,4 +572,33 @@ private Runnable scheduleTimeout(TimeValue timeout, ThreadPool threadPool, Execu
     private Object compareAndExchangeState(Object expectedValue, Object newValue) {
         return VH_STATE_FIELD.compareAndExchange(this, expectedValue, newValue);
     }
+
+    @SuppressWarnings("rawtypes")
+    private static final SubscribableListener NULL_SUCCESS = newSucceeded(null);
+
+    /**
+     * Same as {@link #newSucceeded(Object)} but always returns the same instance with result value {@code null}.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> SubscribableListener<T> nullSuccess() {
+        return NULL_SUCCESS;
+    }
+
+    /**
+     * Create a {@link SubscribableListener} which has already succeeded with the given result.
+     */
+    public static <T> SubscribableListener<T> newSucceeded(T result) {
+        var res = new SubscribableListener<T>();
+        VH_STATE_FIELD.setRelease(res, new SuccessResult<>(result));
+        return res;
+    }
+
+    /**
+     * Create a {@link SubscribableListener} which has already failed with the given exception.
+     */
+    public static <T> SubscribableListener<T> newFailed(Exception exception) {
+        var res = new SubscribableListener<T>();
+        VH_STATE_FIELD.setRelease(res, new FailureResult(exception, exception));
+        return res;
+    }
 }

server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java

@@ -243,7 +243,7 @@ protected void doClose() {}
      * Kind of a hack tbh, we can't be sure the shard locks are fully released when this is completed so there's all sorts of retries and
      * other lenience to handle that. It'd be better to wait for the shard locks to be released and then delete the data. See #74149.
      */
-    private volatile SubscribableListener<Void> lastClusterStateShardsClosedListener = SubscribableListener.newSucceeded(null);
+    private volatile SubscribableListener<Void> lastClusterStateShardsClosedListener = SubscribableListener.nullSuccess();
 
     @Nullable // if not currently applying a cluster state
     private RefCountingListener currentClusterStateShardsClosedListeners;
@@ -397,7 +397,7 @@ private void deleteIndices(final ClusterChangedEvent event) {
                 );
             } else if (project.isPresent() && project.get().hasIndex(index)) {
                 // The deleted index was part of the previous cluster state, but not loaded on the local node
-                indexServiceClosedListener = SubscribableListener.newSucceeded(null);
+                indexServiceClosedListener = SubscribableListener.nullSuccess();
                 final IndexMetadata metadata = project.get().index(index);
                 indexSettings = new IndexSettings(metadata, settings);
                 indicesService.deleteUnassignedIndex("deleted index was not assigned to local node", metadata, state);
@@ -411,7 +411,7 @@ private void deleteIndices(final ClusterChangedEvent event) {
                 // previous cluster state is not initialized/recovered.
                 assert state.metadata().projects().values().stream().anyMatch(p -> p.indexGraveyard().containsIndex(index))
                     || previousState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK);
-                indexServiceClosedListener = SubscribableListener.newSucceeded(null);
+                indexServiceClosedListener = SubscribableListener.nullSuccess();
                 final IndexMetadata metadata = indicesService.verifyIndexIsDeleted(index, event.state());
                 if (metadata != null) {
                     indexSettings = new IndexSettings(metadata, settings);

test/framework/src/main/java/org/elasticsearch/indices/recovery/RecoveryClusterStateDelayListeners.java

@@ -64,7 +64,7 @@ public SubscribableListener<Void> getClusterStateDelayListener(long clusterState
                 refCounted.decRef();
             }
         } else {
-            return SubscribableListener.newSucceeded(null);
+            return SubscribableListener.nullSuccess();
         }
     }
 

test/framework/src/main/java/org/elasticsearch/test/TestCluster.java

@@ -127,46 +127,40 @@ private void deleteTemplates(Set<String> excludeTemplates, ActionListener<Void>
             l -> client().execute(GetComponentTemplateAction.INSTANCE, new GetComponentTemplateAction.Request(TEST_REQUEST_TIMEOUT, "*"), l)
         );
 
-        SubscribableListener
-
-            // dummy start step for symmetry
-            .newSucceeded(null)
-
-            // delete composable templates
-            .<GetComposableIndexTemplateAction.Response>andThen(getComposableTemplates::addListener)
-            .<AcknowledgedResponse>andThen((l, r) -> {
-                var templates = r.indexTemplates()
-                    .keySet()
-                    .stream()
-                    .filter(template -> excludeTemplates.contains(template) == false)
-                    .toArray(String[]::new);
-                if (templates.length == 0) {
-                    l.onResponse(AcknowledgedResponse.TRUE);
-                } else {
-                    var request = new TransportDeleteComposableIndexTemplateAction.Request(templates);
-                    client().execute(TransportDeleteComposableIndexTemplateAction.TYPE, request, l);
-                }
-            })
-            .andThenAccept(ElasticsearchAssertions::assertAcked)
-
-            // then delete component templates
-            .<GetComponentTemplateAction.Response>andThen(getComponentTemplates::addListener)
-            .<AcknowledgedResponse>andThen((l, response) -> {
-                var componentTemplates = response.getComponentTemplates()
-                    .keySet()
-                    .stream()
-                    .filter(template -> excludeTemplates.contains(template) == false)
-                    .toArray(String[]::new);
-                if (componentTemplates.length == 0) {
-                    l.onResponse(AcknowledgedResponse.TRUE);
-                } else {
-                    client().execute(
-                        TransportDeleteComponentTemplateAction.TYPE,
-                        new TransportDeleteComponentTemplateAction.Request(componentTemplates),
-                        l
-                    );
-                }
-            })
+        // dummy start step for symmetry
+        SubscribableListener.nullSuccess()
+        // delete composable templates
+        .<GetComposableIndexTemplateAction.Response>andThen(getComposableTemplates::addListener).<AcknowledgedResponse>andThen((l, r) -> {
+            var templates = r.indexTemplates()
+                .keySet()
+                .stream()
+                .filter(template -> excludeTemplates.contains(template) == false)
+                .toArray(String[]::new);
+            if (templates.length == 0) {
+                l.onResponse(AcknowledgedResponse.TRUE);
+            } else {
+                var request = new TransportDeleteComposableIndexTemplateAction.Request(templates);
+                client().execute(TransportDeleteComposableIndexTemplateAction.TYPE, request, l);
+            }
+        }).andThenAccept(ElasticsearchAssertions::assertAcked)
+
+        // then delete component templates
+        .<GetComponentTemplateAction.Response>andThen(getComponentTemplates::addListener).<AcknowledgedResponse>andThen((l, response) -> {
+            var componentTemplates = response.getComponentTemplates()
+                .keySet()
+                .stream()
+                .filter(template -> excludeTemplates.contains(template) == false)
+                .toArray(String[]::new);
+            if (componentTemplates.length == 0) {
+                l.onResponse(AcknowledgedResponse.TRUE);
+            } else {
+                client().execute(
+                    TransportDeleteComponentTemplateAction.TYPE,
+                    new TransportDeleteComponentTemplateAction.Request(componentTemplates),
+                    l
+                );
+            }
+        })
             .andThenAccept(ElasticsearchAssertions::assertAcked)
 
             // and finish

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/Operator.java

@@ -92,7 +92,7 @@ default IsBlockedResult isBlocked() {
         return NOT_BLOCKED;
     }
 
-    IsBlockedResult NOT_BLOCKED = new IsBlockedResult(SubscribableListener.newSucceeded(null), "not blocked");
+    IsBlockedResult NOT_BLOCKED = new IsBlockedResult(SubscribableListener.nullSuccess(), "not blocked");
 
     /**
      * A factory for creating intermediate operators.

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java

@@ -357,7 +357,7 @@ public void fetchPageAsync(boolean allSourcesFinished, ActionListener<ExchangeRe
             }
             doFetchPageAsync(false, ActionListener.wrap(r -> {
                 if (r.finished()) {
-                    completionListenerRef.compareAndSet(null, SubscribableListener.newSucceeded(null));
+                    completionListenerRef.compareAndSet(null, SubscribableListener.nullSuccess());
                 }
                 listener.onResponse(r);
             }, e -> close(ActionListener.running(() -> listener.onFailure(e)))));

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/interceptor/BulkShardRequestInterceptor.java

@@ -81,6 +81,6 @@ public SubscribableListener<Void> intercept(
                 }
             }
         }
-        return SubscribableListener.newSucceeded(null);
+        return SubscribableListener.nullSuccess();
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/interceptor/DlsFlsLicenseRequestInterceptor.java

@@ -101,6 +101,6 @@ public SubscribableListener<Void> intercept(
                 }
             }
         }
-        return SubscribableListener.newSucceeded(null);
+        return SubscribableListener.nullSuccess();
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/interceptor/FieldAndDocumentLevelSecurityRequestInterceptor.java

@@ -77,7 +77,7 @@ && supports(indicesRequest)
                 return listener;
             }
         }
-        return SubscribableListener.newSucceeded(null);
+        return SubscribableListener.nullSuccess();
     }
 
     abstract void disableFeatures(

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/interceptor/IndicesAliasesRequestInterceptor.java

@@ -125,7 +125,7 @@ public SubscribableListener<Void> intercept(
             );
             return listener;
         } else {
-            return SubscribableListener.newSucceeded(null);
+            return SubscribableListener.nullSuccess();
         }
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/interceptor/ResizeRequestInterceptor.java

@@ -103,7 +103,7 @@ public SubscribableListener<Void> intercept(
             );
             return listener;
         } else {
-            return SubscribableListener.newSucceeded(null);
+            return SubscribableListener.nullSuccess();
         }
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/interceptor/SearchRequestCacheDisablingInterceptor.java

@@ -49,7 +49,7 @@ && hasRemoteIndices(searchRequest)
                 searchRequest.requestCache(false);
             }
         }
-        return SubscribableListener.newSucceeded(null);
+        return SubscribableListener.nullSuccess();
     }
 
     // package private for test

x-pack/plugin/snapshot-repo-test-kit/src/internalClusterTest/java/org/elasticsearch/repositories/blobstore/testkit/integrity/RepositoryVerifyIntegrityIT.java

@@ -191,7 +191,7 @@ public void testTaskStatus() throws IOException {
                         assertEquals(0L, status.indexSnapshotsVerified());
                         assertEquals(0L, status.blobsVerified());
                         assertEquals(0L, status.blobBytesVerified());
-                        yield SubscribableListener.newSucceeded(null);
+                        yield SubscribableListener.nullSuccess();
                     }
                     case INDEX_RESTORABILITY -> {
                         // several of these chunks might arrive concurrently; we want to verify the task status before processing any of
@@ -210,7 +210,7 @@ public void testTaskStatus() throws IOException {
                             assertEquals(0L, status.indicesVerified());
                         });
                     }
-                    case SNAPSHOT_INFO -> SubscribableListener.newSucceeded(null);
+                    case SNAPSHOT_INFO -> SubscribableListener.nullSuccess();
                     case ANOMALY -> fail(null, "should not see anomalies");
                 };
 

x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/integrity/RepositoryIntegrityVerifier.java

@@ -798,7 +798,7 @@ protected InputStream openSlice(int slice) throws IOException {
                     })));
                 } else {
                     blobBytesVerified.addAndGet(fileInfo.length());
-                    return SubscribableListener.newSucceeded(null);
+                    return SubscribableListener.nullSuccess();
                 }
             });
         }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportSetTransformUpgradeModeAction.java

@@ -131,7 +131,7 @@ private void unassignTransforms(ClusterState state, ActionListener<Void> listene
 
         // chain each call one at a time
         // because that is what we are doing for ML, and that is all that is supported in the persistentTasksClusterService (for now)
-        SubscribableListener<PersistentTasksCustomMetadata.PersistentTask<?>> chainListener = SubscribableListener.newSucceeded(null);
+        SubscribableListener<PersistentTasksCustomMetadata.PersistentTask<?>> chainListener = SubscribableListener.nullSuccess();
         for (var task : transformTasks) {
             @FixForMultiProject
             final var projectId = Metadata.DEFAULT_PROJECT_ID;