CR: comments

Author: original-brownbear

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

@@ -266,14 +266,21 @@ public final boolean isDone() {
         return isDone(state);
     }
 
+    /**
+     * @return return {@code true} if and only if this listener is done and has been completed successfully
+     */
+    public final boolean isSuccess() {
+        return state instanceof SuccessResult;
+    }
+
     /**
      * @return the result with which this listener completed successfully, or throw the exception with which it failed.
      *
      * @throws AssertionError if this listener is not complete yet and assertions are enabled.
      * @throws IllegalStateException if this listener is not complete yet and assertions are disabled.
      */
     @SuppressWarnings({ "unchecked", "rawtypes" })
-    public final T rawResult() throws Exception {
+    protected final T rawResult() throws Exception {
         final Object currentState = state;
         if (currentState instanceof SuccessResult result) {
             return (T) result.result();

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/xpack/security/profile/ProfileCancellationIntegTests.java

@@ -22,7 +22,6 @@
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.shard.SearchOperationListener;
 import org.elasticsearch.plugins.ActionPlugin;
@@ -414,7 +413,7 @@ public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
                     AsyncSupplier<ResolvedIndices> indicesAsyncSupplier,
                     ProjectMetadata metadata
                 ) {
-                    return ListenableFuture.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
+                    return SubscribableListener.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
                 }
 
                 @Override

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

@@ -486,13 +486,13 @@ private void authorizeAction(
             final AsyncSupplier<ResolvedIndices> resolvedIndicesAsyncSupplier = new CachingAsyncSupplier<>(() -> {
                 if (request instanceof SearchRequest searchRequest && searchRequest.pointInTimeBuilder() != null) {
                     var resolvedIndices = indicesAndAliasesResolver.resolvePITIndices(searchRequest);
-                    return ListenableFuture.newSucceeded(resolvedIndices);
+                    return SubscribableListener.newSucceeded(resolvedIndices);
                 }
                 final ResolvedIndices resolvedIndices = indicesAndAliasesResolver.tryResolveWithoutWildcards(action, request);
                 if (resolvedIndices != null) {
-                    return ListenableFuture.newSucceeded(resolvedIndices);
+                    return SubscribableListener.newSucceeded(resolvedIndices);
                 } else {
-                    final SubscribableListener<ResolvedIndices> resolvedIndicesListener = new SubscribableListener<ResolvedIndices>();
+                    final SubscribableListener<ResolvedIndices> resolvedIndicesListener = new SubscribableListener<>();
                     authzEngine.loadAuthorizedIndices(
                         requestInfo,
                         authzInfo,
@@ -637,29 +637,21 @@ private void runRequestInterceptors(
             final Iterator<RequestInterceptor> requestInterceptorIterator = requestInterceptors.iterator();
             while (requestInterceptorIterator.hasNext()) {
                 var res = requestInterceptorIterator.next().intercept(requestInfo, authorizationEngine, authorizationInfo);
-                if (res.isDone() == false) {
+                if (res.isSuccess() == false) {
                     res.addListener(new DelegatingActionListener<>(listener) {
                         @Override
                         public void onResponse(Void unused) {
                             if (requestInterceptorIterator.hasNext()) {
-                                var nestedRest = requestInterceptorIterator.next()
-                                    .intercept(requestInfo, authorizationEngine, authorizationInfo);
-                                if (nestedRest.isDone() == false) {
-                                    nestedRest.addListener(this);
-                                }
+                                requestInterceptorIterator.next()
+                                    .intercept(requestInfo, authorizationEngine, authorizationInfo)
+                                    .addListener(this);
                             } else {
-                                listener.onResponse(null);
+                                delegate.onResponse(null);
                             }
                         }
                     });
                     return;
                 }
-                try {
-                    res.rawResult();
-                } catch (Exception e) {
-                    listener.onFailure(e);
-                    return;
-                }
             }
             listener.onResponse(null);
         }
@@ -887,7 +879,7 @@ private void authorizeBulkItems(
                 authzEngine.authorizeIndexAction(
                     bulkItemInfo,
                     authzInfo,
-                    () -> ListenableFuture.newSucceeded(new ResolvedIndices(new ArrayList<>(indices), Collections.emptyList())),
+                    () -> SubscribableListener.newSucceeded(new ResolvedIndices(new ArrayList<>(indices), Collections.emptyList())),
                     projectMetadata
                 )
                     .addListener(

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

@@ -41,7 +41,6 @@
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.CachedSupplier;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.shard.ShardId;
@@ -328,12 +327,12 @@ public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
         try {
             role = ensureRBAC(authorizationInfo).getRole();
         } catch (Exception e) {
-            return ListenableFuture.newFailed(e);
+            return SubscribableListener.newFailed(e);
         }
         if (TransportActionProxy.isProxyAction(action) || shouldAuthorizeIndexActionNameOnly(action, request)) {
             // we've already validated that the request is a proxy request so we can skip that but we still
             // need to validate that the action is allowed and then move on
-            return ListenableFuture.newSucceeded(
+            return SubscribableListener.newSucceeded(
                 role.checkIndicesAction(action) ? IndexAuthorizationResult.EMPTY : IndexAuthorizationResult.DENIED
             );
         } else if (request instanceof IndicesRequest == false) {
@@ -353,7 +352,7 @@ public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
                 // index and if they cannot, we can fail the request early before we allow the execution of the action and in
                 // turn the shard actions
                 if (TransportSearchScrollAction.TYPE.name().equals(action)) {
-                    final ListenableFuture<IndexAuthorizationResult> listener = new ListenableFuture<>();
+                    final SubscribableListener<IndexAuthorizationResult> listener = new SubscribableListener<>();
                     ActionRunnable.supply(listener.delegateFailureAndWrap((l, parsedScrollId) -> {
                         if (parsedScrollId.hasLocalIndices()) {
                             l.onResponse(
@@ -373,42 +372,42 @@ public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
                     // The DLS/FLS permissions are used inside the {@code DirectoryReader} that {@code SecurityIndexReaderWrapper}
                     // built while handling the initial search request. In addition, for consistency, the DLS/FLS permissions from
                     // the originating search request are attached to the thread context upon validating the scroll.
-                    return ListenableFuture.newSucceeded(IndexAuthorizationResult.EMPTY);
+                    return SubscribableListener.newSucceeded(IndexAuthorizationResult.EMPTY);
                 }
             } else if (isAsyncRelatedAction(action)) {
                 if (SubmitAsyncSearchAction.NAME.equals(action)) {
                     // authorize submit async search but don't fill in the DLS/FLS permissions
                     // the `null` IndicesAccessControl parameter indicates that this action has *not* determined
                     // which DLS/FLS controls should be applied to this action
-                    return ListenableFuture.newSucceeded(IndexAuthorizationResult.EMPTY);
+                    return SubscribableListener.newSucceeded(IndexAuthorizationResult.EMPTY);
                 } else {
                     // async-search actions other than submit have a custom security layer that checks if the current user is
                     // the same as the user that submitted the original request so no additional checks are needed here.
-                    return ListenableFuture.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
+                    return SubscribableListener.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
                 }
             } else if (action.equals(TransportClosePointInTimeAction.TYPE.name())) {
-                return ListenableFuture.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
+                return SubscribableListener.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
             } else {
                 assert false
                     : "only scroll and async-search related requests are known indices api that don't "
                         + "support retrieving the indices they relate to";
-                return ListenableFuture.newFailed(
+                return SubscribableListener.newFailed(
                     new IllegalStateException(
                         "only scroll and async-search related requests are known indices "
                             + "api that don't support retrieving the indices they relate to"
                     )
                 );
             }
         } else if (isChildActionAuthorizedByParentOnLocalNode(requestInfo, authorizationInfo)) {
-            return ListenableFuture.newSucceeded(
+            return SubscribableListener.newSucceeded(
                 new IndexAuthorizationResult(requestInfo.getOriginatingAuthorizationContext().getIndicesAccessControl())
             );
         } else if (PreAuthorizationUtils.shouldPreAuthorizeChildByParentAction(requestInfo, authorizationInfo)) {
             // We only pre-authorize child actions if DLS/FLS is not configured,
             // hence we can allow here access for all requested indices.
-            return ListenableFuture.newSucceeded(new IndexAuthorizationResult(IndicesAccessControl.allowAll()));
+            return SubscribableListener.newSucceeded(new IndexAuthorizationResult(IndicesAccessControl.allowAll()));
         } else if (allowsRemoteIndices(request) || role.checkIndicesAction(action)) {
-            final ListenableFuture<IndexAuthorizationResult> listener = new ListenableFuture<>();
+            final SubscribableListener<IndexAuthorizationResult> listener = new SubscribableListener<>();
             indicesAsyncSupplier.getAsync().addListener(listener.delegateFailureAndWrap((delegateListener, resolvedIndices) -> {
                 assert resolvedIndices.isEmpty() == false
                     : "every indices request needs to have its indices set thus the resolved indices must not be empty";
@@ -446,7 +445,7 @@ public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
             }));
             return listener;
         } else {
-            return ListenableFuture.newSucceeded(IndexAuthorizationResult.DENIED);
+            return SubscribableListener.newSucceeded(IndexAuthorizationResult.DENIED);
         }
     }
 

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

@@ -10,7 +10,6 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
 import org.elasticsearch.action.support.SubscribableListener;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.license.XPackLicenseState;
@@ -99,7 +98,7 @@ public SubscribableListener<Void> intercept(
                     list.addAll(toMerge);
                     return list;
                 }));
-            final SubscribableListener<Void> listener = new ListenableFuture<>();
+            final SubscribableListener<Void> listener = new SubscribableListener<>();
             authorizationEngine.validateIndexPermissionsAreSubset(
                 requestInfo,
                 authorizationInfo,
@@ -126,7 +125,7 @@ public SubscribableListener<Void> intercept(
             );
             return listener;
         } else {
-            return ListenableFuture.newSucceeded(null);
+            return SubscribableListener.newSucceeded(null);
         }
     }
 }

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/RBACEngineTests.java

@@ -20,6 +20,7 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.ElasticsearchClient;
 import org.elasticsearch.cluster.metadata.DataStream;
@@ -31,7 +32,6 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexVersion;
@@ -1971,7 +1971,7 @@ private void authorizeIndicesAction(
         final ResolvedIndices resolvedIndices = new ResolvedIndices(List.of(indices), List.of());
         final TransportRequest searchRequest = new SearchRequest(indices);
         final RequestInfo requestInfo = createRequestInfo(searchRequest, action, parentAuthorization);
-        final AsyncSupplier<ResolvedIndices> indicesAsyncSupplier = () -> ListenableFuture.newSucceeded(resolvedIndices);
+        final AsyncSupplier<ResolvedIndices> indicesAsyncSupplier = () -> SubscribableListener.newSucceeded(resolvedIndices);
 
         Metadata.Builder metadata = Metadata.builder();
         Stream.of(indices)