brandboat commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r3488173565
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2276,39 +2282,79 @@ private CompletionStage<AddRaftVoterResponseData>
handleAddVoterRequest(
);
}
- Optional<ReplicaKey> newVoter = RaftUtil.addVoterRequestVoterKey(data);
- if (newVoter.isEmpty() || newVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Add voter request didn't include a valid
voter")
- );
- }
-
- Endpoints newVoterEndpoints =
Endpoints.fromAddVoterRequest(data.listeners());
- if (newVoterEndpoints.address(channel.listenerName()).isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage(
- String.format(
- "Add voter request didn't include the endpoint
(%s) for the default listener %s",
- newVoterEndpoints,
- channel.listenerName()
- )
- )
- );
+ final ReplicaKey newVoter;
+ final Endpoints newVoterEndpoints;
+ try {
+ newVoter = resolveAddVoterKey(data, requestMetadata.apiVersion(),
currentTimeMs);
+ newVoterEndpoints = resolveAddVoterEndpoints(data,
requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return
completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST,
e.getMessage()));
}
return addVoterHandler.handleAddVoterRequest(
quorum.leaderStateOrThrow(),
- newVoter.get(),
+ newVoter,
newVoterEndpoints,
data.ackWhenCommitted(),
currentTimeMs
);
}
+ private ReplicaKey resolveAddVoterKey(
+ AddRaftVoterRequestData data,
+ short apiVersion,
+ long currentTimeMs
+ ) {
+ if (apiVersion >= 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID))
{
+ List<ReplicaKey> matchingObservers = quorum.leaderStateOrThrow()
+ .observerStates(currentTimeMs)
+ .keySet()
+ .stream()
+ .filter(key -> key.id() == data.voterId())
+ .toList();
+ if (matchingObservers.size() > 1) {
+ throw new IllegalArgumentException(
+ String.format("Multiple observers with node ID %d detected
(%s).", data.voterId(), matchingObservers)
+ );
+ } else if (matchingObservers.isEmpty() ||
matchingObservers.get(0).directoryId().isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Could not find a unique observer with node ID %d to
derive its directory ID. " +
+ "Ensure the node is running and fetching before
adding it as a voter.",
+ data.voterId()
+ )
+ );
+ } else {
+ return matchingObservers.get(0);
+ }
+ }
+
+ return RaftUtil.addVoterRequestVoterKey(data)
+ .filter(voterKey -> voterKey.directoryId().isPresent())
+ .orElseThrow(() -> new IllegalArgumentException("Add voter request
didn't include a valid voter"));
+ }
+
+ private Endpoints resolveAddVoterEndpoints(
+ AddRaftVoterRequestData data,
+ short apiVersion
+ ) {
+ Endpoints endpoints = apiVersion < 2 || !data.listeners().isEmpty() ?
+ Endpoints.fromAddVoterRequest(data.listeners()) :
+ nodeEndpointProvider.endpointsOf(data.voterId());
+
+ if (endpoints.address(channel.listenerName()).isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Add voter request didn't include the endpoint (%s) for
the default listener %s",
Review Comment:
Good catch, thanks! I'll change it to below
```
Could not find an endpoint for voter %d on listener %s
```
--
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]
