chia7712 commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r3488234183
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2276,39 +2281,82 @@ 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 were found: %s. " +
+ "Remove all but one of these observers and try again.",
Review Comment:
I guess this scenario usually happens when a node recovers with a new log
directory? If so, the error message could be more precise by explicitly
reminding users to remove the stale/inactive observer.
##########
clients/src/main/java/org/apache/kafka/clients/admin/Admin.java:
##########
@@ -1864,6 +1864,22 @@ default ListClientMetricsResourcesResult
listClientMetricsResources() {
*/
Uuid clientInstanceId(Duration timeout);
+ /**
+ * Add a new voter node to the KRaft metadata quorum.
+ *
+ * <p>
+ * This is a convenience method that derives the voter's directory ID and
endpoints
+ * from the active controller's in-memory state. It is not idempotent.
+ * For scenarios such as node disk failure where multiple observers may
share the same
+ * node ID with different directory UUIDs, use
+ * {@link #addRaftVoter(int, Uuid, Set, AddRaftVoterOptions)} instead.
+ *
+ * @param voterId The node ID of the voter to add.
+ */
+ default AddRaftVoterResult addRaftVoter(int voterId) {
Review Comment:
I noticed a usability gap with the current convenience method: users who
want to use `AddRaftVoterOptions` (e.g., to verify the `clusterId`) are forced
to fall back to the verbose API.
Since this PR effectively shifts `voterDirectoryId` and endpoints from being
required payloads to optional guard/validation conditions, wouldn't it make
sense to move them into `AddRaftVoterOptions`? They now serve a very similar
purpose to `clusterId` in verifying the target.
If we do this, we could converge on `addRaftVoter(int voterId,
AddRaftVoterOptions options)` as the standard API and deprecate the older, more
verbose methods. WDYT?
##########
core/src/main/scala/kafka/server/SharedServer.scala:
##########
@@ -115,6 +115,22 @@ class SharedServer(
val brokerConfig = new KafkaConfig(sharedServerConfig.props, false)
val controllerConfig = new KafkaConfig(sharedServerConfig.props, false)
val supportedConfigChecker: SupportedConfigChecker = new
DefaultSupportedConfigChecker()
+ val controllerRegistrationsPublisher:
Option[ControllerRegistrationsPublisher] =
Review Comment:
we need to call `controllerRegistrationsPublisher#close` on the stop phase,
right?
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2415,22 +2463,42 @@ private CompletionStage<RemoveRaftVoterResponseData>
handleRemoveVoterRequest(
);
}
- Optional<ReplicaKey> oldVoter =
RaftUtil.removeVoterRequestVoterKey(data);
- if (oldVoter.isEmpty() || oldVoter.get().directoryId().isEmpty()) {
+ final ReplicaKey oldVoter;
+ try {
+ oldVoter = resolveRemoveVoterKey(data,
requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
Review Comment:
Could we leverage `RaftUtil.removeVoterResponse`?
##########
clients/src/main/resources/common/message/AddRaftVoterRequest.json:
##########
@@ -19,7 +19,9 @@
"listeners": ["controller", "broker"],
"name": "AddRaftVoterRequest",
// Version 1 adds the AckWhenCommitted field.
- "validVersions": "0-1",
+ // Version 2 allows VoterDirectoryId to be ZERO_UUID and Listeners to be
empty, in which
Review Comment:
Should we add checks to `AddRaftVoterRequest#build` to fail fast?
--
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]
