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

Code before:
                    current.voterEndpoints(),
                    new SupportedVersionRange(
                        supportedKraftVersions.get().minVersion(),
                        supportedKraftVersions.get().maxVersion()
                    )
                )
            )
            .orElseThrow(() ->
                new IllegalStateException(
                    String.format(
                        "Unable to add %s to the set of voters %s",
                        current.voterKey(),
                        partitionState.lastVoterSet()
                    )
                )
            );
        current.setLastOffset(leaderState.appendVotersRecord(newVoters, currentTimeMs));
        if (!current.ackWhenCommitted()) {
            // complete the future to send response, but do not reset the state,
            // since the new voter set is not yet committed
            current.future().complete(
                RaftUtil.addVoterResponse(
                    Errors.NONE,
                    null
                )
            );
        }
        return true;
    }

    public void highWatermarkUpdated(LeaderState<?> leaderState) {
        leaderState.addVoterHandlerState().ifPresent(current -> {
            leaderState.highWatermark().ifPresent(highWatermark -> {
                current.lastOffset().ifPresent(lastOffset -> {
                    if (highWatermark.offset() > lastOffset) {
                        // VotersRecord with the added voter was committed; complete the RPC
                        leaderState.resetAddVoterHandlerState(Errors.NONE, null, Optional.empty());
                    }
                });
            });
        });
    }

    private ApiVersionsRequestData buildApiVersionsRequest() {
        return new ApiVersionsRequest.Builder().build().data();
    }

    private boolean validVersionRange(
        KRaftVersion finalizedVersion,
        Optional<ApiVersionsResponseData.SupportedFeatureKey> supportedKraftVersions
    ) {

Review comment (suggestion):
Feel free to join some of these lines if they are less than about 100 characters. E.g.:
```java
            current.future().complete(
                RaftUtil.addVoterResponse(Errors.NONE, null)
            );
```

Code after:
                    current.voterEndpoints(),
                    new SupportedVersionRange(
                        supportedKraftVersions.get().minVersion(),
                        supportedKraftVersions.get().maxVersion()
                    )
                )
            )
            .orElseThrow(() ->
                new IllegalStateException(
                    String.format(
                        "Unable to add %s to the set of voters %s",
                        current.voterKey(),
                        partitionState.lastVoterSet()
                    )
                )
            );
        current.setLastOffset(leaderState.appendVotersRecord(newVoters, currentTimeMs));
        if (!current.ackWhenCommitted()) {
            // complete the future to send response, but do not reset the state,
            // since the new voter set is not yet committed
            current.future().complete(RaftUtil.addVoterResponse(Errors.NONE, null));
        }
        return true;
    }

    public void highWatermarkUpdated(LeaderState<?> leaderState) {
        leaderState.addVoterHandlerState().ifPresent(current -> {
            leaderState.highWatermark().ifPresent(highWatermark -> {
                current.lastOffset().ifPresent(lastOffset -> {
                    if (highWatermark.offset() > lastOffset) {
                        // VotersRecord with the added voter was committed; complete the RPC
                        leaderState.resetAddVoterHandlerState(Errors.NONE, null, Optional.empty());
                    }
                });
            });
        });
    }

    private ApiVersionsRequestData buildApiVersionsRequest() {
        return new ApiVersionsRequest.Builder().build().data();
    }

    private boolean validVersionRange(
        KRaftVersion finalizedVersion,
        Optional<ApiVersionsResponseData.SupportedFeatureKey> supportedKraftVersions
    ) {
        return supportedKraftVersions.isPresent() &&
            (supportedKraftVersions.get().minVersion() <= finalizedVersion.featureLevel() &&
             supportedKraftVersions.get().maxVersion() >= finalizedVersion.featureLevel());
    }
}