Repository: apache/kafka
File: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java (Java)

Code before:
                    .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()
                        )
                    )
            );
        }

        return addVoterHandler.handleAddVoterRequest(
            quorum.leaderStateOrThrow(),
            newVoter.get(),
            newVoterEndpoints,
            currentTimeMs,
            data.ackWhenCommitted()
        );
    }

    private boolean handleApiVersionsResponse(
        RaftResponse.Inbound responseMetadata,
        long currentTimeMs
    ) {
        if (!quorum.isLeader()) {
            // Not the leader anymore just ignore the API_VERSIONS response
            return true;
        }

        ApiVersionsResponseData response = (ApiVersionsResponseData) responseMetadata.data();

        Errors error = Errors.forCode(response.errorCode());
        Optional<ApiVersionsResponseData.SupportedFeatureKey> supportedKraftVersions =
            Optional.ofNullable(response.supportedFeatures().find(KRaftVersion.FEATURE_NAME));

        return addVoterHandler.handleApiVersionsResponse(
            quorum.leaderStateOrThrow(),
            responseMetadata.source(),
            error,
            supportedKraftVersions,
            currentTimeMs
        );

Review comment (suggestion):
Let's swap the order of `currentTimeMs` and `ackWhenCommitted`. "Ack when committed" is more important and raft tends to pass the current time as the last parameter.

Code after:
                    .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()
                        )
                    )
            );
        }

        return addVoterHandler.handleAddVoterRequest(
            quorum.leaderStateOrThrow(),
            newVoter.get(),
            newVoterEndpoints,
            data.ackWhenCommitted(),
            currentTimeMs
        );
    }

    private boolean handleApiVersionsResponse(
        RaftResponse.Inbound responseMetadata,
        long currentTimeMs
    ) {
        if (!quorum.isLeader()) {
            // Not the leader anymore just ignore the API_VERSIONS response
            return true;
        }

        ApiVersionsResponseData response = (ApiVersionsResponseData) responseMetadata.data();

        Errors error = Errors.forCode(response.errorCode());
        Optional<ApiVersionsResponseData.SupportedFeatureKey> supportedKraftVersions =
            Optional.ofNullable(response.supportedFeatures().find(KRaftVersion.FEATURE_NAME));

        return addVoterHandler.handleApiVersionsResponse(
            quorum.leaderStateOrThrow(),
            responseMetadata.source(),
            error,
            supportedKraftVersions,
            currentTimeMs
        );