junrao commented on code in PR #22505:
URL: https://github.com/apache/kafka/pull/22505#discussion_r3431765495
##########
raft/src/main/java/org/apache/kafka/raft/UnattachedState.java:
##########
@@ -87,6 +87,11 @@ public int epoch() {
return epoch;
}
+ @Override
+ public LeaderAndEpoch leaderAndEpoch() {
+ return new LeaderAndEpoch(leaderId, epoch);
Review Comment:
In this state, should the leader be empty?
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -3714,7 +3714,7 @@ public void testLeaderGracefulShutdownTimeout(boolean
withKip853Rpc) throws Exce
// Now shutdown
int shutdownTimeoutMs = 5000;
- CompletableFuture<Void> shutdownFuture =
context.client.shutdown(shutdownTimeoutMs);
+ CompletableFuture<Void> shutdownFuture =
context.client.shutdown(shutdownTimeoutMs).toCompletableFuture();
Review Comment:
Previously, we used `var`. Could we be consistent? Ditto below.
##########
raft/src/testFixtures/java/org/apache/kafka/raft/MockNetworkChannel.java:
##########
@@ -80,10 +88,15 @@ public boolean hasSentRequests() {
}
public void mockReceive(RaftResponse.Inbound response) {
- RaftRequest.Outbound request =
awaitingResponse.get(response.correlationId());
- if (request == null) {
+ var future = awaitingResponse.get(response.correlationId());
+ if (future == null) {
throw new IllegalStateException("Received response for a request
which is not being awaited");
}
- request.completion.complete(response);
+ future.complete(response);
}
+
+ private static record RequestEntry(
Review Comment:
`static` seems redundant.
##########
raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java:
##########
@@ -968,14 +968,14 @@ void deliverRequest(ApiMessage request, short version) {
versionedRequest,
time.milliseconds()
);
- inboundRequest.completion.whenComplete((response, exception) -> {
+ client.handle(inboundRequest).whenComplete((response, exception) -> {
if (exception != null) {
+ // TODO: this doesn't do anything interesting. Figure out a
way to fail the test
Review Comment:
Should we resolve the TODO in this PR?
##########
raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandler.java:
##########
@@ -150,17 +152,21 @@ public CompletableFuture<RemoveRaftVoterResponseData>
handleRemoveVoterRequest(
leaderState.appendVotersRecord(newVoters.get(), currentTimeMs),
time.timer(requestTimeoutMs)
);
- leaderState.resetRemoveVoterHandlerState(Errors.UNKNOWN_SERVER_ERROR,
null, Optional.of(state));
+
changeVoterState.resetRemoveVoterHandlerState(Errors.UNKNOWN_SERVER_ERROR,
null, Optional.of(state));
return state.future();
}
- public void highWatermarkUpdated(LeaderState<?> leaderState) {
- leaderState.removeVoterHandlerState().ifPresent(current ->
- leaderState.highWatermark().ifPresent(highWatermark -> {
- if (highWatermark.offset() > current.lastOffset()) {
+ public void highWatermarkUpdated(LeaderState<?> leaderState, long
highWatermark) {
+ var changeVoterState = leaderState.changeVoterState();
+
+ changeVoterState
+ .removeVoterHandlerState()
+ .ifPresent(current -> {
+ if (highWatermark > current.lastOffset()) {
// VotersRecord with the removed voter was committed;
complete the RPC
- leaderState.resetRemoveVoterHandlerState(Errors.NONE,
null, Optional.empty());
+ changeVoterState
+ .resetRemoveVoterHandlerState(Errors.NONE, null,
Optional.empty());
Review Comment:
Could this be merged with previous line?
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3652,8 +3656,15 @@ private void wakeup() {
*
* @param request The inbound request
*/
- public void handle(RaftRequest.Inbound request) {
- messageQueue.add(Objects.requireNonNull(request));
+ public CompletionStage<RaftResponse.Outbound> handle(RaftRequest.Inbound
request) {
Review Comment:
> * {@link RaftRequest.Inbound#completion}.
We need to adjust the javadoc doc above 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]
