cmccabe commented on a change in pull request #9032:
URL: https://github.com/apache/kafka/pull/9032#discussion_r479546505


##########
File path: 
clients/src/main/java/org/apache/kafka/clients/admin/DescribeUserScramCredentialsResult.java
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.errors.ResourceNotFoundException;
+import 
org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.protocol.Errors;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+/**
+ * The result of the {@link Admin#describeUserScramCredentials()} call.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeUserScramCredentialsResult {
+    private final KafkaFuture<DescribeUserScramCredentialsResponseData> 
dataFuture;
+
+    /**
+     * Package-private constructor
+     *
+     * @param dataFuture the future indicating response data from the call
+     */
+    
DescribeUserScramCredentialsResult(KafkaFuture<DescribeUserScramCredentialsResponseData>
 dataFuture) {
+        this.dataFuture = Objects.requireNonNull(dataFuture);
+    }
+
+    /**
+     *
+     * @return a future for the results of all described users with map keys 
(one per user) being consistent with the
+     * contents of the list returned by {@link #users()}. The future will 
complete successfully only if all such user
+     * descriptions complete successfully.
+     */
+    public KafkaFuture<Map<String, UserScramCredentialsDescription>> all() {
+        return KafkaFuture.allOf(dataFuture).thenApply(v -> {
+            DescribeUserScramCredentialsResponseData data = 
valueFromFutureGuaranteedToSucceedAtThisPoint(dataFuture);
+            /* Check to make sure every individual described user succeeded.  
Note that a successfully described user
+             * is one that appears with *either* a NONE error code or a 
RESOURCE_NOT_FOUND error code. The
+             * RESOURCE_NOT_FOUND means the client explicitly requested a 
describe of that particular user but it could
+             * not be described because it does not exist; such a user will 
not appear as a key in the returned map.
+             */
+            
Optional<DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult>
 optionalFirstFailedDescribe =
+                    data.results().stream().filter(result ->
+                        result.errorCode() != Errors.NONE.code() && 
result.errorCode() != Errors.RESOURCE_NOT_FOUND.code()).findFirst();
+            if (optionalFirstFailedDescribe.isPresent()) {
+                throw 
Errors.forCode(optionalFirstFailedDescribe.get().errorCode()).exception(optionalFirstFailedDescribe.get().errorMessage());
+            }
+            Map<String, UserScramCredentialsDescription> retval = new 
HashMap<>();
+            data.results().stream().forEach(userResult ->
+                    retval.put(userResult.user(), new 
UserScramCredentialsDescription(userResult.user(),
+                            getScramCredentialInfosFor(userResult))));
+            return retval;
+        });
+    }
+
+    /**
+     *
+     * @return a future indicating the distinct users that meet the request 
criteria and that have at least one
+     * credential.  The future will not complete successfully if the user is 
not authorized to perform the describe
+     * operation; otherwise, it will complete successfully as long as the list 
of users with credentials can be
+     * successfully determined within some hard-coded timeout period. Note 
that the returned list will not include users
+     * that do not exist/have no credentials: a request to describe an 
explicit list of users, none of which existed/had
+     * a credential, will result in a future that returns an empty list being 
returned here. A returned list will
+     * include users that have a credential but that could not be described.
+     */
+    public KafkaFuture<List<String>> users() {
+        return KafkaFuture.allOf(dataFuture).thenApply(v -> {
+            DescribeUserScramCredentialsResponseData data = 
valueFromFutureGuaranteedToSucceedAtThisPoint(dataFuture);
+            return data.results().stream()
+                    .filter(result -> result.errorCode() != 
Errors.RESOURCE_NOT_FOUND.code())
+                    .map(result -> result.user()).collect(Collectors.toList());
+        });
+    }
+
+    /**
+     *
+     * @param userName the name of the user description being requested
+     * @return a future indicating the description results for the given user. 
The future will complete exceptionally if
+     * the future returned by {@link #users()} completes exceptionally.  Note 
that if the given user does not exist in
+     * the list of described users then the returned future will complete 
exceptionally with
+     * {@link org.apache.kafka.common.errors.ResourceNotFoundException}.
+     */
+    public KafkaFuture<UserScramCredentialsDescription> description(String 
userName) {
+        return KafkaFuture.allOf(dataFuture).thenApply(v -> {
+            DescribeUserScramCredentialsResponseData data = 
valueFromFutureGuaranteedToSucceedAtThisPoint(dataFuture);
+            // it is possible that there is no future for this user (for 
example, the original describe request was for
+            // users 1, 2, and 3 but this is looking for user 4), so 
explicitly take care of that case
+            
Optional<DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult>
 optionalUserResult =
+                    data.results().stream().filter(result -> 
result.user().equals(userName)).findFirst();
+            if (!optionalUserResult.isPresent()) {
+                throw new ResourceNotFoundException("No such user: " + 
userName);
+            }
+            
DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult 
userResult = optionalUserResult.get();
+            if (userResult.errorCode() != Errors.NONE.code()) {
+                // RESOURCE_NOT_FOUND is included here
+                throw 
Errors.forCode(userResult.errorCode()).exception(userResult.errorMessage());
+            }
+            return new UserScramCredentialsDescription(userResult.user(), 
getScramCredentialInfosFor(userResult));
+        });
+    }
+
+    private static List<ScramCredentialInfo> getScramCredentialInfosFor(
+            
DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult 
userResult) {
+        return userResult.credentialInfos().stream().map(c ->
+                new 
ScramCredentialInfo(ScramMechanism.fromType(c.mechanism()), c.iterations()))
+                .collect(Collectors.toList());
+    }
+
+    private static <T> T 
valueFromFutureGuaranteedToSucceedAtThisPoint(KafkaFuture<T> future) {
+        try {
+            return future.get();

Review comment:
       The call to `get()` here isn't needed.  See the comment about about 
using `KafkaFuture#whenComplete`.  Chaining future results is a big part of the 
power of `CompletableFuture` (which `KafkaFuture` is a clone of... long 
story...)




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


Reply via email to