Skip to content

MINOR: Style fixes to KafkaRaftClient #10809

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
abbccdda merged 1 commit into from
Jun 4, 2021
Merged

MINOR: Style fixes to KafkaRaftClient #10809

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 21 additions & 31 deletions raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -309,17 +309,15 @@ private void updateListenersProgress(List<ListenerContext> listenerContexts, lon
for (ListenerContext listenerContext : listenerContexts) {
listenerContext.nextExpectedOffset().ifPresent(nextExpectedOffset -> {
if (nextExpectedOffset < log.startOffset() && nextExpectedOffset < highWatermark) {
SnapshotReader<T> snapshot = latestSnapshot().orElseThrow(() -> {
return new IllegalStateException(
String.format(
"Snapshot expected since next offset of %s is %s, log start offset is %s and high-watermark is %s",
listenerContext.listener.getClass().getTypeName(),
nextExpectedOffset,
log.startOffset(),
highWatermark
)
);
});
SnapshotReader<T> snapshot = latestSnapshot().orElseThrow(() -> new IllegalStateException(
String.format(
"Snapshot expected since next offset of %s is %s, log start offset is %s and high-watermark is %s",
listenerContext.listener.getClass().getTypeName(),
nextExpectedOffset,
log.startOffset(),
highWatermark
)
));
listenerContext.fireHandleSnapshot(snapshot);
}
});
Expand Down Expand Up @@ -347,14 +345,10 @@ private Optional<SnapshotReader<T>> latestSnapshot() {
private void maybeFireHandleCommit(long baseOffset, int epoch, List<T> records) {
for (ListenerContext listenerContext : listenerContexts) {
OptionalLong nextExpectedOffsetOpt = listenerContext.nextExpectedOffset();
if (!nextExpectedOffsetOpt.isPresent()) {
continue;
}

long nextExpectedOffset = nextExpectedOffsetOpt.getAsLong();
if (nextExpectedOffset == baseOffset) {
listenerContext.fireHandleCommit(baseOffset, epoch, records);
}
nextExpectedOffsetOpt.ifPresent(nextOffset -> {
if (nextOffset == baseOffset)
listenerContext.fireHandleCommit(baseOffset, epoch, records);
});
}
}

Expand Down Expand Up @@ -388,7 +382,6 @@ public void initialize() {
if (quorum.isVoter()
&& quorum.remoteVoters().isEmpty()
&& !quorum.isCandidate()) {

transitionToCandidate(currentTimeMs);
}
} catch (IOException e) {
Expand Down Expand Up @@ -449,7 +442,7 @@ private void onBecomeLeader(long currentTimeMs) throws IOException {
}

private void flushLeaderLog(LeaderState<T> state, long currentTimeMs) {
// We update the end offset before flushing so that parked fetches can return sooner
// We update the end offset before flushing so that parked fetches can return sooner.
updateLeaderEndOffsetAndTimestamp(state, currentTimeMs);
log.flush();
}
Expand Down Expand Up @@ -502,11 +495,11 @@ private void onBecomeFollower(long currentTimeMs) {
resetConnections();

// After becoming a follower, we need to complete all pending fetches so that
// they can be resent to the leader without waiting for their expiration
// they can be re-sent to the leader without waiting for their expirations
fetchPurgatory.completeAllExceptionally(new NotLeaderOrFollowerException(
"Cannot process the fetch request because the node is no longer the leader."));

// Clearing the append purgatory should complete all future exceptionally since this node is no longer the leader
// Clearing the append purgatory should complete all futures exceptionally since this node is no longer the leader
appendPurgatory.completeAllExceptionally(new NotLeaderOrFollowerException(
"Failed to receive sufficient acknowledgments for this append before leader change."));
}
Expand Down Expand Up @@ -552,7 +545,7 @@ private VoteResponseData handleVoteRequest(
}

if (!hasValidTopicPartition(request, log.topicPartition())) {
// Until we support multi-raft, we treat topic partition mismatches as invalid requests
// Until we support multi-raft, we treat individual topic partition mismatches as invalid requests
return new VoteResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
}

Expand Down Expand Up @@ -638,7 +631,6 @@ private boolean handleVoteResponse(
binaryExponentialElectionBackoffMs(state.retries())
);
}

}
} else {
logger.debug("Ignoring vote response {} since we are no longer a candidate in epoch {}",
Expand Down Expand Up @@ -1072,7 +1064,7 @@ private boolean handleFetchResponse(
FetchResponseData.EpochEndOffset divergingEpoch = partitionResponse.divergingEpoch();
if (divergingEpoch.epoch() >= 0) {
// The leader is asking us to truncate before continuing
OffsetAndEpoch divergingOffsetAndEpoch = new OffsetAndEpoch(
final OffsetAndEpoch divergingOffsetAndEpoch = new OffsetAndEpoch(
divergingEpoch.endOffset(), divergingEpoch.epoch());

state.highWatermark().ifPresent(highWatermark -> {
Expand Down Expand Up @@ -1104,7 +1096,7 @@ private boolean handleFetchResponse(
);
return false;
} else {
OffsetAndEpoch snapshotId = new OffsetAndEpoch(
final OffsetAndEpoch snapshotId = new OffsetAndEpoch(
partitionResponse.snapshotId().endOffset(),
partitionResponse.snapshotId().epoch()
);
Expand Down Expand Up @@ -1193,7 +1185,7 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
*/
private FetchSnapshotResponseData handleFetchSnapshotRequest(
RaftRequest.Inbound requestMetadata
) throws IOException {
) {
FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;

if (!hasValidClusterId(data.clusterId())) {
Expand Down Expand Up @@ -2038,7 +2030,7 @@ private long pollFollowerAsObserver(FollowerState state, long currentTimeMs) thr
}
}

private long maybeSendFetchOrFetchSnapshot(FollowerState state, long currentTimeMs) throws IOException {
private long maybeSendFetchOrFetchSnapshot(FollowerState state, long currentTimeMs) {
final Supplier<ApiMessage> requestSupplier;

if (state.fetchingSnapshot().isPresent()) {
Expand Down Expand Up @@ -2465,7 +2457,5 @@ public synchronized void onClose(BatchReader<T> reader) {
wakeup();
}
}

}

}