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

Jump to bottom

KAFKA-9769: Finish operations for leaderEpoch-updated partitions up to point ZK Exception #8479

Merged
merged 12 commits into from Jul 6, 2020
Merged

KAFKA-9769: Finish operations for leaderEpoch-updated partitions up to point ZK Exception #8479

Changes from 10 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
45378cd
Finish operations for leaderEpoch-updated partitions up to point ZK e…
Apr 14, 2020
7de15ed
Merge branch 'trunk' of https://github.com/apache/kafka into andchoi_…
Jun 4, 2020
26c6fba
Merge branch 'trunk' of https://github.com/apache/kafka into andchoi_…
Jun 11, 2020
b6d405c
populate the response map with error
Jun 12, 2020
4491d3b
Merge branch 'trunk' of https://github.com/apache/kafka into andchoi_…
Jun 16, 2020
9647461
Merge branch 'andchoi_zkCatch' of https://github.com/andrewchoi5/kafk…
Jun 16, 2020
dfa33eb
remove sending back the error code. since the controller only checks …
Jun 16, 2020
320a567
change the logic to partition.scala
Jun 26, 2020
c3300a9
remove import
Jun 26, 2020
889e375
accurate error log statement
Jun 26, 2020
f95844a
removal of error and keep the state change log error
Jun 26, 2020
1851420
remove this keyword
Jul 1, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
23 changes: 21 additions & 2 deletions core/src/main/scala/kafka/cluster/Partition.scala
View file
Open in desktop
Expand Up @@ -29,6 +29,7 @@ import kafka.server.checkpoints.OffsetCheckpoints
import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
import kafka.utils._
import kafka.zk.{AdminZkClient, KafkaZkClient}
import kafka.zookeeper.ZooKeeperClientException
import org.apache.kafka.common.errors._
import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
import org.apache.kafka.common.protocol.Errors
Expand Down Expand Up @@ -499,7 +500,16 @@ class Partition(val topicPartition: TopicPartition,
addingReplicas = addingReplicas,
removingReplicas = removingReplicas
)
createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints)
try {
this.createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints)
andrewchoi5 marked this conversation as resolved.
Show resolved Hide resolved
} catch {
case e: ZooKeeperClientException =>
stateChangeLogger.info(s"A ZooKeeper client exception has occurred and makeLeader will be skipping the " +
s"state change for the partition with leader epoch: $leaderEpoch ", e)
error(s"ZooKeeper client error occurred while this partition was becoming the leader for $topicPartition.", e)
andrewchoi5 marked this conversation as resolved.
Show resolved Hide resolved

return false
}

val leaderLog = localLogOrException
val leaderEpochStartOffset = leaderLog.logEndOffset
Expand Down Expand Up @@ -570,7 +580,16 @@ class Partition(val topicPartition: TopicPartition,
addingReplicas = partitionState.addingReplicas.asScala.map(_.toInt),
removingReplicas = partitionState.removingReplicas.asScala.map(_.toInt)
)
createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints)
try {
this.createLogIfNotExists(partitionState.isNew, isFutureReplica = false, highWatermarkCheckpoints)
} catch {
case e: ZooKeeperClientException =>
stateChangeLogger.info(s"A ZooKeeper client exception has occurred. makeFollower will be skipping the " +
s"state change for the partition $topicPartition with leader epoch: $leaderEpoch.", e)
error(s"ZooKeeper client error occurred while this partition was becoming the follower for $topicPartition.", e)

return false
}

val followerLog = localLogOrException
val leaderEpochEndOffset = followerLog.logEndOffset
Expand Down