skaundinya15 commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r655134390
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
##########
@@ -17,84 +17,153 @@
package org.apache.kafka.clients.admin.internals;
import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.common.errors.InvalidGroupIdException;
import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
import org.apache.kafka.common.message.FindCoordinatorRequestData;
+import org.apache.kafka.common.message.FindCoordinatorResponseData.Coordinator;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.FindCoordinatorRequest.CoordinatorType;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.stream.Collectors;
public class CoordinatorStrategy implements
AdminApiLookupStrategy<CoordinatorKey> {
+
+ private static final ApiRequestScope GROUP_REQUEST_SCOPE = new
ApiRequestScope() { };
+ private static final ApiRequestScope TXN_REQUEST_SCOPE = new
ApiRequestScope() { };
+
private final Logger log;
+ private final FindCoordinatorRequest.CoordinatorType type;
+ private boolean batch = true;
+ private Set<CoordinatorKey> unrepresentableKeys = Collections.emptySet();
public CoordinatorStrategy(
+ FindCoordinatorRequest.CoordinatorType type,
LogContext logContext
) {
+ this.type = type;
this.log = logContext.logger(CoordinatorStrategy.class);
}
@Override
public ApiRequestScope lookupScope(CoordinatorKey key) {
- // The `FindCoordinator` API does not support batched lookups, so we
use a
- // separate lookup context for each coordinator key we need to lookup
- return new LookupRequestScope(key);
+ if (batch) {
+ if (type == CoordinatorType.GROUP) {
+ return GROUP_REQUEST_SCOPE;
+ } else {
+ return TXN_REQUEST_SCOPE;
+ }
+ } else {
+ // If the `FindCoordinator` API does not support batched lookups,
we use a
+ // separate lookup context for each coordinator key we need to
lookup
+ return new LookupRequestScope(key);
+ }
}
@Override
public FindCoordinatorRequest.Builder buildRequest(Set<CoordinatorKey>
keys) {
- CoordinatorKey key = requireSingleton(keys);
- return new FindCoordinatorRequest.Builder(
- new FindCoordinatorRequestData()
- .setKey(key.idValue)
- .setKeyType(key.type.id())
- );
+ unrepresentableKeys = keys.stream().filter(k ->
!isRepresentableKey(k.idValue)).collect(Collectors.toSet());
+ keys = keys.stream().filter(k ->
isRepresentableKey(k.idValue)).collect(Collectors.toSet());
+ if (batch) {
+ keys = requireSameType(keys);
+ FindCoordinatorRequestData data = new FindCoordinatorRequestData()
+ .setKeyType(type.id())
+ .setCoordinatorKeys(keys.stream().map(k ->
k.idValue).collect(Collectors.toList()));
+ return new FindCoordinatorRequest.Builder(data);
+ } else {
+ CoordinatorKey key = requireSingleton(keys);
+ return new FindCoordinatorRequest.Builder(
+ new FindCoordinatorRequestData()
+ .setKey(key.idValue)
+ .setKeyType(key.type.id())
+ );
+ }
}
@Override
public LookupResult<CoordinatorKey> handleResponse(
Set<CoordinatorKey> keys,
AbstractResponse abstractResponse
) {
- CoordinatorKey key = requireSingleton(keys);
+ Map<CoordinatorKey, Integer> mappedKeys = new HashMap<>();
+ Map<CoordinatorKey, Throwable> failedKeys = new HashMap<>();
+
+ for (CoordinatorKey key : unrepresentableKeys) {
+ failedKeys.put(key, new InvalidGroupIdException("The given group
id '" +
+ key.idValue + "' cannot be represented in a
request."));
+ }
FindCoordinatorResponse response = (FindCoordinatorResponse)
abstractResponse;
- Errors error = response.error();
+ if (batch) {
+ for (Coordinator coordinator : response.data().coordinators()) {
+ handleError(Errors.forCode(coordinator.errorCode()),
+ new CoordinatorKey(type, coordinator.key()),
+ coordinator.nodeId(),
+ mappedKeys,
+ failedKeys);
+ }
+ } else {
+ CoordinatorKey key = requireSingleton(keys);
+ Errors error = response.error();
+ handleError(error, key, response.node().id(), mappedKeys,
failedKeys);
+ }
+ return new LookupResult<>(failedKeys, mappedKeys);
+ }
+
+ public void disableBatch() {
+ batch = false;
+ }
+
+ private static CoordinatorKey requireSingleton(Set<CoordinatorKey> keys) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected lookup key set");
+ }
+ return keys.iterator().next();
+ }
+ private static Set<CoordinatorKey> requireSameType(Set<CoordinatorKey>
keys) {
+ if (keys.stream().map(k -> k.type).collect(Collectors.toSet()).size()
!= 1) {
+ throw new IllegalArgumentException("Unexpected lookup key set");
+ }
+ return keys;
+ }
+
+ private static boolean isRepresentableKey(String groupId) {
+ return groupId != null;
+ }
+
+ private void handleError(Errors error, CoordinatorKey key, int nodeId,
Map<CoordinatorKey, Integer> mappedKeys, Map<CoordinatorKey, Throwable>
failedKeys) {
switch (error) {
case NONE:
- return LookupResult.mapped(key, response.data().nodeId());
-
+ mappedKeys.put(key, nodeId);
+ break;
case COORDINATOR_NOT_AVAILABLE:
Review comment:
@mimaison Gotcha, makes sense. Can we still we can still log at `DEBUG`
level? In case we need to debug, we have evidence of this in the logs.
--
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
