Tidy up GroupedActionListener (#39633)

Today the `GroupedActionListener` accepts a `defaults` parameter but all
callers pass an empty list. Also it is permitted to pass an empty group but
this is trappy because the delegated listener is never be called in that case.
This commit removes the `defaults` parameter and forbids an empty group.

Author: DaveCTurner

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

@@ -29,8 +29,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
- * An action listener that delegates it's results to another listener once
- * it has received one or more failures or N results. This allows synchronous
+ * An action listener that delegates its results to another listener once
+ * it has received N results (either successes or failures). This allows synchronous
  * tasks to be forked off in a loop with the same listener and respond to a
  * higher level listener once all tasks responded.
  */
@@ -39,20 +39,20 @@ public final class GroupedActionListener<T> implements ActionListener<T> {
     private final AtomicInteger pos = new AtomicInteger();
     private final AtomicArray<T> results;
     private final ActionListener<Collection<T>> delegate;
-    private final Collection<T> defaults;
     private final AtomicReference<Exception> failure = new AtomicReference<>();
 
     /**
      * Creates a new listener
      * @param delegate the delegate listener
      * @param groupSize the group size
      */
-    public GroupedActionListener(ActionListener<Collection<T>> delegate, int groupSize,
-                                 Collection<T> defaults) {
+    public GroupedActionListener(ActionListener<Collection<T>> delegate, int groupSize) {
+        if (groupSize <= 0) {
+            throw new IllegalArgumentException("groupSize must be greater than 0 but was " + groupSize);
+        }
         results = new AtomicArray<>(groupSize);
         countDown = new CountDown(groupSize);
         this.delegate = delegate;
-        this.defaults = defaults;
     }
 
     @Override
@@ -63,7 +63,6 @@ public void onResponse(T element) {
                 delegate.onFailure(failure.get());
             } else {
                 List<T> collect = this.results.asList();
-                collect.addAll(defaults);
                 delegate.onResponse(Collections.unmodifiableList(collect));
             }
         }

server/src/test/java/org/elasticsearch/action/support/GroupedActionListenerTests.java

@@ -51,10 +51,7 @@ public void onFailure(Exception e) {
         };
         final int groupSize = randomIntBetween(10, 1000);
         AtomicInteger count = new AtomicInteger();
-        Collection<Integer> defaults = randomBoolean() ? Collections.singletonList(-1) :
-            Collections.emptyList();
-        GroupedActionListener<Integer> listener = new GroupedActionListener<>(result, groupSize,
-            defaults);
+        GroupedActionListener<Integer> listener = new GroupedActionListener<>(result, groupSize);
         int numThreads = randomIntBetween(2, 5);
         Thread[] threads = new Thread[numThreads];
         CyclicBarrier barrier = new CyclicBarrier(numThreads);
@@ -65,7 +62,7 @@ public void onFailure(Exception e) {
                 } catch (Exception e) {
                     throw new AssertionError(e);
                 }
-                int c = 0;
+                int c;
                 while((c = count.incrementAndGet()) <= groupSize) {
                     listener.onResponse(c-1);
                 }
@@ -78,10 +75,9 @@ public void onFailure(Exception e) {
         assertNotNull(resRef.get());
         ArrayList<Integer> list = new ArrayList<>(resRef.get());
         Collections.sort(list);
-        int expectedSize = groupSize + defaults.size();
-        assertEquals(expectedSize, resRef.get().size());
-        int expectedValue = defaults.isEmpty() ? 0 : -1;
-        for (int i = 0; i < expectedSize; i++) {
+        assertEquals(groupSize, resRef.get().size());
+        int expectedValue = 0;
+        for (int i = 0; i < groupSize; i++) {
             assertEquals(Integer.valueOf(expectedValue++), list.get(i));
         }
     }
@@ -101,9 +97,8 @@ public void onFailure(Exception e) {
                 excRef.set(e);
             }
         };
-        Collection<Integer> defaults = randomBoolean() ? Collections.singletonList(-1) : Collections.emptyList();
         int size = randomIntBetween(3, 4);
-        GroupedActionListener<Integer> listener = new GroupedActionListener<>(result, size, defaults);
+        GroupedActionListener<Integer> listener = new GroupedActionListener<>(result, size);
         listener.onResponse(0);
         IOException ioException = new IOException();
         RuntimeException rtException = new RuntimeException();
@@ -124,8 +119,7 @@ public void onFailure(Exception e) {
     public void testConcurrentFailures() throws InterruptedException {
         AtomicReference<Exception> finalException = new AtomicReference<>();
         int numGroups = randomIntBetween(10, 100);
-        GroupedActionListener<Void> listener = new GroupedActionListener<>(
-            ActionListener.wrap(r -> {}, finalException::set), numGroups, Collections.emptyList());
+        GroupedActionListener<Void> listener = new GroupedActionListener<>(ActionListener.wrap(r -> {}, finalException::set), numGroups);
         ExecutorService executorService = Executors.newFixedThreadPool(numGroups);
         for (int i = 0; i < numGroups; i++) {
             executorService.submit(() -> listener.onFailure(new IOException()));

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java

@@ -433,7 +433,7 @@ private void checkAutoFollowPattern(String autoFollowPattenName,
                 ActionListener.wrap(
                     rs -> resultHandler.accept(new AutoFollowResult(autoFollowPattenName, new ArrayList<>(rs))),
                     e -> { throw new AssertionError("must never happen", e); }),
-                leaderIndicesToFollow.size(), Collections.emptyList());
+                leaderIndicesToFollow.size());
 
             for (final Index indexToFollow : leaderIndicesToFollow) {
                 List<String> otherMatchingPatterns = patternsForTheSameRemoteCluster.stream()

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportUnfollowAction.java

@@ -41,7 +41,6 @@
 import org.elasticsearch.xpack.core.ccr.action.UnfollowAction;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 
@@ -137,8 +136,8 @@ public void onFailure(final Exception e) {
                             }
 
                         },
-                        numberOfShards,
-                        Collections.emptyList());
+                        numberOfShards
+                );
                 for (int i = 0; i < numberOfShards; i++) {
                     final ShardId followerShardId = new ShardId(indexMetaData.getIndex(), i);
                     final ShardId leaderShardId = new ShardId(leaderIndex, i);

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/saml/TransportSamlInvalidateSessionAction.java

@@ -28,7 +28,6 @@
 import org.elasticsearch.xpack.security.authc.saml.SamlUtils;
 import org.opensaml.saml.saml2.core.LogoutResponse;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -98,9 +97,7 @@ private void findAndInvalidateTokens(SamlRealm realm, SamlLogoutRequestHandler.R
                     listener.onResponse(0);
                 } else {
                     GroupedActionListener<TokensInvalidationResult> groupedListener = new GroupedActionListener<>(
-                        ActionListener.wrap(collection -> listener.onResponse(collection.size()), listener::onFailure),
-                        tokens.size(), Collections.emptyList()
-                    );
+                        ActionListener.wrap(collection -> listener.onResponse(collection.size()), listener::onFailure), tokens.size());
                     tokens.forEach(tuple -> invalidateTokenPair(tuple, groupedListener));
                 }
             }, listener::onFailure

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/user/TransportGetUsersAction.java

@@ -70,7 +70,7 @@ protected void doExecute(Task task, final GetUsersRequest request, final ActionL
                 listener.onResponse(new GetUsersResponse(users));
             }, listener::onFailure);
         final GroupedActionListener<Collection<User>> groupListener =
-                new GroupedActionListener<>(sendingListener, 2, Collections.emptyList());
+                new GroupedActionListener<>(sendingListener, 2);
         // We have two sources for the users object, the reservedRealm and the usersStore, we query both at the same time with a
         // GroupedActionListener
         if (realmLookup.isEmpty()) {
@@ -84,8 +84,7 @@ protected void doExecute(Task task, final GetUsersRequest request, final ActionL
         } else {
             // nested group listener action here - for each of the users we got and fetch it concurrently - once we are done we notify
             // the "global" group listener.
-            GroupedActionListener<User> realmGroupListener = new GroupedActionListener<>(groupListener, realmLookup.size(),
-                    Collections.emptyList());
+            GroupedActionListener<User> realmGroupListener = new GroupedActionListener<>(groupListener, realmLookup.size());
             for (String user : realmLookup) {
                 reservedRealm.lookupUser(user, realmGroupListener);
             }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/support/mapper/CompositeRoleMapper.java

@@ -7,7 +7,6 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -43,7 +42,7 @@ private CompositeRoleMapper(UserRoleMapper... delegates) {
     public void resolveRoles(UserData user, ActionListener<Set<String>> listener) {
         GroupedActionListener<Set<String>> groupListener = new GroupedActionListener<>(ActionListener.wrap(
                 composite -> listener.onResponse(composite.stream().flatMap(Set::stream).collect(Collectors.toSet())), listener::onFailure
-        ), delegates.size(), Collections.emptyList());
+        ), delegates.size());
         this.delegates.forEach(mapper -> mapper.resolveRoles(user, groupListener));
     }
 

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

@@ -506,7 +506,7 @@ private void authorizeBulkItems(RequestInfo requestInfo, AuthorizationInfo authz
                         listener.onResponse(null);
                     }, listener::onFailure);
                 final ActionListener<Tuple<String, IndexAuthorizationResult>> groupedActionListener = wrapPreservingContext(
-                    new GroupedActionListener<>(bulkAuthzListener, actionToIndicesMap.size(), Collections.emptyList()), threadContext);
+                    new GroupedActionListener<>(bulkAuthzListener, actionToIndicesMap.size()), threadContext);
 
                 actionToIndicesMap.forEach((bulkItemAction, indices) -> {
                     final RequestInfo bulkItemInfo =

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

@@ -191,7 +191,7 @@ public void putPrivileges(Collection<ApplicationPrivilegeDescriptor> privileges,
                         .map(NativePrivilegeStore::nameFromDocId)
                         .collect(TUPLES_TO_MAP);
                     clearRolesCache(listener, createdNames);
-                }, listener::onFailure), privileges.size(), Collections.emptyList());
+                }, listener::onFailure), privileges.size());
             for (ApplicationPrivilegeDescriptor privilege : privileges) {
                 innerPutPrivilege(privilege, refreshPolicy, groupListener);
             }
@@ -232,7 +232,7 @@ public void deletePrivileges(String application, Collection<String> names, Write
                             .map(NativePrivilegeStore::nameFromDocId)
                             .collect(TUPLES_TO_MAP);
                         clearRolesCache(listener, deletedNames);
-                    }, listener::onFailure), names.size(), Collections.emptyList());
+                    }, listener::onFailure), names.size());
                 for (String name : names) {
                     ClientHelper.executeAsyncWithOrigin(client.threadPool().getThreadContext(), SECURITY_ORIGIN,
                         client.prepareDelete(SECURITY_INDEX_NAME, SINGLE_MAPPING_NAME, toDocId(application, name))