{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/116158769","pull_request_review_id":37759830,"id":116158769,"diff_hunk":"@@ -451,59 +490,106 @@ class GroupMetadataManager(brokerId: Int,\n           }\n \n           memRecords.batches.asScala.foreach { batch =>\n-            for (record <- batch.asScala) {\n-              require(record.hasKey, \"Group metadata/offset entry key should not be null\")\n-              GroupMetadataManager.readMessageKey(record.key) match {\n-\n-                case offsetKey: OffsetKey =>\n-                  // load offset\n-                  val key = offsetKey.key\n-                  if (!record.hasValue) {\n-                    loadedOffsets.remove(key)\n-                    removedOffsets.add(key)\n-                  } else {\n-                    val value = GroupMetadataManager.readOffsetMessageValue(record.value)\n-                    loadedOffsets.put(key, value)\n-                    removedOffsets.remove(key)\n-                  }\n-\n-                case groupMetadataKey: GroupMetadataKey =>\n-                  // load group metadata\n-                  val groupId = groupMetadataKey.key\n-                  val groupMetadata = GroupMetadataManager.readGroupMessageValue(groupId, record.value)\n-                  if (groupMetadata != null) {\n-                    trace(s\"Loaded group metadata for group $groupId with generation ${groupMetadata.generationId}\")\n-                    removedGroups.remove(groupId)\n-                    loadedGroups.put(groupId, groupMetadata)\n-                  } else {\n-                    loadedGroups.remove(groupId)\n-                    removedGroups.add(groupId)\n+            val isTxnOffsetCommit = batch.isTransactional\n+            if (batch.isControlBatch) {\n+              val record = batch.iterator.next()\n+              val controlRecord = ControlRecordType.parse(record.key)\n+              if (controlRecord == ControlRecordType.COMMIT) {\n+                val pendingOffsetsFromProducer = pendingOffsets.getOrElse(batch.producerId, null)\n+                if (pendingOffsetsFromProducer != null) {\n+                  pendingOffsetsFromProducer.foreach { case (groupTopicPartition, offsetAndMetadata) =>\n+                    loadedOffsets.put(groupTopicPartition, offsetAndMetadata)\n                   }\n-\n-                case unknownKey =>\n-                  throw new IllegalStateException(s\"Unexpected message key $unknownKey while loading offsets and group metadata\")\n+                 pendingOffsets.remove(batch.producerId)\n+                }\n+              } else {\n+                pendingOffsets.remove(batch.producerId)\n+              }\n+            } else {\n+              for (record <- batch.asScala) {\n+                require(record.hasKey, \"Group metadata/offset entry key should not be null\")\n+                GroupMetadataManager.readMessageKey(record.key) match {\n+\n+                  case offsetKey: OffsetKey =>\n+                    if (isTxnOffsetCommit && !pendingOffsets.contains(batch.producerId))\n+                      pendingOffsets.put(batch.producerId, mutable.Map[GroupTopicPartition, OffsetAndMetadata]())\n+\n+                    // load offset\n+                    val key = offsetKey.key\n+                    if (!record.hasValue) {\n+                      if (isTxnOffsetCommit)\n+                        pendingOffsets(batch.producerId).remove(key)","path":"core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala","position":196,"original_position":187,"commit_id":"b2cbd2cb5288cc4a7f9e7c71c68812ea52a8a741","original_commit_id":"8c974ec9b439b5820f723bcd2473cadc48a4b5b7","user":{"login":"apurvam","id":830220,"avatar_url":"https://avatars3.githubusercontent.com/u/830220?v=3","gravatar_id":"","url":"https://api.github.com/users/apurvam","html_url":"https://github.com/apurvam","followers_url":"https://api.github.com/users/apurvam/followers","following_url":"https://api.github.com/users/apurvam/following{/other_user}","gists_url":"https://api.github.com/users/apurvam/gists{/gist_id}","starred_url":"https://api.github.com/users/apurvam/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apurvam/subscriptions","organizations_url":"https://api.github.com/users/apurvam/orgs","repos_url":"https://api.github.com/users/apurvam/repos","events_url":"https://api.github.com/users/apurvam/events{/privacy}","received_events_url":"https://api.github.com/users/apurvam/received_events","type":"User","site_admin":false},"body":"Actually, I got rid of `producerGroups` all together. I know derive the producer groups from `pendingOffsets` after we load all the data. We are iterating over the `pendingOffsets` map anyway, so I just extract the active groups at that point and call `addProducerGroup` on each.","created_at":"2017-05-12T06:17:02Z","updated_at":"2017-05-12T06:17:02Z","html_url":"https://github.com/apache/kafka/pull/2970#discussion_r116158769","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/2970","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/116158769"},"html":{"href":"https://github.com/apache/kafka/pull/2970#discussion_r116158769"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/2970"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/2970","id":118885866,"html_url":"https://github.com/apache/kafka/pull/2970","diff_url":"https://github.com/apache/kafka/pull/2970.diff","patch_url":"https://github.com/apache/kafka/pull/2970.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/2970","number":2970,"state":"open","locked":false,"title":"Kafka-5160; KIP-98 Broker side support for TxnOffsetCommitRequest","user":{"login":"apurvam","id":830220,"avatar_url":"https://avatars3.githubusercontent.com/u/830220?v=3","gravatar_id":"","url":"https://api.github.com/users/apurvam","html_url":"https://github.com/apurvam","followers_url":"https://api.github.com/users/apurvam/followers","following_url":"https://api.github.com/users/apurvam/following{/other_user}","gists_url":"https://api.github.com/users/apurvam/gists{/gist_id}","starred_url":"https://api.github.com/users/apurvam/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apurvam/subscriptions","organizations_url":"https://api.github.com/users/apurvam/orgs","repos_url":"https://api.github.com/users/apurvam/repos","events_url":"https://api.github.com/users/apurvam/events{/privacy}","received_events_url":"https://api.github.com/users/apurvam/received_events","type":"User","site_admin":false},"body":"This patch adds support for the `TxnOffsetCommitRequest` added in KIP-98. Desired handling for this request is [described here](https://docs.google.com/document/d/11Jqy_GjUGtdXJK94XGsEIK7CP1SnQGdp2eF0wSw9ra8/edit#bookmark=id.55yzhvkppi6m) . \r\n\r\nThe functionality includes handling the stable state of receiving `TxnOffsetCommitRequests` and materializing results only when the commit marker for the transaction is received. It also handles partition emigration and immigration and rebuilds the required data structures on these events.\r\n\r\nTests are included for all the functionality.","created_at":"2017-05-04T01:21:34Z","updated_at":"2017-05-12T06:17:02Z","closed_at":null,"merged_at":null,"merge_commit_sha":"1d0890dd87e89ab38361a439ac999cf555067045","assignee":null,"assignees":[],"requested_reviewers":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/2970/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/2970/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/2970/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/b2cbd2cb5288cc4a7f9e7c71c68812ea52a8a741","head":{"label":"apurvam:KAFKA-5160-broker-side-support-for-txnoffsetcommitrequest","ref":"KAFKA-5160-broker-side-support-for-txnoffsetcommitrequest","sha":"b2cbd2cb5288cc4a7f9e7c71c68812ea52a8a741","user":{"login":"apurvam","id":830220,"avatar_url":"https://avatars3.githubusercontent.com/u/830220?v=3","gravatar_id":"","url":"https://api.github.com/users/apurvam","html_url":"https://github.com/apurvam","followers_url":"https://api.github.com/users/apurvam/followers","following_url":"https://api.github.com/users/apurvam/following{/other_user}","gists_url":"https://api.github.com/users/apurvam/gists{/gist_id}","starred_url":"https://api.github.com/users/apurvam/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apurvam/subscriptions","organizations_url":"https://api.github.com/users/apurvam/orgs","repos_url":"https://api.github.com/users/apurvam/repos","events_url":"https://api.github.com/users/apurvam/events{/privacy}","received_events_url":"https://api.github.com/users/apurvam/received_events","type":"User","site_admin":false},"repo":{"id":69067896,"name":"kafka","full_name":"apurvam/kafka","owner":{"login":"apurvam","id":830220,"avatar_url":"https://avatars3.githubusercontent.com/u/830220?v=3","gravatar_id":"","url":"https://api.github.com/users/apurvam","html_url":"https://github.com/apurvam","followers_url":"https://api.github.com/users/apurvam/followers","following_url":"https://api.github.com/users/apurvam/following{/other_user}","gists_url":"https://api.github.com/users/apurvam/gists{/gist_id}","starred_url":"https://api.github.com/users/apurvam/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apurvam/subscriptions","organizations_url":"https://api.github.com/users/apurvam/orgs","repos_url":"https://api.github.com/users/apurvam/repos","events_url":"https://api.github.com/users/apurvam/events{/privacy}","received_events_url":"https://api.github.com/users/apurvam/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/apurvam/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/apurvam/kafka","forks_url":"https://api.github.com/repos/apurvam/kafka/forks","keys_url":"https://api.github.com/repos/apurvam/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apurvam/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apurvam/kafka/teams","hooks_url":"https://api.github.com/repos/apurvam/kafka/hooks","issue_events_url":"https://api.github.com/repos/apurvam/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apurvam/kafka/events","assignees_url":"https://api.github.com/repos/apurvam/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apurvam/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apurvam/kafka/tags","blobs_url":"https://api.github.com/repos/apurvam/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apurvam/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apurvam/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apurvam/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apurvam/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apurvam/kafka/languages","stargazers_url":"https://api.github.com/repos/apurvam/kafka/stargazers","contributors_url":"https://api.github.com/repos/apurvam/kafka/contributors","subscribers_url":"https://api.github.com/repos/apurvam/kafka/subscribers","subscription_url":"https://api.github.com/repos/apurvam/kafka/subscription","commits_url":"https://api.github.com/repos/apurvam/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apurvam/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apurvam/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apurvam/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apurvam/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apurvam/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apurvam/kafka/merges","archive_url":"https://api.github.com/repos/apurvam/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apurvam/kafka/downloads","issues_url":"https://api.github.com/repos/apurvam/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apurvam/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apurvam/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apurvam/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apurvam/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apurvam/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apurvam/kafka/deployments","created_at":"2016-09-23T23:38:16Z","updated_at":"2016-09-23T23:38:22Z","pushed_at":"2017-05-12T05:19:46Z","git_url":"git://github.com/apurvam/kafka.git","ssh_url":"git@github.com:apurvam/kafka.git","clone_url":"https://github.com/apurvam/kafka.git","svn_url":"https://github.com/apurvam/kafka","homepage":null,"size":48248,"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,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"a420d20c0e3da2c9637820f7d1a344706fd835fa","user":{"login":"apache","id":47359,"avatar_url":"https://avatars3.githubusercontent.com/u/47359?v=3","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://avatars3.githubusercontent.com/u/47359?v=3","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":"2017-05-12T01:02:53Z","pushed_at":"2017-05-12T05:39:03Z","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":47506,"stargazers_count":5158,"watchers_count":5158,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":3206,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":363,"forks":3206,"open_issues":363,"watchers":5158,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/2970"},"html":{"href":"https://github.com/apache/kafka/pull/2970"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/2970"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/2970/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/2970/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/2970/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/b2cbd2cb5288cc4a7f9e7c71c68812ea52a8a741"}}}}
{"actor":{"display_login":"apurvam"}}