[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r660956483
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
##
@@ -250,6 +252,13 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+
+} else if (t instanceof NoBatchedFindCoordinatorsException) {
Review comment:
I opened https://issues.apache.org/jira/browse/KAFKA-13013
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r660952842
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
##
@@ -0,0 +1,156 @@
+/*
+ * 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.internals;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.OffsetCommitRequestData;
+import
org.apache.kafka.common.message.OffsetCommitRequestData.OffsetCommitRequestPartition;
+import
org.apache.kafka.common.message.OffsetCommitRequestData.OffsetCommitRequestTopic;
+import
org.apache.kafka.common.message.OffsetCommitResponseData.OffsetCommitResponsePartition;
+import
org.apache.kafka.common.message.OffsetCommitResponseData.OffsetCommitResponseTopic;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.OffsetCommitRequest;
+import org.apache.kafka.common.requests.OffsetCommitResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest.CoordinatorType;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+public class AlterConsumerGroupOffsetsHandler implements
AdminApiHandler> {
+
+private final CoordinatorKey groupId;
+private final Map offsets;
+private final Logger log;
+private final AdminApiLookupStrategy lookupStrategy;
+
+public AlterConsumerGroupOffsetsHandler(
+String groupId,
+Map offsets,
+LogContext logContext
+) {
+this.groupId = CoordinatorKey.byGroupId(groupId);
+this.offsets = offsets;
+this.log = logContext.logger(AlterConsumerGroupOffsetsHandler.class);
+this.lookupStrategy = new CoordinatorStrategy(CoordinatorType.GROUP,
logContext);
+}
+
+@Override
+public String apiName() {
+return "offsetCommit";
+}
+
+@Override
+public AdminApiLookupStrategy lookupStrategy() {
+return lookupStrategy;
+}
+
+public static AdminApiFuture.SimpleAdminApiFuture> newFuture(
+String groupId
+) {
+return
AdminApiFuture.forKeys(Collections.singleton(CoordinatorKey.byGroupId(groupId)));
+}
+
+@Override
+public OffsetCommitRequest.Builder buildRequest(int brokerId,
Set keys) {
+List topics = new ArrayList<>();
+Map> offsetData = new
HashMap<>();
+for (Map.Entry entry :
offsets.entrySet()) {
+String topic = entry.getKey().topic();
+OffsetAndMetadata oam = entry.getValue();
+offsetData.compute(topic, (key, value) -> {
+if (value == null) {
+value = new ArrayList<>();
+}
+OffsetCommitRequestPartition partition = new
OffsetCommitRequestPartition()
+.setCommittedOffset(oam.offset())
+.setCommittedLeaderEpoch(oam.leaderEpoch().orElse(-1))
+.setCommittedMetadata(oam.metadata())
+.setPartitionIndex(entry.getKey().partition());
+value.add(partition);
+return value;
+});
+}
+for (Map.Entry> entry :
offsetData.entrySet()) {
+OffsetCommitRequestTopic topic = new OffsetCommitRequestTopic()
+.setName(entry.getKey())
+.setPartitions(entry.getValue());
+topics.add(topic);
+}
+OffsetCommitRequestData data = new OffsetCommitRequestData()
+.setGroupId(groupId.idValue)
+.setTopics(topics);
+return new OffsetCommitRequest.Builder(data);
+}
+
+@Override
+public ApiResult>
handleResponse(Node broker, Set groupIds,
+AbstractResponse abstractResponse) {
+
+final OffsetCommitResponse response =
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r659637223
##
File path:
clients/src/main/java/org/apache/kafka/common/errors/NoBatchedFindCoordinatorsException.java
##
@@ -0,0 +1,33 @@
+/*
+ * 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.common.errors;
+
+/**
+ * Indicates that it is not possible to lookup coordinators in batches with
FindCoordinator. Instead
+ * coordinators must be looked up one by one.
+ */
+public class NoBatchedFindCoordinatorsException extends
UnsupportedVersionException {
Review comment:
I pushed a change that moves `NoBatchedFindCoordinatorsException` into
`FindCoordinatorRequest`. Do you have any further concerns?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r659636618
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
##
@@ -250,6 +252,13 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+
+} else if (t instanceof NoBatchedFindCoordinatorsException) {
Review comment:
`AdminApiDriver` was still evolving rapidly while I was implementing
this KIP so I went for the straighforward approach.
But I agree, it would be best to avoid this type of logic here. The goal
would be to find a mechanism that works for all clients. @tombentley suggested
an alternative option in
https://github.com/apache/kafka/pull/10743#discussion_r649872433.
I've not had the time to look into better alternatives yet.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r659601729
##
File path:
clients/src/main/java/org/apache/kafka/common/errors/NoBatchedFindCoordinatorsException.java
##
@@ -0,0 +1,33 @@
+/*
+ * 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.common.errors;
+
+/**
+ * Indicates that it is not possible to lookup coordinators in batches with
FindCoordinator. Instead
+ * coordinators must be looked up one by one.
+ */
+public class NoBatchedFindCoordinatorsException extends
UnsupportedVersionException {
Review comment:
I agree, I'll move it to an inner class of `FindCoordinatorRequest` and
update the KIP accordingly
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r659600886
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
##
@@ -17,84 +17,160 @@
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 {
+
+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 Set unrepresentableKeys = Collections.emptySet();
+
+boolean batch = true;
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;
+}
Review comment:
Right, I merged both as `BATCH_REQUEST_SCOPE`
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r659597050
##
File path: clients/src/test/java/org/apache/kafka/clients/MockClient.java
##
@@ -245,10 +246,18 @@ public void send(ClientRequest request, long now) {
unsupportedVersionException = new UnsupportedVersionException(
"Api " + request.apiKey() + " with version " +
version);
} else {
-AbstractRequest abstractRequest =
request.requestBuilder().build(version);
-if (!futureResp.requestMatcher.matches(abstractRequest))
-throw new IllegalStateException("Request matcher did not
match next-in-line request "
-+ abstractRequest + " with prepared response " +
futureResp.responseBody);
+try {
+AbstractRequest abstractRequest =
request.requestBuilder().build(version);
+if (!futureResp.requestMatcher.matches(abstractRequest))
+throw new IllegalStateException("Request matcher did
not match next-in-line request "
++ abstractRequest + " with prepared response "
+ futureResp.responseBody);
+} catch (NoBatchedFindCoordinatorsException uble) {
Review comment:
Sure, I raised https://issues.apache.org/jira/browse/KAFKA-13000
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r658016506
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##
@@ -4806,7 +4275,7 @@ public ListTransactionsResult
listTransactions(ListTransactionsOptions options)
@Override
void handleResponse(AbstractResponse response) {
long currentTimeMs = time.milliseconds();
-driver.onResponse(currentTimeMs, spec, response);
+driver.onResponse(currentTimeMs, spec, response,
nodeProvider.provide());
Review comment:
All of this is specifically because `ConsumerGroupDescription` needs the
coordinator as a `Node` when describing consumer groups.
At the moment, handlers only get the brokerId and that's not enough to build
a `Node` object. So instead in this PR, I replaced the `brokerId` argument by
the full `Node` object. The `brokerId` can be retrieved from `spec.scope` but
for the `Node`, we need to get that from the `Call` itself.
Good catch, `nodeProvider.provide()` could return a different `Node`! As far
as I can tell, we can use `curNode()` here, that should have been set by
`maybeDrainPendingCall()`.
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743: URL: https://github.com/apache/kafka/pull/10743#discussion_r657891229 ## File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorKey.java ## @@ -24,7 +24,7 @@ public final String idValue; Review comment: This package is part of the public API. I'm not sure we want to expose `CoordinatorKey` which is currently only used internally. That said, [KIP-692](https://cwiki.apache.org/confluence/display/KAFKA/KIP-692%3A+Make+AdminClient+value+object+constructors+public) proposes exposing the constructors of all the `*Result` classes. Maybe that discussion should happen there? -- 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: [email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657884154
##
File path:
clients/src/main/java/org/apache/kafka/common/errors/NoBatchedFindCoordinatorsException.java
##
@@ -0,0 +1,33 @@
+/*
+ * 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.common.errors;
+
+/**
+ * Indicates that it is not possible to lookup coordinators in batches with
FindCoordinator. Instead
+ * coordinators must be looked up one by one.
+ */
+public class NoBatchedFindCoordinatorsException extends
UnsupportedVersionException {
Review comment:
Right, I'll update the KIP and thread
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657881689
##
File path: clients/src/test/java/org/apache/kafka/clients/MockClient.java
##
@@ -245,10 +246,18 @@ public void send(ClientRequest request, long now) {
unsupportedVersionException = new UnsupportedVersionException(
"Api " + request.apiKey() + " with version " +
version);
} else {
-AbstractRequest abstractRequest =
request.requestBuilder().build(version);
-if (!futureResp.requestMatcher.matches(abstractRequest))
-throw new IllegalStateException("Request matcher did not
match next-in-line request "
-+ abstractRequest + " with prepared response " +
futureResp.responseBody);
+try {
+AbstractRequest abstractRequest =
request.requestBuilder().build(version);
+if (!futureResp.requestMatcher.matches(abstractRequest))
+throw new IllegalStateException("Request matcher did
not match next-in-line request "
++ abstractRequest + " with prepared response "
+ futureResp.responseBody);
+} catch (NoBatchedFindCoordinatorsException uble) {
Review comment:
Yes I looked into it but I think there's 1 test in
TransactionManagerTest that expects `UnsupportedVersionException` to be thrown.
MockClient works in slightly different ways than the real client and it
would be good to address this but I'd rather defer to a follow up PR.
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657873384
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorKey.java
##
@@ -24,7 +24,7 @@
public final String idValue;
public final FindCoordinatorRequest.CoordinatorType type;
-private CoordinatorKey(String idValue,
FindCoordinatorRequest.CoordinatorType type) {
+public CoordinatorKey(FindCoordinatorRequest.CoordinatorType type, String
idValue) {
Review comment:
I was using it in `CoordinatorStrategy.handleResponse()` but I get
easily get around it. It makes sense to keep it private
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657870204
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/DescribeConsumerGroupsResult.java
##
@@ -34,39 +35,40 @@
@InterfaceStability.Evolving
public class DescribeConsumerGroupsResult {
-private final Map> futures;
+private final Map> futures;
-public DescribeConsumerGroupsResult(final Map> futures) {
+public DescribeConsumerGroupsResult(Map> futures) {
Review comment:
Explained in
https://github.com/apache/kafka/pull/10743#discussion_r657473433
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657863835
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/DeleteConsumerGroupsResult.java
##
@@ -29,9 +32,9 @@
*/
@InterfaceStability.Evolving
public class DeleteConsumerGroupsResult {
-private final Map> futures;
+private final Map> futures;
-DeleteConsumerGroupsResult(final Map> futures) {
+DeleteConsumerGroupsResult(Map>
futures) {
Review comment:
Yes that's exactly the reason why I made this change. I'll add the
`final` modifier back though
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657856869
##
File path:
clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorResponse.java
##
@@ -95,4 +96,21 @@ public static FindCoordinatorResponse prepareResponse(Errors
error, Node node) {
.setPort(node.port());
return new FindCoordinatorResponse(data);
}
+
+public static FindCoordinatorResponse prepareResponse(Errors error, String
key, Node node) {
+FindCoordinatorResponseData data = new FindCoordinatorResponseData();
+data.setCoordinators(Collections.singletonList(
+new FindCoordinatorResponseData.Coordinator()
+.setErrorCode(error.code())
+.setErrorMessage(error.message())
+.setKey(key)
+.setHost(node.host())
+.setPort(node.port())
+.setNodeId(node.id(;
+return new FindCoordinatorResponse(data);
+}
+
+public static boolean isBatch(RequestHeader header) {
+return header.apiVersion() >= 4;
+}
Review comment:
Right, deleted
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r657856438
##
File path: clients/src/main/resources/common/message/FindCoordinatorRequest.json
##
@@ -23,12 +23,16 @@
// Version 2 is the same as version 1.
//
// Version 3 is the first flexible version.
- "validVersions": "0-3",
+ //
+ // Version 4 adds CoordinatorKeys
+ "validVersions": "0-4",
"flexibleVersions": "3+",
"fields": [
-{ "name": "Key", "type": "string", "versions": "0+",
+{ "name": "Key", "type": "string", "versions": "0-3",
"about": "The coordinator key." },
{ "name": "KeyType", "type": "int8", "versions": "1+", "default": "0",
"ignorable": false,
Review comment:
`KeyType` was added in version 1 to resolve transaction coordinators.
Before that, FindCoordinator was only able to resolve group coordinators.
It's still used in version 4 as we typically never want to resolve
transaction and consumer coordinators at the same time so it determines the
type of all keys in the request.
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r654558597
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##
@@ -813,34 +815,56 @@ public void handle(SyncGroupResponse syncResponse,
*/
private RequestFuture sendFindCoordinatorRequest(Node node) {
// initiate the group metadata request
-log.debug("Sending FindCoordinator request to broker {}", node);
-FindCoordinatorRequest.Builder requestBuilder =
-new FindCoordinatorRequest.Builder(
-new FindCoordinatorRequestData()
-.setKeyType(CoordinatorType.GROUP.id())
-.setKey(this.rebalanceConfig.groupId));
+log.debug("Sending FindCoordinator request to broker {} with
batch={}", node, batchFindCoordinator);
+FindCoordinatorRequestData data = new FindCoordinatorRequestData()
+.setKeyType(CoordinatorType.GROUP.id());
+if (batchFindCoordinator) {
+
data.setCoordinatorKeys(Collections.singletonList(this.rebalanceConfig.groupId));
+} else {
+data.setKey(this.rebalanceConfig.groupId);
+}
+FindCoordinatorRequest.Builder requestBuilder = new
FindCoordinatorRequest.Builder(data);
return client.send(node, requestBuilder)
-.compose(new FindCoordinatorResponseHandler());
+.compose(new
FindCoordinatorResponseHandler(batchFindCoordinator));
}
private class FindCoordinatorResponseHandler extends
RequestFutureAdapter {
+private boolean batch;
+FindCoordinatorResponseHandler(boolean batch) {
+this.batch = batch;
+}
@Override
public void onSuccess(ClientResponse resp, RequestFuture future)
{
log.debug("Received FindCoordinator response {}", resp);
FindCoordinatorResponse findCoordinatorResponse =
(FindCoordinatorResponse) resp.responseBody();
-Errors error = findCoordinatorResponse.error();
+if (batch && findCoordinatorResponse.data().coordinators().size()
!= 1) {
+log.error("Group coordinator lookup failed: Invalid response
containing more than a single coordinator");
+future.raise(new IllegalStateException("Group coordinator
lookup failed: Invalid response containing more than a single coordinator"));
+}
+Errors error = batch
+?
Errors.forCode(findCoordinatorResponse.data().coordinators().get(0).errorCode())
+: findCoordinatorResponse.error();
if (error == Errors.NONE) {
synchronized (AbstractCoordinator.this) {
+int nodeId = batch
+?
findCoordinatorResponse.data().coordinators().get(0).nodeId()
Review comment:
This variable would only exist if `batch` is true which makes it pretty
awkward.
##
File path: clients/src/test/java/org/apache/kafka/clients/MockClient.java
##
@@ -245,10 +246,17 @@ public void send(ClientRequest request, long now) {
unsupportedVersionException = new UnsupportedVersionException(
"Api " + request.apiKey() + " with version " +
version);
} else {
-AbstractRequest abstractRequest =
request.requestBuilder().build(version);
-if (!futureResp.requestMatcher.matches(abstractRequest))
-throw new IllegalStateException("Request matcher did not
match next-in-line request "
-+ abstractRequest + " with prepared response " +
futureResp.responseBody);
+try {
+AbstractRequest abstractRequest =
request.requestBuilder().build(version);
+if (!futureResp.requestMatcher.matches(abstractRequest))
+continue;
Review comment:
Oops! Yes I removed this to silence failing tests while refactoring.
I'll undo this.
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
##
@@ -57,13 +58,13 @@
* Note that keys which received a retriable error should be left out of
the
* result. They will be retried automatically.
*
- * @param brokerId the brokerId that the associated request was sent to
+ * @param broker the broker that the associated request was sent to
Review comment:
Not entirely sure about the new naming in place now, but does that still
count as a broker?
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
##
@@ -57,13 +58,13 @@
* Note that keys which received a retriable error should be left out of
the
* result. They will be retried automatically.
*
- *
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r649213934
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##
@@ -858,6 +885,12 @@ public void onSuccess(ClientResponse resp,
RequestFuture future) {
public void onFailure(RuntimeException e, RequestFuture future) {
log.debug("FindCoordinator request failed due to {}",
e.toString());
+if (e instanceof UnsupportedBatchLookupException) {
Review comment:
I've only taken a very brief look and I think this approach would work
well for Connect, Producer and Consumer, however it's a bit more complicated
with Admin.
In Admin, requests are built by lookup strategies. Lookups can be sent to
any broker so knowing the max version for a specific call is not completely
trivial. That said, it's not impossible either so if there's concensus it would
be preferable I can give that a try.
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743: URL: https://github.com/apache/kafka/pull/10743#discussion_r649208497 ## File path: clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java ## @@ -63,7 +64,7 @@ * * @return result indicating key completion, failure, and unmapping */ -ApiResult handleResponse(int brokerId, Set keys, AbstractResponse response); +ApiResult handleResponse(int brokerId, Set keys, AbstractResponse response, Node node); Review comment: I've actually replaced the first argument `int brokerId` by `Node broker` and removed the last argument. -- 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: [email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648552780
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
##
@@ -204,7 +206,8 @@ private void completeLookup(Map
brokerIdMapping) {
public void onResponse(
long currentTimeMs,
RequestSpec spec,
-AbstractResponse response
+AbstractResponse response,
+Node node
Review comment:
Good catch!
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648537766
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
##
@@ -17,20 +17,34 @@
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 {
+
+private static final ApiRequestScope GROUP_REQUEST_SCOPE = new
ApiRequestScope() { };
+private static final ApiRequestScope TXN_REQUEST_SCOPE = new
ApiRequestScope() { };
+
private final Logger log;
+private boolean batch = true;
+private FindCoordinatorRequest.CoordinatorType type;
Review comment:
That's a good idea, yes
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648526428
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
##
@@ -250,6 +254,13 @@ public void onFailure(
.filter(future.lookupKeys()::contains)
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
+
+} else if (t instanceof UnsupportedBatchLookupException) {
+((CoordinatorStrategy) handler.lookupStrategy()).disableBatch();
Review comment:
Right `UnsupportedBatchLookupException` is not a great name!
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648525658
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorKey.java
##
@@ -24,7 +24,7 @@
public final String idValue;
public final FindCoordinatorRequest.CoordinatorType type;
-private CoordinatorKey(String idValue,
FindCoordinatorRequest.CoordinatorType type) {
+public CoordinatorKey(String idValue,
FindCoordinatorRequest.CoordinatorType type) {
Review comment:
I agree, since there are very few callers, I'll make the change
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648522353
##
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##
@@ -1305,66 +1305,102 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleFindCoordinatorRequest(request: RequestChannel.Request): Unit = {
+val version = request.header.apiVersion
+if (version < 4) {
+ handleFindCoordinatorRequestLessThanV4(request)
+} else {
+ handleFindCoordinatorRequestV4AndAbove(request)
Review comment:
We need a different `createResponse()` for each version. Both methods
call `getCoordinator()` and have no logic.
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648520281
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
##
@@ -267,7 +278,11 @@ public void onFailure(
private void clearInflightRequest(long currentTimeMs, RequestSpec spec)
{
RequestState requestState = requestStates.get(spec.scope);
if (requestState != null) {
-requestState.clearInflight(currentTimeMs);
+if (spec.scope instanceof FulfillmentScope) {
+requestState.clearInflight(currentTimeMs + retryBackoffMs);
+} else {
+requestState.clearInflight(currentTimeMs);
Review comment:
That surprised me too. It is the current behaviour that
all`*RetryBackoff` tests in `KafkaAdminClientTest` enforce
--
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:
[email protected]
[GitHub] [kafka] mimaison commented on a change in pull request #10743: KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time
mimaison commented on a change in pull request #10743:
URL: https://github.com/apache/kafka/pull/10743#discussion_r648519181
##
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
##
@@ -40,61 +54,116 @@ public CoordinatorStrategy(
@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 (key.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
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);
+type = keys.iterator().next().type;
Review comment:
Yes `requireSameType` ensures there is 1 type
--
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:
[email protected]
