Skip to content
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

KAFKA-14144: Compare AlterPartition LeaderAndIsr before fencing partition epoch #12489

Merged
merged 2 commits into from
Aug 9, 2022
Merged

KAFKA-14144: Compare AlterPartition LeaderAndIsr before fencing partition epoch #12489

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a6da5f5
Compare AlterPartition LeaderAndIsr before fencing partition epoch
splett2 Aug 6, 2022
77005e3
Fence AlterPartition requests with greater partition epochs
splett2 Aug 9, 2022
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
9 changes: 5 additions & 4 deletions core/src/main/scala/kafka/controller/KafkaController.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2339,14 +2339,15 @@ class KafkaController(val config: KafkaConfig,
if (newLeaderAndIsr.leaderEpoch != currentLeaderAndIsr.leaderEpoch) {
partitionResponses(tp) = Left(Errors.FENCED_LEADER_EPOCH)
None
} else if (newLeaderAndIsr.partitionEpoch < currentLeaderAndIsr.partitionEpoch) {
partitionResponses(tp) = Left(Errors.INVALID_UPDATE_VERSION)
None
} else if (newLeaderAndIsr.equalsIgnorePartitionEpoch(currentLeaderAndIsr)) {
// If a partition is already in the desired state, just return it
// this check must be done before fencing based on partition epoch to maintain idempotency
partitionResponses(tp) = Right(currentLeaderAndIsr)
None
} else if (newLeaderAndIsr.leaderRecoveryState == LeaderRecoveryState.RECOVERING && newLeaderAndIsr.isr.length > 1) {
} else if (newLeaderAndIsr.partitionEpoch < currentLeaderAndIsr.partitionEpoch) {
partitionResponses(tp) = Left(Errors.INVALID_UPDATE_VERSION)
None
} else if (newLeaderAndIsr.leaderRecoveryState == LeaderRecoveryState.RECOVERING && newLeaderAndIsr.isr.length > 1) {
partitionResponses(tp) = Left(Errors.INVALID_REQUEST)
info(
s"Rejecting AlterPartition from node $brokerId for $tp because leader is recovering and ISR is greater than 1: " +
Expand Down
78 changes: 45 additions & 33 deletions core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1002,47 +1002,59 @@ class ControllerIntegrationTest extends QuorumTestHarness {

val controller = getController().kafkaController
val leaderIsrAndControllerEpochMap = zkClient.getTopicPartitionStates(Seq(tp))
val newLeaderAndIsr = leaderIsrAndControllerEpochMap(tp).leaderAndIsr
val oldLeaderAndIsr = leaderIsrAndControllerEpochMap(tp).leaderAndIsr
val newIsr = List(oldLeaderAndIsr.leader)
val newPartitionEpoch = oldLeaderAndIsr.partitionEpoch + 1
val topicId = controller.controllerContext.topicIds(tp.topic)
val brokerId = otherBroker.config.brokerId
val brokerEpoch = controller.controllerContext.liveBrokerIdAndEpochs(otherBroker.config.brokerId)

// When re-sending the current ISR, we should not get and error or any ISR changes
val alterPartitionRequest = new AlterPartitionRequestData()
.setBrokerId(brokerId)
.setBrokerEpoch(brokerEpoch)
.setTopics(Seq(new AlterPartitionRequestData.TopicData()
.setTopicId(topicId)
.setPartitions(Seq(new AlterPartitionRequestData.PartitionData()
.setPartitionIndex(tp.partition)
.setLeaderEpoch(newLeaderAndIsr.leaderEpoch)
.setPartitionEpoch(newLeaderAndIsr.partitionEpoch)
.setNewIsr(newLeaderAndIsr.isr.map(Int.box).asJava)
.setLeaderRecoveryState(newLeaderAndIsr.leaderRecoveryState.value)
def sendAndVerifyAlterPartitionResponse(requestPartitionEpoch: Int): Unit = {
val alterPartitionRequest = new AlterPartitionRequestData()
.setBrokerId(brokerId)
.setBrokerEpoch(brokerEpoch)
.setTopics(Seq(new AlterPartitionRequestData.TopicData()
.setTopicId(topicId)
.setPartitions(Seq(new AlterPartitionRequestData.PartitionData()
.setPartitionIndex(tp.partition)
.setLeaderEpoch(oldLeaderAndIsr.leaderEpoch)
.setPartitionEpoch(requestPartitionEpoch)
.setNewIsr(newIsr.map(Int.box).asJava)
.setLeaderRecoveryState(oldLeaderAndIsr.leaderRecoveryState.value)
).asJava)
).asJava)
).asJava)

val future = new CompletableFuture[AlterPartitionResponseData]()
controller.eventManager.put(AlterPartitionReceived(
alterPartitionRequest,
AlterPartitionRequestData.HIGHEST_SUPPORTED_VERSION,
future.complete
))
val future = new CompletableFuture[AlterPartitionResponseData]()
controller.eventManager.put(AlterPartitionReceived(
alterPartitionRequest,
AlterPartitionRequestData.HIGHEST_SUPPORTED_VERSION,
future.complete
))

val expectedAlterPartitionResponse = new AlterPartitionResponseData()
.setTopics(Seq(new AlterPartitionResponseData.TopicData()
.setTopicId(topicId)
.setPartitions(Seq(new AlterPartitionResponseData.PartitionData()
.setPartitionIndex(tp.partition)
.setLeaderId(brokerId)
.setLeaderEpoch(newLeaderAndIsr.leaderEpoch)
.setPartitionEpoch(newLeaderAndIsr.partitionEpoch)
.setIsr(newLeaderAndIsr.isr.map(Int.box).asJava)
.setLeaderRecoveryState(newLeaderAndIsr.leaderRecoveryState.value)
// When re-sending an ISR update, we should not get and error or any ISR changes
val expectedAlterPartitionResponse = new AlterPartitionResponseData()
.setTopics(Seq(new AlterPartitionResponseData.TopicData()
.setTopicId(topicId)
.setPartitions(Seq(new AlterPartitionResponseData.PartitionData()
.setPartitionIndex(tp.partition)
.setLeaderId(brokerId)
.setLeaderEpoch(oldLeaderAndIsr.leaderEpoch)
.setPartitionEpoch(newPartitionEpoch)
.setIsr(newIsr.map(Int.box).asJava)
.setLeaderRecoveryState(oldLeaderAndIsr.leaderRecoveryState.value)
).asJava)
).asJava)
).asJava)
assertEquals(expectedAlterPartitionResponse, future.get(10, TimeUnit.SECONDS))
}

assertEquals(expectedAlterPartitionResponse, future.get(10, TimeUnit.SECONDS))
// send a request, expect the partition epoch to be incremented
sendAndVerifyAlterPartitionResponse(oldLeaderAndIsr.partitionEpoch)

// re-send the same request with various partition epochs (less/equal/greater than the current
// epoch), expect it to succeed while the partition epoch remains the same
sendAndVerifyAlterPartitionResponse(oldLeaderAndIsr.partitionEpoch)
sendAndVerifyAlterPartitionResponse(newPartitionEpoch)
sendAndVerifyAlterPartitionResponse(newPartitionEpoch + 1)
Copy link

@hachikuji hachikuji Aug 8, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do find it a little odd that the partition epoch is ignored completely when the ISR matches the desired state. We do have the leader epoch check, so at least we can ensure that an old leader won't be mislead into thinking that its change was successfully applied. How about a case when the request is sent to an old controller? Suppose a scenario like this:

  1. Controller A has leader=1, isr=[1,2], partition epoch=10
  2. Controller B is elected
  3. Leader sends AlterPartition(epoch=10) to B to remove 2 from ISR => partition epoch = 11
  4. Leader sends AlterPartition(epoch=11) to A to add 2 back to the ISR => A accepts, but there is no bump

I think this case is ruled out because the leader has to find the new controller and then revert back. The controller epoch probably would catch that case. What if we add a restart between steps 3 and 4? Would it be possible to find the old controller after restarting? Probably not, but I think I'd sleep better if we could at least reject requests where the partition epoch is greater than what the controller has in its cache. Does that make sense?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems fairly reasonable to me.

}

@Test
Expand Down Expand Up @@ -1100,7 +1112,6 @@ class ControllerIntegrationTest extends QuorumTestHarness {

assertAlterPartition(
partitionError = Errors.UNKNOWN_TOPIC_ID,
topicPartition = tp,
topicIdOpt = Some(Uuid.randomUuid())
)

Expand All @@ -1118,6 +1129,7 @@ class ControllerIntegrationTest extends QuorumTestHarness {

assertAlterPartition(
partitionError = Errors.INVALID_UPDATE_VERSION,
isr = Set(leaderId),
partitionEpoch = partitionEpoch - 1
)

Expand Down