From 15b344590573a3e8a9e1d4543f647554ab398b42 Mon Sep 17 00:00:00 2001 From: dengziming Date: Wed, 10 Feb 2021 20:10:04 +0800 Subject: [PATCH 1/4] MINOR: Add FetchSnapshot API doc in KafkaRaftClient --- .../java/org/apache/kafka/raft/KafkaRaftClient.java | 11 ++++++++--- .../java/org/apache/kafka/raft/NetworkChannel.java | 4 +--- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java index 09f8672d287cd..2d7aa8455920e 100644 --- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java +++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java @@ -126,9 +126,14 @@ * 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 fetch records, 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 when FetchResponse include a snapshot id, + * this is similar to the Fetch API since the snapshot is also stored as FileRecords, but we use + * {@link UnalignedRecords} in FetchSnapshotResponse because the records is not necessarily offset-aligned. * */ public class KafkaRaftClient implements RaftClient { diff --git a/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java b/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java index b88241b033c31..7b347dd2de76e 100644 --- a/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java +++ b/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java @@ -30,10 +30,8 @@ public interface NetworkChannel extends Closeable { int newCorrelationId(); /** - * Send an outbound message. This could be either an outbound request + * Send an outbound message. This could be 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)} - * (i.e. an instance of {@link org.apache.kafka.raft.RaftResponse.Outbound}). */ void send(RaftRequest.Outbound request); From 8b47703861c42bb14da7b4047d5d0cb3b2ea53af Mon Sep 17 00:00:00 2001 From: dengziming Date: Wed, 10 Feb 2021 20:14:54 +0800 Subject: [PATCH 2/4] minor change --- raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java index 2d7aa8455920e..dd7a5ae9bb713 100644 --- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java +++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java @@ -127,7 +127,7 @@ * begin a new election. * * 4) {@link FetchRequestData}: This is the same as the usual Fetch API in Kafka, but we add snapshot - * check before fetch records, and we also piggyback some additional metadata on responses (i.e. current + * 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. * From a96006d1a34dd8bd0c6ebf8c7fc346bbbffba4b7 Mon Sep 17 00:00:00 2001 From: dengziming Date: Thu, 11 Feb 2021 09:42:06 +0800 Subject: [PATCH 3/4] resolve jsancio's comments --- .../java/org/apache/kafka/raft/KafkaRaftClient.java | 10 ++++++---- .../java/org/apache/kafka/raft/NetworkChannel.java | 5 +++-- .../main/java/org/apache/kafka/raft/ReplicatedLog.java | 2 +- 3 files changed, 10 insertions(+), 7 deletions(-) diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java index dd7a5ae9bb713..fb5539e8f3ac8 100644 --- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java +++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java @@ -131,9 +131,11 @@ * 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 when FetchResponse include a snapshot id, - * this is similar to the Fetch API since the snapshot is also stored as FileRecords, but we use - * {@link UnalignedRecords} in FetchSnapshotResponse because the records is not necessarily offset-aligned. + * 5) {@link FetchSnapshotRequestData}: Sent by the follower to the epoch leader to fetch snapshot when + * FetchResponse include a snapshot id, this happens when the follower's log end offset is less than + * the leader's log start offset. This is similar to the Fetch API since the snapshot is also stored + * as FileRecords, but we use {@link UnalignedRecords} in FetchSnapshotResponse because the records is + * not necessarily offset-aligned. * */ public class KafkaRaftClient implements RaftClient { @@ -1364,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() diff --git a/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java b/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java index 7b347dd2de76e..f023955beba8e 100644 --- a/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java +++ b/raft/src/main/java/org/apache/kafka/raft/NetworkChannel.java @@ -30,8 +30,9 @@ public interface NetworkChannel extends Closeable { int newCorrelationId(); /** - * Send an outbound message. This could be an outbound request - * (i.e. an instance of {@link org.apache.kafka.raft.RaftRequest.Outbound}) + * Send an outbound request message. + * + * @param request outbound request to send */ void send(RaftRequest.Outbound request); diff --git a/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java b/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java index 714135dc9cab5..417b7690eb4ec 100644 --- a/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java +++ b/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java @@ -264,7 +264,7 @@ default long truncateToEndOffset(OffsetAndEpoch endOffset) { Optional 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); From ec97b411256b33027bb4e4e6c33c061ab7cd3d36 Mon Sep 17 00:00:00 2001 From: dengziming Date: Fri, 12 Feb 2021 10:23:34 +0800 Subject: [PATCH 4/4] Update raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java Co-authored-by: David Arthur --- .../java/org/apache/kafka/raft/KafkaRaftClient.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java index fb5539e8f3ac8..9b6eb576c4752 100644 --- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java +++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java @@ -131,11 +131,11 @@ * 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 to fetch snapshot when - * FetchResponse include a snapshot id, this happens when the follower's log end offset is less than - * the leader's log start offset. This is similar to the Fetch API since the snapshot is also stored - * as FileRecords, but we use {@link UnalignedRecords} in FetchSnapshotResponse because the records is - * not necessarily offset-aligned. + * 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 implements RaftClient {