From dc7d9fc4790c5f77d412923ca9385f47373df2ef Mon Sep 17 00:00:00 2001 From: Divij Vaidya Date: Wed, 22 Jun 2022 13:29:22 +0200 Subject: [PATCH 1/3] Make scheduleAppend() API for LocalLogManager consistent with the interface contract --- .../apache/kafka/metalog/LocalLogManager.java | 52 +++++++++++++++---- .../org/apache/kafka/raft/RaftClient.java | 5 +- .../raft/internals/BatchAccumulator.java | 8 +-- 3 files changed, 50 insertions(+), 15 deletions(-) diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java index 36fe3ec2c1c58..dc85e1beec388 100644 --- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java +++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java @@ -31,6 +31,7 @@ import org.apache.kafka.raft.LeaderAndEpoch; import org.apache.kafka.raft.OffsetAndEpoch; import org.apache.kafka.raft.RaftClient; +import org.apache.kafka.raft.errors.NotLeaderException; import org.apache.kafka.raft.internals.MemoryBatchReader; import org.apache.kafka.server.common.ApiMessageAndVersion; import org.apache.kafka.snapshot.MockRawSnapshotReader; @@ -226,16 +227,19 @@ synchronized long tryAppend(int nodeId, int epoch, List ba } synchronized long tryAppend(int nodeId, int epoch, LocalBatch batch) { - if (epoch != leader.epoch()) { - log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " + - "match the current leader epoch of {}.", nodeId, epoch, leader.epoch()); - return Long.MAX_VALUE; - } if (!leader.isLeader(nodeId)) { - log.trace("tryAppend(nodeId={}, epoch={}): the given node id does not " + - "match the current leader id of {}.", nodeId, epoch, leader.leaderId()); - return Long.MAX_VALUE; + log.debug("tryAppend(nodeId={}, epoch={}): the given node id does not " + + "match the current leader id of {}.", nodeId, epoch, leader.leaderId()); + throw new NotLeaderException("Append failed because the replication is not the current leader"); + } + + if (epoch < leader.epoch()) { + throw new NotLeaderException("Append failed because the epoch because the given epoch is stale"); + } else if (epoch > leader.epoch()) { + throw new IllegalArgumentException("Attempt to append from epoch " + epoch + + " which is larger than the current epoch " + leader.epoch()); } + log.trace("tryAppend(nodeId={}): appending {}.", nodeId, batch); long offset = append(batch); electLeaderIfNeeded(); @@ -723,9 +727,35 @@ public long scheduleAtomicAppend(int epoch, List batch) { @Override public void resign(int epoch) { - LeaderAndEpoch curLeader = leader; - LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), curLeader.epoch() + 1); - shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader)); + if (epoch < 0) { + throw new IllegalArgumentException("Attempt to resign from an invalid negative epoch " + epoch); + } + + LeaderAndEpoch leaderAndEpoch = leaderAndEpoch(); + int currentEpoch = leaderAndEpoch.epoch(); + + if (epoch > currentEpoch) { + throw new IllegalArgumentException("Attempt to resign from epoch " + epoch + + " which is larger than the current epoch " + currentEpoch); + } else if (epoch < currentEpoch) { + // If the passed epoch is smaller than the current epoch, then it might mean + // that the listener has not been notified about a leader change that already + // took place. In this case, we consider the call as already fulfilled and + // take no further action. + log.debug("Ignoring call to resign from epoch {} since it is smaller than the " + + "current epoch {}", epoch, currentEpoch); + return; + } + + LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), currentEpoch + 1); + try { + shared.tryAppend(nodeId, currentEpoch, new LeaderChangeBatch(nextLeader)); + } catch (NotLeaderException exp) { + // the leader epoch has already advanced. resign is a no op. + log.debug("Ignoring call to resign from epoch {}. Either we are not the leader or the provided epoch is " + + "smaller than the current epoch {}", epoch, currentEpoch); + return; + } } @Override diff --git a/raft/src/main/java/org/apache/kafka/raft/RaftClient.java b/raft/src/main/java/org/apache/kafka/raft/RaftClient.java index 952cc60a38793..51f859c6c0731 100644 --- a/raft/src/main/java/org/apache/kafka/raft/RaftClient.java +++ b/raft/src/main/java/org/apache/kafka/raft/RaftClient.java @@ -205,8 +205,11 @@ default void beginShutdown() {} * Notification of successful resignation can be observed through * {@link Listener#handleLeaderChange(LeaderAndEpoch)}. * - * @param epoch the epoch to resign from. If this does not match the current epoch, this + * @param epoch the epoch to resign from. If this epoch is smaller than the current epoch, this * call will be ignored. + * + * @throws IllegalArgumentException - if the passed epoch is invalid (negative or greater than current) or + * if the listener is not the leader associated with this epoch. */ void resign(int epoch); diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java b/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java index 77efd2f7e1579..790eae8ba36a0 100644 --- a/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java +++ b/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java @@ -104,7 +104,8 @@ public BatchAccumulator( * @throws RecordBatchTooLargeException if the size of one record T is greater than the maximum * batch size; if this exception is throw some of the elements in records may have * been committed - * @throws NotLeaderException if the epoch doesn't match the leader epoch + * @throws NotLeaderException if the epoch is less than the leader epoch + * @throws IllegalArgumentException if the epoch is invalid (greater than the leader epoch) * @throws BufferAllocationException if we failed to allocate memory for the records */ public long append(int epoch, List records) { @@ -123,7 +124,8 @@ public long append(int epoch, List records) { * @throws RecordBatchTooLargeException if the size of the records is greater than the maximum * batch size; if this exception is throw none of the elements in records were * committed - * @throws NotLeaderException if the epoch doesn't match the leader epoch + * @throws NotLeaderException if the epoch is less than the leader epoch + * @throws IllegalArgumentException if the epoch is invalid (greater than the leader epoch) * @throws BufferAllocationException if we failed to allocate memory for the records */ public long appendAtomic(int epoch, List records) { @@ -132,7 +134,7 @@ public long appendAtomic(int epoch, List records) { private long append(int epoch, List records, boolean isAtomic) { if (epoch < this.epoch) { - throw new NotLeaderException("Append failed because the epoch doesn't match"); + throw new NotLeaderException("Append failed because the given epoch is stale"); } else if (epoch > this.epoch) { throw new IllegalArgumentException("Attempt to append from epoch " + epoch + " which is larger than the current epoch " + this.epoch); From 7569524ee8ec182cf8f7f5db463da6b52b74a3f5 Mon Sep 17 00:00:00 2001 From: Divij Vaidya Date: Tue, 5 Jul 2022 23:46:00 +0200 Subject: [PATCH 2/3] Add current and older epoch to the exception message --- .../org/apache/kafka/raft/internals/BatchAccumulator.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java b/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java index 790eae8ba36a0..44654d81ca6e1 100644 --- a/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java +++ b/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java @@ -134,7 +134,9 @@ public long appendAtomic(int epoch, List records) { private long append(int epoch, List records, boolean isAtomic) { if (epoch < this.epoch) { - throw new NotLeaderException("Append failed because the given epoch is stale"); + throw new NotLeaderException( + String.format("Append failed because the given epoch %d is older than current epoch %d", + epoch, this.epoch)); } else if (epoch > this.epoch) { throw new IllegalArgumentException("Attempt to append from epoch " + epoch + " which is larger than the current epoch " + this.epoch); From ab1c176deb1a1c823f75be036250960cf29d0381 Mon Sep 17 00:00:00 2001 From: Divij Vaidya Date: Tue, 5 Jul 2022 23:50:33 +0200 Subject: [PATCH 3/3] Add current and older epoch to the exception message --- .../test/java/org/apache/kafka/metalog/LocalLogManager.java | 3 ++- .../org/apache/kafka/raft/internals/BatchAccumulator.java | 5 ++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java index dc85e1beec388..c8e39ae32896e 100644 --- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java +++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java @@ -234,7 +234,8 @@ synchronized long tryAppend(int nodeId, int epoch, LocalBatch batch) { } if (epoch < leader.epoch()) { - throw new NotLeaderException("Append failed because the epoch because the given epoch is stale"); + throw new NotLeaderException("Append failed because the given epoch " + epoch + " is stale. " + + "Current leader epoch = " + leader.epoch()); } else if (epoch > leader.epoch()) { throw new IllegalArgumentException("Attempt to append from epoch " + epoch + " which is larger than the current epoch " + leader.epoch()); diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java b/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java index 44654d81ca6e1..323f393aebe4c 100644 --- a/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java +++ b/raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java @@ -134,9 +134,8 @@ public long appendAtomic(int epoch, List records) { private long append(int epoch, List records, boolean isAtomic) { if (epoch < this.epoch) { - throw new NotLeaderException( - String.format("Append failed because the given epoch %d is older than current epoch %d", - epoch, this.epoch)); + throw new NotLeaderException("Append failed because the given epoch " + epoch + " is stale. " + + "Current leader epoch = " + this.epoch()); } else if (epoch > this.epoch) { throw new IllegalArgumentException("Attempt to append from epoch " + epoch + " which is larger than the current epoch " + this.epoch);