Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
15 changes: 11 additions & 4 deletions raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -126,9 +126,16 @@
* gracefully resign from the current epoch. This causes remaining voters to immediately
* begin a new election.
*
* 4) {@link FetchRequestData}: This is the same as the usual Fetch API in Kafka, but we piggyback
* some additional metadata on responses (i.e. current leader and epoch). Unlike partition replication,
* we also piggyback truncation detection on this API rather than through a separate truncation state.
* 4) {@link FetchRequestData}: This is the same as the usual Fetch API in Kafka, but we add snapshot
* check before responding, and we also piggyback some additional metadata on responses (i.e. current
* leader and epoch). Unlike partition replication, we also piggyback truncation detection on this API
* rather than through a separate truncation state.
*
* 5) {@link FetchSnapshotRequestData}: Sent by the follower to the epoch leader in order to fetch a snapshot.
* This happens when a FetchResponse includes a snapshot ID due to the follower's log end offset being less
* than the leader's log start offset. This API is similar to the Fetch API since the snapshot is stored
* as FileRecords, but we use {@link UnalignedRecords} in FetchSnapshotResponse because the records
* are not necessarily offset-aligned.
*
*/
public class KafkaRaftClient<T> implements RaftClient<T> {
Expand Down Expand Up @@ -1359,7 +1366,7 @@ private boolean handleFetchSnapshotResponse(
} else {
throw new IllegalStateException(
String.format(
"Full log trunctation expected but didn't happen. Snapshot of %s, log end offset %s, last fetched %s",
"Full log truncation expected but didn't happen. Snapshot of %s, log end offset %s, last fetched %s",
snapshot.snapshotId(),
log.endOffset(),
log.lastFetchedEpoch()
Expand Down
7 changes: 3 additions & 4 deletions raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,9 @@ public interface NetworkChannel extends Closeable {
int newCorrelationId();

/**
* Send an outbound message. This could be either an outbound request
* (i.e. an instance of {@link org.apache.kafka.raft.RaftRequest.Outbound})
* or a response to a request that was received through {@link #receive(long)}
Comment thread
dengziming marked this conversation as resolved.
* (i.e. an instance of {@link org.apache.kafka.raft.RaftResponse.Outbound}).
* Send an outbound request message.
*
* @param request outbound request to send
*/
void send(RaftRequest.Outbound request);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,7 @@ default long truncateToEndOffset(OffsetAndEpoch endOffset) {
Optional<OffsetAndEpoch> oldestSnapshotId();

/**
* Notifies the replicted log when a new snapshot is available.
* Notifies the replicated log when a new snapshot is available.
*/
void onSnapshotFrozen(OffsetAndEpoch snapshotId);

Expand Down