Skip to content

KAFKA-5758: Don't fail fetch request if replica is no longer a follower for a partition #3954

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

Closed
ijuma wants to merge 9 commits into from
Closed

KAFKA-5758: Don't fail fetch request if replica is no longer a follower for a partition #3954

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
09ece7e
KAFKA-5758: Don't fail fetch request if replica is no longer a follow…
ijuma Sep 25, 2017
5361233
Centralise some of the offline partition handling
ijuma Sep 25, 2017
4eb1774
Remove unnecessary tryCompleteDelayedProduce
ijuma Sep 25, 2017
a533a7d
Add test
ijuma Sep 25, 2017
011cdf7
Return updated hw in the immediate return case
ijuma Sep 25, 2017
b2e52fe
Fix test failures in `GroupMetadataManagerTest`
ijuma Sep 25, 2017
46c9397
Tweak `ReplicaManager.fetchMessages` to make it clearer and remove re…
ijuma Oct 1, 2017
45f52b3
Tweak `lowWatermarkIfLeader` to be a bit more direct.
ijuma Oct 1, 2017
babd8a0
Return empty fetch response if follower is not one of the assigned re…
ijuma Oct 1, 2017
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
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/admin/AdminUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -639,7 +639,7 @@ object AdminUtils extends Logging with AdminUtilities {
def fetchEntityConfig(zkUtils: ZkUtils, rootEntityType: String, sanitizedEntityName: String): Properties = {
val entityConfigPath = getEntityConfigPath(rootEntityType, sanitizedEntityName)
// readDataMaybeNull returns Some(null) if the path exists, but there is no data
val str: String = zkUtils.readDataMaybeNull(entityConfigPath)._1.orNull
val str = zkUtils.readDataMaybeNull(entityConfigPath)._1.orNull
val props = new Properties()
if (str != null) {
Json.parseFull(str).foreach { jsValue =>
Expand Down
71 changes: 34 additions & 37 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 @@ -22,9 +22,8 @@ import java.util.concurrent.locks.ReentrantReadWriteLock
import com.yammer.metrics.core.Gauge
import kafka.admin.AdminUtils
import kafka.api.LeaderAndIsr
import kafka.common.NotAssignedReplicaException
import kafka.controller.KafkaController
import kafka.log.LogConfig
import kafka.log.{LogAppendInfo, LogConfig}
import kafka.metrics.KafkaMetricsGroup
import kafka.server._
import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
Expand Down Expand Up @@ -264,37 +263,31 @@ class Partition(val topic: String,
}

/**
* Update the log end offset of a certain replica of this partition
* Update the the follower's state in the leader based on the last fetch request. See
* [[kafka.cluster.Replica#updateLogReadResult]] for details.
*
* @return true if the leader's log start offset or high watermark have been updated
*/
def updateReplicaLogReadResult(replicaId: Int, logReadResult: LogReadResult) {
getReplica(replicaId) match {
case Some(replica) =>
// No need to calculate low watermark if there is no delayed DeleteRecordsRequest
val oldLeaderLW = if (replicaManager.delayedDeleteRecordsPurgatory.delayed > 0) lowWatermarkIfLeader else -1L
replica.updateLogReadResult(logReadResult)
val newLeaderLW = if (replicaManager.delayedDeleteRecordsPurgatory.delayed > 0) lowWatermarkIfLeader else -1L
// check if the LW of the partition has incremented
// since the replica's logStartOffset may have incremented
val leaderLWIncremented = newLeaderLW > oldLeaderLW
// check if we need to expand ISR to include this replica
// if it is not in the ISR yet
val leaderHWIncremented = maybeExpandIsr(replicaId, logReadResult)

// some delayed operations may be unblocked after HW or LW changed
if (leaderLWIncremented || leaderHWIncremented)
tryCompleteDelayedRequests()

debug("Recorded replica %d log end offset (LEO) position %d."
.format(replicaId, logReadResult.info.fetchOffsetMetadata.messageOffset))
case None =>
throw new NotAssignedReplicaException(("Leader %d failed to record follower %d's position %d since the replica" +
" is not recognized to be one of the assigned replicas %s for partition %s.")
.format(localBrokerId,
replicaId,
logReadResult.info.fetchOffsetMetadata.messageOffset,
assignedReplicas.map(_.brokerId).mkString(","),
topicPartition))
}
def updateReplicaLogReadResult(replica: Replica, logReadResult: LogReadResult): Boolean = {
val replicaId = replica.brokerId
// No need to calculate low watermark if there is no delayed DeleteRecordsRequest
val oldLeaderLW = if (replicaManager.delayedDeleteRecordsPurgatory.delayed > 0) lowWatermarkIfLeader else -1L
replica.updateLogReadResult(logReadResult)
val newLeaderLW = if (replicaManager.delayedDeleteRecordsPurgatory.delayed > 0) lowWatermarkIfLeader else -1L
// check if the LW of the partition has incremented
// since the replica's logStartOffset may have incremented
val leaderLWIncremented = newLeaderLW > oldLeaderLW
// check if we need to expand ISR to include this replica
// if it is not in the ISR yet
val leaderHWIncremented = maybeExpandIsr(replicaId, logReadResult)

val result = leaderLWIncremented || leaderHWIncremented
// some delayed operations may be unblocked after HW or LW changed
if (result)

@junrao junrao Sep 30, 2017

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.

This seems to be an existing problem, but it seems that we only need to call tryCompleteDelayedRequests() when the HW increments.

@ijuma ijuma Oct 1, 2017

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.

I think tryCompleteDelayedDeleteRecords is affected by leaderLWIncremented.

private def tryCompleteDelayedRequests() {
  val requestKey = new TopicPartitionOperationKey(topicPartition)
  replicaManager.tryCompleteDelayedFetch(requestKey)
  replicaManager.tryCompleteDelayedProduce(requestKey)
  replicaManager.tryCompleteDelayedDeleteRecords(requestKey)
}

We could make the tryComplete calls more specific depending on what changed, if you think that's worth it.

tryCompleteDelayedRequests()

debug(s"Recorded replica $replicaId log end offset (LEO) position ${logReadResult.info.fetchOffsetMetadata.messageOffset}.")
result
}

/**
Expand All @@ -305,7 +298,9 @@ class Partition(val topic: String,
* even if its log end offset is >= HW. However, to be consistent with how the follower determines
* whether a replica is in-sync, we only check HW.
*
* This function can be triggered when a replica's LEO has incremented
* This function can be triggered when a replica's LEO has incremented.
*
* @return true if the high watermark has been updated
*/
def maybeExpandIsr(replicaId: Int, logReadResult: LogReadResult): Boolean = {
inWriteLock(leaderIsrUpdateLock) {
Expand All @@ -314,7 +309,7 @@ class Partition(val topic: String,
case Some(leaderReplica) =>
val replica = getReplica(replicaId).get
val leaderHW = leaderReplica.highWatermark
if(!inSyncReplicas.contains(replica) &&
if (!inSyncReplicas.contains(replica) &&
assignedReplicas.map(_.brokerId).contains(replicaId) &&
replica.logEndOffset.offsetDiff(leaderHW) >= 0) {
val newInSyncReplicas = inSyncReplicas + replica
Expand Down Expand Up @@ -421,8 +416,10 @@ class Partition(val topic: String,
def lowWatermarkIfLeader: Long = {
if (!isLeaderReplicaLocal)
throw new NotLeaderForPartitionException("Leader not local for partition %s on broker %d".format(topicPartition, localBrokerId))
assignedReplicas.filter(replica =>
replicaManager.metadataCache.isBrokerAlive(replica.brokerId)).map(_.logStartOffset).reduceOption(_ min _).getOrElse(0L)
val logStartOffsets = assignedReplicas.collect {
case replica if replicaManager.metadataCache.isBrokerAlive(replica.brokerId) => replica.logStartOffset
}
CoreUtils.min(logStartOffsets, 0L)

@tedyu tedyu Oct 2, 2017

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.

Wouldn't this return 0 in most cases ?

@ijuma ijuma Oct 2, 2017

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.

No, it only returns 0 if the collection is empty.

}

/**
Expand Down Expand Up @@ -485,7 +482,7 @@ class Partition(val topic: String,
laggingReplicas
}

def appendRecordsToLeader(records: MemoryRecords, isFromClient: Boolean, requiredAcks: Int = 0) = {
def appendRecordsToLeader(records: MemoryRecords, isFromClient: Boolean, requiredAcks: Int = 0): LogAppendInfo = {
val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
leaderReplicaIfLocal match {
case Some(leaderReplica) =>
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/cluster/Replica.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ class Replica(val brokerId: Int,
* fetch request is always smaller than the leader's LEO, which can happen if small produce requests are received at
* high frequency.
*/
def updateLogReadResult(logReadResult : LogReadResult) {
def updateLogReadResult(logReadResult: LogReadResult) {
if (logReadResult.info.fetchOffsetMetadata.messageOffset >= logReadResult.leaderLogEndOffset)
_lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, logReadResult.fetchTimeMs)
else if (logReadResult.info.fetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset)
Expand Down
3 changes: 1 addition & 2 deletions core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -697,8 +697,7 @@ class GroupMetadataManager(brokerId: Int,
val timestampType = TimestampType.CREATE_TIME
val timestamp = time.milliseconds()

val partitionOpt = replicaManager.getPartition(appendPartition).filter(_ ne ReplicaManager.OfflinePartition)
partitionOpt.foreach { partition =>
replicaManager.nonOfflinePartition(appendPartition).foreach { partition =>
val tombstones = ListBuffer.empty[SimpleRecord]
removedOffsets.foreach { case (topicPartition, offsetAndMetadata) =>
trace(s"Removing expired/deleted offset and metadata for $groupId, $topicPartition: $offsetAndMetadata")
Expand Down
Loading