Skip to content

KAFKA-13916; Fenced replicas should not be allowed to join the ISR in KRaft (KIP-841, Part 2) #12181

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
dajac merged 15 commits into from
Jun 14, 2022
Merged

KAFKA-13916; Fenced replicas should not be allowed to join the ISR in KRaft (KIP-841, Part 2) #12181

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
b6cb295
KAFKA-13916; Fenced replicas should not be allowed to join the ISR in…
dajac May 19, 2022
e04cb33
add unit test
dajac Jun 3, 2022
7a6e3cd
refactor
dajac Jun 3, 2022
384681a
refactor
dajac Jun 3, 2022
4d0621f
refactor
dajac Jun 3, 2022
c024f96
introduce NEW_LEADER_ELECTED error
dajac Jun 3, 2022
827512b
cleanup
dajac Jun 3, 2022
66a0df5
address comments
dajac Jun 8, 2022
3b869aa
address reviews
dajac Jun 9, 2022
6addd52
ensure that AlterPartition v2 is not advertised when the controller i…
dajac Jun 9, 2022
2743a96
Revert "ensure that AlterPartition v2 is not advertised when the cont…
dajac Jun 10, 2022
e567abe
address reviews and refactor
dajac Jun 10, 2022
ef38b4f
address reviews
dajac Jun 10, 2022
a717d4a
doc
dajac Jun 10, 2022
b15a6cd
address minor comments
dajac Jun 14, 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
8 changes: 8 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,14 @@ public static class Builder extends AbstractRequest.Builder<AlterPartitionReques

private final AlterPartitionRequestData data;

/**
* Constructs a builder for AlterPartitionRequest.
*
* @param data The data to be sent. Note that because the version of the
* request is not known at this time, it is expected that all
* topics have a topic id and a topic name set.
* @param canUseTopicIds True if version 2 and above can be used.
*/
public Builder(AlterPartitionRequestData data, boolean canUseTopicIds) {
Comment thread
dajac marked this conversation as resolved.
super(
ApiKeys.ALTER_PARTITION,
Expand Down
51 changes: 29 additions & 22 deletions core/src/main/scala/kafka/cluster/Partition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,6 @@ sealed trait PartitionState {
* the high watermark as well as determining which replicas are required for acks=all produce requests.
*
* Only applicable as of IBP 2.7-IV2, for older versions this will return the committed ISR
*
*/
def maximalIsr: Set[Int]

Expand All @@ -159,7 +158,7 @@ sealed trait PartitionState {
}

sealed trait PendingPartitionChange extends PartitionState {
def lastCommittedState: PartitionState
def lastCommittedState: CommittedPartitionState
def sentLeaderAndIsr: LeaderAndIsr

override val leaderRecoveryState: LeaderRecoveryState = LeaderRecoveryState.RECOVERED
Expand All @@ -168,11 +167,11 @@ sealed trait PendingPartitionChange extends PartitionState {
}

case class PendingExpandIsr(
isr: Set[Int],
newInSyncReplicaId: Int,
sentLeaderAndIsr: LeaderAndIsr,
lastCommittedState: PartitionState
lastCommittedState: CommittedPartitionState
) extends PendingPartitionChange {
val isr = lastCommittedState.isr
val maximalIsr = isr + newInSyncReplicaId
val isInflight = true

Expand All @@ -191,11 +190,11 @@ case class PendingExpandIsr(
}

case class PendingShrinkIsr(
isr: Set[Int],
outOfSyncReplicaIds: Set[Int],
sentLeaderAndIsr: LeaderAndIsr,
lastCommittedState: PartitionState
lastCommittedState: CommittedPartitionState
) extends PendingPartitionChange {
val isr = lastCommittedState.isr
val maximalIsr = isr
val isInflight = true

Expand Down Expand Up @@ -869,7 +868,7 @@ class Partition(val topicPartition: TopicPartition,
val current = partitionState
!current.isInflight &&
!current.isr.contains(followerReplicaId) &&
isBrokerIsrEligible(followerReplicaId)
isReplicaIsrEligible(followerReplicaId)
}

private def isFollowerInSync(followerReplica: Replica): Boolean = {
Expand All @@ -879,10 +878,10 @@ class Partition(val topicPartition: TopicPartition,
}
}

private def isBrokerIsrEligible(brokerId: Int): Boolean = {
private def isReplicaIsrEligible(followerReplicaId: Int): Boolean = {
// In KRaft mode, only replicas which are not fenced nor in controlled shutdown are
// allowed to join the ISR. This does not apply to ZK mode.
!metadataCache.isBrokerFenced(brokerId) && !metadataCache.isBrokerInControlledShutdown(brokerId)
!metadataCache.isBrokerFenced(followerReplicaId) && !metadataCache.isBrokerShuttingDown(followerReplicaId)
}

/*
Expand Down Expand Up @@ -1534,10 +1533,12 @@ class Partition(val topicPartition: TopicPartition,
partitionEpoch
)
val updatedState = PendingExpandIsr(
partitionState.isr,
newInSyncReplicaId,
newLeaderAndIsr,
partitionState
// The current partition state must be of type CommittedPartitionState
// if we are here. CommittedPartitionState is the only one with `isInflight`
// equals to false.
partitionState.asInstanceOf[CommittedPartitionState]
Comment thread
dajac marked this conversation as resolved.
Outdated
)
partitionState = updatedState
updatedState
Expand All @@ -1556,10 +1557,12 @@ class Partition(val topicPartition: TopicPartition,
partitionEpoch
)
val updatedState = PendingShrinkIsr(
partitionState.isr,
outOfSyncReplicaIds,
newLeaderAndIsr,
partitionState
// The current partition state must be of type CommittedPartitionState
// if we are here. CommittedPartitionState is the only one with `isInflight`
// equals to false.
partitionState.asInstanceOf[CommittedPartitionState]
)
partitionState = updatedState
updatedState
Expand Down Expand Up @@ -1621,41 +1624,45 @@ class Partition(val topicPartition: TopicPartition,
case Errors.OPERATION_NOT_ATTEMPTED =>
// Since the operation was not attempted, it is safe to reset back to the committed state.
partitionState = proposedIsrState.lastCommittedState
Comment thread
dajac marked this conversation as resolved.
Outdated
debug(s"Failed to alter partition to $proposedIsrState since there is a pending AlterPartition still inflight. " +
info(s"Failed to alter partition to $proposedIsrState since there is a pending AlterPartition still inflight. " +
s"Partition state has been reset to the latest committed state $partitionState")
false
case Errors.INELIGIBLE_REPLICA =>
Comment thread
dajac marked this conversation as resolved.
Outdated
// Since the operation was rejected, it is safe to reset back to the committed state. This
// assumes that the current state was still the correct expected state.
// This is only raised in KRaft mode.
partitionState = proposedIsrState.lastCommittedState

@artemlivshits artemlivshits Jun 14, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

In KRaft mode, could the state be updated via metadata and applied concurrently such that processing this would override a concurrently updated last state?

@dajac dajac Jun 14, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

We rollback the previous partition state here only if the the partition state still matches our proposed partition state. If it does not, it means that the partition was updated via the metadata log in the mean time. This check is in submitAlterPartition before calling handleAlterPartitionError.

@artemlivshits artemlivshits Jun 14, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Sounds good. This invariant isn't immediately visible from this code so maybe a comment and / or assert would make it more clear.

debug(s"Failed to alter partition to $proposedIsrState since the controller rejected at least one replica " +
info(s"Failed to alter partition to $proposedIsrState since the controller rejected at least one replica " +
s"because it is ineligible to join the ISR. Partition state has been reset to the latest committed state $partitionState.")
false
case Errors.UNKNOWN_TOPIC_OR_PARTITION =>
debug(s"Failed to alter partition to $proposedIsrState since the controller doesn't know about " +
"this topic or partition. Giving up.")
"this topic or partition. Partition state may be out of thing, awaiting new the latest metadata.")
Comment thread
dajac marked this conversation as resolved.
Outdated
false
case Errors.UNKNOWN_TOPIC_ID =>
debug(s"Failed to alter partition to $proposedIsrState since the controller doesn't know about " +
"this topic. Giving up.")
"this topic. Partition state may be out of thing, awaiting new the latest metadata.")
false
case Errors.FENCED_LEADER_EPOCH =>
debug(s"Failed to alter partition to $proposedIsrState since the leader epoch is old. Giving up.")
debug(s"Failed to alter partition to $proposedIsrState since the leader epoch is old. " +
"Partition state may be out of thing, awaiting new the latest metadata.")
false
case Errors.INVALID_UPDATE_VERSION =>
debug(s"Failed to alter partition to $proposedIsrState because the partition epoch is invalid. Giving up.")
debug(s"Failed to alter partition to $proposedIsrState because the partition epoch is invalid. " +
"Partition state may be out of thing, awaiting new the latest metadata.")
false
case Errors.INVALID_REQUEST =>
debug(s"Failed to alter partition to $proposedIsrState because the request is invalid. Giving up.")
debug(s"Failed to alter partition to $proposedIsrState because the request is invalid. " +
"Partition state may be out of thing, awaiting new the latest metadata.")
false
case Errors.NEW_LEADER_ELECTED =>
// The operation completed successfully but this replica got removed from the replica set by the controller
// while completing a ongoing reassignment. This replica is no longer the leader but it does not know it
// yet. It should remain in the current pending state until the metadata overrides it.
// This is only raised in KRaft mode.
debug("The alter partition request successfully updated the partition state but this replica got " +
"removed from the replica set while completing a reassignment. Waiting on new metadata to clean up this replica.")
debug(s"The alter partition request successfully updated the partition state to $proposedIsrState but " +
"this replica got removed from the replica set while completing a reassignment. " +
"Waiting on new metadata to clean up this replica.")
false
case _ =>
warn(s"Failed to update ISR to $proposedIsrState due to unexpected $error. Retrying.")
Expand Down
Loading