private <T extends ApiRequestScope> void collectRequests(
        List<RequestSpec<K>> requests,
        BiMultimap<T, K> multimap,
        BiFunction<Set<K>, T, Collection<AdminApiHandler.RequestAndKeys<K>>> buildRequest
    ) {
        for (Map.Entry<T, Set<K>> entry : multimap.entrySet()) {
            T scope = entry.getKey();

            Set<K> keys = entry.getValue();
            if (keys.isEmpty()) {
                continue;
            }

            RequestState requestState = requestStates.computeIfAbsent(scope, c -> new RequestState());
            if (requestState.hasInflight()) {
                continue;
            }

            // Copy the keys to avoid exposing the underlying mutable set
            Set<K> copyKeys = Collections.unmodifiableSet(new HashSet<>(keys));

            Collection<AdminApiHandler.RequestAndKeys<K>> newRequests = buildRequest.apply(copyKeys, scope);
            if (newRequests.isEmpty()) {
                return;
            }

            // Only process the first request; all the remaining requests will be targeted at the same broker
            // and we don't want to issue more than one fulfillment request per broker at a time
            AdminApiHandler.RequestAndKeys<K> newRequest = newRequests.iterator().next();
            RequestSpec<K> spec = new RequestSpec<>(
                handler.apiName() + "(api=" + newRequest.request.apiKey() + ")",
                scope,
                newRequest.keys,
                newRequest.request,
                requestState.nextAllowedRetryMs,
                deadlineMs,
                requestState.tries
            );

            requestState.setInflight(spec);
            requests.add(spec);
        }
    }
Callback that is invoked when a `Call` is failed.
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java