{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/185664468","pull_request_review_id":117054575,"id":185664468,"diff_hunk":"@@ -276,18 +280,88 @@ abstract class AbstractFetcherThread(name: String,\n     *\n     * @param fetchOffsets the partitions to mark truncation complete\n     */\n-  private def markTruncationCompleteAndUpdateFetchOffset(fetchOffsets: Map[TopicPartition, Long]) {\n+  private def markTruncationCompleteAndUpdateFetchOffset(fetchOffsets: Map[TopicPartition, OffsetTruncationState]) {\n     val newStates: Map[TopicPartition, PartitionFetchState] = partitionStates.partitionStates.asScala\n       .map { state =>\n         val maybeTruncationComplete = fetchOffsets.get(state.topicPartition()) match {\n-          case Some(offset) => PartitionFetchState(offset, state.value.delay, truncatingLog = false)\n+          case Some(offsetTruncationState) => PartitionFetchState(offsetTruncationState.offset, state.value.delay, truncatingLog = !offsetTruncationState.truncationCompleted)\n           case None => state.value()\n         }\n         (state.topicPartition(), maybeTruncationComplete)\n       }.toMap\n     partitionStates.set(newStates.asJava)\n   }\n \n+  /**\n+   * Called from ReplicaFetcherThread and ReplicaAlterLogDirsThread maybeTruncate for each topic\n+   * partition. Returns truncation offset and whether this is the final offset to truncate to\n+   * @param tp                    Topic partition\n+   * @param leaderEpochOffset     Epoch end offset received from the leader for this topic partition\n+   * @param replica               Follower's replica, which is either local replica\n+   *                              (ReplicaFetcherThread) or future replica (ReplicaAlterLogDirsThread)\n+   * @param isFutureReplica       true if called from ReplicaAlterLogDirsThread\n+   */\n+  def getOffsetTruncationState(tp: TopicPartition, leaderEpochOffset: EpochEndOffset, replica: Replica, isFutureReplica: Boolean = false): OffsetTruncationState = {\n+    // to make sure we can distinguish log output for fetching from remote leader or local replica\n+    val followerName = if (isFutureReplica) \"future replica\" else \"follower\"\n+\n+    // Called when 'offsetToTruncateTo' is the final offset to truncate to.\n+    def finalFetchLeaderEpochOffset(offsetToTruncateTo: Long, offsetFromLeader: Long): OffsetTruncationState = {\n+      val fetchOffset =\n+        if (offsetToTruncateTo >= replica.logEndOffset.messageOffset) {\n+          val logEndOffset = replica.logEndOffset.messageOffset\n+          info(\n+            s\"Based on $followerName's leader epoch, leader replied with an offset $offsetFromLeader. Min(leader offset, leader epoch end offset on the follower) $offsetToTruncateTo >= the \" +\n+            s\"$followerName's log end offset $logEndOffset in ${replica.topicPartition}. No truncation needed.\")\n+          logEndOffset\n+        } else\n+          offsetToTruncateTo\n+      OffsetTruncationState(fetchOffset, truncationCompleted = true)\n+    }\n+\n+    if (leaderEpochOffset.endOffset == UNDEFINED_EPOCH_OFFSET) {\n+      // truncate to initial offset which is the high watermark for follower replica or current\n+      // replica's truncation offset for the future replica (when the current replica truncates,\n+      // it forces future replica's partition state to 'truncating' and sets initial offset to\n+      // its truncation offset)\n+      warn(s\"Based on $followerName's leader epoch, leader replied with an unknown offset in ${replica.topicPartition}. \" +\n+           s\"The initial fetch offset ${partitionStates.stateValue(tp).fetchOffset} will be used for truncation.\")\n+      OffsetTruncationState(partitionStates.stateValue(tp).fetchOffset, truncationCompleted = true)\n+    } else if (leaderEpochOffset.leaderEpoch == UNDEFINED_EPOCH || !useLeaderEpochInResponse) {\n+      // either leader or follower or both use inter-broker protocol version < KAFKA_2_0_IV0\n+      // (version 0 of OffsetForLeaderEpoch request/response)\n+      warn(s\"Leader or $followerName is on protocol version where leader epoch is not considered in the OffsetsForLeaderEpoch response. \" +\n+           s\"The leader's offset ${leaderEpochOffset.endOffset} will be used for truncation in ${replica.topicPartition}.\")\n+      finalFetchLeaderEpochOffset(leaderEpochOffset.endOffset, leaderEpochOffset.endOffset)\n+    } else {\n+      // get (leader epoch, end offset) pair that corresponds to the largest leader epoch\n+      // less than or equal to the requested epoch.\n+      val (replicaLeaderEpoch, replicaEndOffset) = replica.epochs.get.endOffsetFor(leaderEpochOffset.leaderEpoch)","path":"core/src/main/scala/kafka/server/AbstractFetcherThread.scala","position":102,"original_position":102,"commit_id":"5dabeb718468c5ff9e167c2b5bad4a94b23a4780","original_commit_id":"5dabeb718468c5ff9e167c2b5bad4a94b23a4780","user":{"login":"junrao","id":132330,"avatar_url":"https://avatars2.githubusercontent.com/u/132330?v=4","gravatar_id":"","url":"https://api.github.com/users/junrao","html_url":"https://github.com/junrao","followers_url":"https://api.github.com/users/junrao/followers","following_url":"https://api.github.com/users/junrao/following{/other_user}","gists_url":"https://api.github.com/users/junrao/gists{/gist_id}","starred_url":"https://api.github.com/users/junrao/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/junrao/subscriptions","organizations_url":"https://api.github.com/users/junrao/orgs","repos_url":"https://api.github.com/users/junrao/repos","events_url":"https://api.github.com/users/junrao/events{/privacy}","received_events_url":"https://api.github.com/users/junrao/received_events","type":"User","site_admin":false},"body":"replicaLeaderEpoch and leaderEpochOffset may be confusing. How about followerEpoch and leaderEpochOffset?","created_at":"2018-05-02T23:10:06Z","updated_at":"2018-05-02T23:32:27Z","html_url":"https://github.com/apache/kafka/pull/4882#discussion_r185664468","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/4882","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/185664468"},"html":{"href":"https://github.com/apache/kafka/pull/4882#discussion_r185664468"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/4882"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/4882","id":181973750,"html_url":"https://github.com/apache/kafka/pull/4882","diff_url":"https://github.com/apache/kafka/pull/4882.diff","patch_url":"https://github.com/apache/kafka/pull/4882.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/4882","number":4882,"state":"open","locked":false,"title":" KAFKA-6361: Fix log divergence between leader and follower after fast leader fail over","user":{"login":"apovzner","id":5383840,"avatar_url":"https://avatars0.githubusercontent.com/u/5383840?v=4","gravatar_id":"","url":"https://api.github.com/users/apovzner","html_url":"https://github.com/apovzner","followers_url":"https://api.github.com/users/apovzner/followers","following_url":"https://api.github.com/users/apovzner/following{/other_user}","gists_url":"https://api.github.com/users/apovzner/gists{/gist_id}","starred_url":"https://api.github.com/users/apovzner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apovzner/subscriptions","organizations_url":"https://api.github.com/users/apovzner/orgs","repos_url":"https://api.github.com/users/apovzner/repos","events_url":"https://api.github.com/users/apovzner/events{/privacy}","received_events_url":"https://api.github.com/users/apovzner/received_events","type":"User","site_admin":false},"body":"Implementation of KIP-279 as described here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-279%3A+Fix+log+divergence+between+leader+and+follower+after+fast+leader+fail+over\r\n\r\nIn summary:\r\n- Added leader_epoch to OFFSET_FOR_LEADER_EPOCH_RESPONSE\r\n- Leader replies with the pair( largest epoch less than or equal to the requested epoch, the end offset of this epoch)\r\n- If Follower does not know about the leader epoch that leader replies with, it truncates to the end offset of largest leader epoch less than leader epoch that leader replied with, and sends another OffsetForLeaderEpoch request. That request contains the largest leader epoch less than leader epoch that leader replied with.\r\n\r\nAdded integration test EpochDrivenReplicationProtocolAcceptanceTest.logsShouldNotDivergeOnUncleanLeaderElections that does 3 fast leader changes where unclean leader election is enabled and min isr is 1. The test failed before the fix was implemented.\r\n\r\n### Committer Checklist (excluded from commit message)\r\n- [ ] Verify design and implementation \r\n- [ ] Verify test coverage and CI build status\r\n- [ ] Verify documentation (including upgrade notes)\r\n","created_at":"2018-04-16T20:06:59Z","updated_at":"2018-05-02T23:32:27Z","closed_at":null,"merged_at":null,"merge_commit_sha":"185c92ea232c87793d89e7d642d45c54c8495caf","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/4882/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/4882/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/4882/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/5dabeb718468c5ff9e167c2b5bad4a94b23a4780","head":{"label":"apovzner:kafka-6361","ref":"kafka-6361","sha":"5dabeb718468c5ff9e167c2b5bad4a94b23a4780","user":{"login":"apovzner","id":5383840,"avatar_url":"https://avatars0.githubusercontent.com/u/5383840?v=4","gravatar_id":"","url":"https://api.github.com/users/apovzner","html_url":"https://github.com/apovzner","followers_url":"https://api.github.com/users/apovzner/followers","following_url":"https://api.github.com/users/apovzner/following{/other_user}","gists_url":"https://api.github.com/users/apovzner/gists{/gist_id}","starred_url":"https://api.github.com/users/apovzner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apovzner/subscriptions","organizations_url":"https://api.github.com/users/apovzner/orgs","repos_url":"https://api.github.com/users/apovzner/repos","events_url":"https://api.github.com/users/apovzner/events{/privacy}","received_events_url":"https://api.github.com/users/apovzner/received_events","type":"User","site_admin":false},"repo":{"id":44782457,"name":"kafka","full_name":"apovzner/kafka","owner":{"login":"apovzner","id":5383840,"avatar_url":"https://avatars0.githubusercontent.com/u/5383840?v=4","gravatar_id":"","url":"https://api.github.com/users/apovzner","html_url":"https://github.com/apovzner","followers_url":"https://api.github.com/users/apovzner/followers","following_url":"https://api.github.com/users/apovzner/following{/other_user}","gists_url":"https://api.github.com/users/apovzner/gists{/gist_id}","starred_url":"https://api.github.com/users/apovzner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apovzner/subscriptions","organizations_url":"https://api.github.com/users/apovzner/orgs","repos_url":"https://api.github.com/users/apovzner/repos","events_url":"https://api.github.com/users/apovzner/events{/privacy}","received_events_url":"https://api.github.com/users/apovzner/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/apovzner/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/apovzner/kafka","forks_url":"https://api.github.com/repos/apovzner/kafka/forks","keys_url":"https://api.github.com/repos/apovzner/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apovzner/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apovzner/kafka/teams","hooks_url":"https://api.github.com/repos/apovzner/kafka/hooks","issue_events_url":"https://api.github.com/repos/apovzner/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apovzner/kafka/events","assignees_url":"https://api.github.com/repos/apovzner/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apovzner/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apovzner/kafka/tags","blobs_url":"https://api.github.com/repos/apovzner/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apovzner/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apovzner/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apovzner/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apovzner/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apovzner/kafka/languages","stargazers_url":"https://api.github.com/repos/apovzner/kafka/stargazers","contributors_url":"https://api.github.com/repos/apovzner/kafka/contributors","subscribers_url":"https://api.github.com/repos/apovzner/kafka/subscribers","subscription_url":"https://api.github.com/repos/apovzner/kafka/subscription","commits_url":"https://api.github.com/repos/apovzner/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apovzner/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apovzner/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apovzner/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apovzner/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apovzner/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apovzner/kafka/merges","archive_url":"https://api.github.com/repos/apovzner/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apovzner/kafka/downloads","issues_url":"https://api.github.com/repos/apovzner/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apovzner/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apovzner/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apovzner/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apovzner/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apovzner/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apovzner/kafka/deployments","created_at":"2015-10-23T01:03:29Z","updated_at":"2018-03-02T22:24:28Z","pushed_at":"2018-05-01T23:59:46Z","git_url":"git://github.com/apovzner/kafka.git","ssh_url":"git@github.com:apovzner/kafka.git","clone_url":"https://github.com/apovzner/kafka.git","svn_url":"https://github.com/apovzner/kafka","homepage":null,"size":61553,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"f467c9c2438a8b182083879927ff171a6a2c6f2f","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2018-05-02T20:33:58Z","pushed_at":"2018-05-02T21:41:53Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":64742,"stargazers_count":8150,"watchers_count":8150,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":4798,"mirror_url":null,"archived":false,"open_issues_count":459,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":4798,"open_issues":459,"watchers":8150,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/4882"},"html":{"href":"https://github.com/apache/kafka/pull/4882"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/4882"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/4882/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/4882/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/4882/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/5dabeb718468c5ff9e167c2b5bad4a94b23a4780"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"junrao"}}