{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/42654165","id":42654165,"diff_hunk":"@@ -0,0 +1,595 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE\n+ * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file\n+ * to You under the Apache License, Version 2.0 (the \"License\"); you may not use this file except in compliance with the\n+ * License. You may obtain a copy of the License at\n+ *\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on\n+ * an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the\n+ * specific language governing permissions and limitations under the License.\n+ */\n+package org.apache.kafka.clients.consumer.internals;\n+\n+import org.apache.kafka.clients.ClientResponse;\n+import org.apache.kafka.clients.Metadata;\n+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;\n+import org.apache.kafka.clients.consumer.ConsumerWakeupException;\n+import org.apache.kafka.clients.consumer.OffsetAndMetadata;\n+import org.apache.kafka.clients.consumer.OffsetCommitCallback;\n+import org.apache.kafka.clients.consumer.internals.PartitionAssignor.Assignment;\n+import org.apache.kafka.clients.consumer.internals.PartitionAssignor.Subscription;\n+import org.apache.kafka.common.Cluster;\n+import org.apache.kafka.common.KafkaException;\n+import org.apache.kafka.common.MetricName;\n+import org.apache.kafka.common.TopicPartition;\n+import org.apache.kafka.common.metrics.Measurable;\n+import org.apache.kafka.common.metrics.MetricConfig;\n+import org.apache.kafka.common.metrics.Metrics;\n+import org.apache.kafka.common.metrics.Sensor;\n+import org.apache.kafka.common.metrics.stats.Avg;\n+import org.apache.kafka.common.metrics.stats.Count;\n+import org.apache.kafka.common.metrics.stats.Max;\n+import org.apache.kafka.common.metrics.stats.Rate;\n+import org.apache.kafka.common.protocol.ApiKeys;\n+import org.apache.kafka.common.protocol.Errors;\n+import org.apache.kafka.common.requests.OffsetCommitRequest;\n+import org.apache.kafka.common.requests.OffsetCommitResponse;\n+import org.apache.kafka.common.requests.OffsetFetchRequest;\n+import org.apache.kafka.common.requests.OffsetFetchResponse;\n+import org.apache.kafka.common.utils.Time;\n+import org.apache.kafka.common.utils.Utils;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.io.Closeable;\n+import java.nio.ByteBuffer;\n+import java.util.ArrayList;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.LinkedHashMap;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Set;\n+\n+/**\n+ * This class manages the coordination process with the consumer coordinator.\n+ */\n+public final class ConsumerCoordinator extends AbstractCoordinator implements Closeable {\n+\n+    private static final Logger log = LoggerFactory.getLogger(ConsumerCoordinator.class);\n+\n+    private final Map<String, PartitionAssignor> protocolMap;\n+    private final org.apache.kafka.clients.Metadata metadata;\n+    private final MetadataSnapshot metadataSnapshot;\n+    private final ConsumerCoordinatorMetrics sensors;\n+    private final SubscriptionState subscriptions;\n+    private final OffsetCommitCallback defaultOffsetCommitCallback;\n+    private final boolean autoCommitEnabled;\n+\n+    /**\n+     * Initialize the coordination manager.\n+     */\n+    public ConsumerCoordinator(ConsumerNetworkClient client,\n+                               String groupId,\n+                               int sessionTimeoutMs,\n+                               int heartbeatIntervalMs,\n+                               List<PartitionAssignor> assignors,\n+                               Metadata metadata,\n+                               SubscriptionState subscriptions,\n+                               Metrics metrics,\n+                               String metricGrpPrefix,\n+                               Map<String, String> metricTags,\n+                               Time time,\n+                               long requestTimeoutMs,\n+                               long retryBackoffMs,\n+                               OffsetCommitCallback defaultOffsetCommitCallback,\n+                               boolean autoCommitEnabled,\n+                               long autoCommitIntervalMs) {\n+        super(client,\n+                groupId,\n+                sessionTimeoutMs,\n+                heartbeatIntervalMs,\n+                metrics,\n+                metricGrpPrefix,\n+                metricTags,\n+                time,\n+                requestTimeoutMs,\n+                retryBackoffMs);\n+        this.metadata = metadata;\n+\n+        this.metadata.requestUpdate();\n+        this.metadataSnapshot = new MetadataSnapshot();\n+        this.subscriptions = subscriptions;\n+        this.defaultOffsetCommitCallback = defaultOffsetCommitCallback;\n+        this.autoCommitEnabled = autoCommitEnabled;\n+\n+        this.protocolMap = new HashMap<>();\n+        for (PartitionAssignor assignor : assignors)\n+            this.protocolMap.put(assignor.name(), assignor);\n+\n+        addMetadataListener();\n+\n+        if (autoCommitEnabled)\n+            scheduleAutoCommitTask(autoCommitIntervalMs);\n+\n+        this.sensors = new ConsumerCoordinatorMetrics(metrics, metricGrpPrefix, metricTags);\n+    }\n+\n+    @Override\n+    public String protocolType() {\n+        return \"consumer\";\n+    }\n+\n+    @Override\n+    public LinkedHashMap<String, ByteBuffer> metadata() {\n+        LinkedHashMap<String, ByteBuffer> metadata = new LinkedHashMap<>();\n+        for (PartitionAssignor assignor : protocolMap.values()) {\n+            Subscription subscription = assignor.subscription(subscriptions.subscription());\n+            metadata.put(assignor.name(), ConsumerProtocol.serializeSubscription(subscription));\n+        }\n+        return metadata;\n+    }\n+\n+    private void addMetadataListener() {\n+        this.metadata.addListener(new Metadata.Listener() {\n+            @Override\n+            public void onMetadataUpdate(Cluster cluster) {\n+                if (subscriptions.hasPatternSubscription()) {\n+                    final List<String> topicsToSubscribe = new ArrayList<>();\n+\n+                    for (String topic : cluster.topics())\n+                        if (subscriptions.getSubscribedPattern().matcher(topic).matches())\n+                            topicsToSubscribe.add(topic);\n+\n+                    subscriptions.changeSubscription(topicsToSubscribe);\n+                    metadata.setTopics(subscriptions.groupSubscription());\n+                }\n+\n+                // check if there are any changes to the metadata which should trigger a rebalance\n+                if (metadataSnapshot.update(subscriptions, cluster) && subscriptions.partitionsAutoAssigned())\n+                    subscriptions.needReassignment();\n+            }\n+        });\n+    }\n+\n+    @Override\n+    protected void onJoin(int generation,\n+                          String memberId,\n+                          String assignmentStrategy,\n+                          ByteBuffer assignmentBuffer) {\n+        PartitionAssignor assignor = protocolMap.get(assignmentStrategy);\n+        if (assignor == null)\n+            throw new IllegalStateException(\"Coordinator selected invalid assignment protocol: \" + assignmentStrategy);\n+\n+        Assignment assignment = ConsumerProtocol.deserializeAssignment(assignmentBuffer);\n+\n+        // set the flag to refresh last committed offsets\n+        subscriptions.needRefreshCommits();\n+\n+        // update partition assignment\n+        subscriptions.changePartitionAssignment(assignment.partitions());\n+\n+        // give the assignor a chance to update internal state based on the received assignment\n+        assignor.onAssignment(assignment);","path":"clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java","position":175,"original_position":175,"commit_id":"fd6c593b6a7144c0d693f17bab9b39d5fe265ec7","original_commit_id":"acfc9a88c9fdec5453edd95f9bfa4ea119a28f58","user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars.githubusercontent.com/u/10729597?v=3","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"body":"```onAssignment``` is only triggered after the syncing phase by everyone, not only by the leader. It is different from ```ConsumerRebalanceListener.onPartitionsAssigned()``` such that the latter only gives the list of partitions, while the former contains extra userData that can be associated with the assigned partitions. For Kafka Streams it is important to use this userData to infer the further mapping from consumer's partitions to task's partitions.","created_at":"2015-10-21T17:24:16Z","updated_at":"2015-10-21T17:24:16Z","html_url":"https://github.com/apache/kafka/pull/165#discussion_r42654165","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/165","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/42654165"},"html":{"href":"https://github.com/apache/kafka/pull/165#discussion_r42654165"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/165"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/165","id":43248947,"html_url":"https://github.com/apache/kafka/pull/165","diff_url":"https://github.com/apache/kafka/pull/165.diff","patch_url":"https://github.com/apache/kafka/pull/165.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/165","number":165,"state":"open","locked":false,"title":"KAFKA-2464: client-side assignment for new consumer","user":{"login":"hachikuji","id":12502538,"avatar_url":"https://avatars.githubusercontent.com/u/12502538?v=3","gravatar_id":"","url":"https://api.github.com/users/hachikuji","html_url":"https://github.com/hachikuji","followers_url":"https://api.github.com/users/hachikuji/followers","following_url":"https://api.github.com/users/hachikuji/following{/other_user}","gists_url":"https://api.github.com/users/hachikuji/gists{/gist_id}","starred_url":"https://api.github.com/users/hachikuji/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hachikuji/subscriptions","organizations_url":"https://api.github.com/users/hachikuji/orgs","repos_url":"https://api.github.com/users/hachikuji/repos","events_url":"https://api.github.com/users/hachikuji/events{/privacy}","received_events_url":"https://api.github.com/users/hachikuji/received_events","type":"User","site_admin":false},"body":"","created_at":"2015-08-25T04:36:29Z","updated_at":"2015-10-21T17:24:16Z","closed_at":null,"merged_at":null,"merge_commit_sha":"da50d6674f2f066f793706ebc10f0c090f44e8c0","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/165/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/165/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/165/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/fd6c593b6a7144c0d693f17bab9b39d5fe265ec7","head":{"label":"hachikuji:KAFKA-2464","ref":"KAFKA-2464","sha":"fd6c593b6a7144c0d693f17bab9b39d5fe265ec7","user":{"login":"hachikuji","id":12502538,"avatar_url":"https://avatars.githubusercontent.com/u/12502538?v=3","gravatar_id":"","url":"https://api.github.com/users/hachikuji","html_url":"https://github.com/hachikuji","followers_url":"https://api.github.com/users/hachikuji/followers","following_url":"https://api.github.com/users/hachikuji/following{/other_user}","gists_url":"https://api.github.com/users/hachikuji/gists{/gist_id}","starred_url":"https://api.github.com/users/hachikuji/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hachikuji/subscriptions","organizations_url":"https://api.github.com/users/hachikuji/orgs","repos_url":"https://api.github.com/users/hachikuji/repos","events_url":"https://api.github.com/users/hachikuji/events{/privacy}","received_events_url":"https://api.github.com/users/hachikuji/received_events","type":"User","site_admin":false},"repo":{"id":35842084,"name":"kafka","full_name":"hachikuji/kafka","owner":{"login":"hachikuji","id":12502538,"avatar_url":"https://avatars.githubusercontent.com/u/12502538?v=3","gravatar_id":"","url":"https://api.github.com/users/hachikuji","html_url":"https://github.com/hachikuji","followers_url":"https://api.github.com/users/hachikuji/followers","following_url":"https://api.github.com/users/hachikuji/following{/other_user}","gists_url":"https://api.github.com/users/hachikuji/gists{/gist_id}","starred_url":"https://api.github.com/users/hachikuji/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hachikuji/subscriptions","organizations_url":"https://api.github.com/users/hachikuji/orgs","repos_url":"https://api.github.com/users/hachikuji/repos","events_url":"https://api.github.com/users/hachikuji/events{/privacy}","received_events_url":"https://api.github.com/users/hachikuji/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/hachikuji/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/hachikuji/kafka","forks_url":"https://api.github.com/repos/hachikuji/kafka/forks","keys_url":"https://api.github.com/repos/hachikuji/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/hachikuji/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/hachikuji/kafka/teams","hooks_url":"https://api.github.com/repos/hachikuji/kafka/hooks","issue_events_url":"https://api.github.com/repos/hachikuji/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/hachikuji/kafka/events","assignees_url":"https://api.github.com/repos/hachikuji/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/hachikuji/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/hachikuji/kafka/tags","blobs_url":"https://api.github.com/repos/hachikuji/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/hachikuji/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/hachikuji/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/hachikuji/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/hachikuji/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/hachikuji/kafka/languages","stargazers_url":"https://api.github.com/repos/hachikuji/kafka/stargazers","contributors_url":"https://api.github.com/repos/hachikuji/kafka/contributors","subscribers_url":"https://api.github.com/repos/hachikuji/kafka/subscribers","subscription_url":"https://api.github.com/repos/hachikuji/kafka/subscription","commits_url":"https://api.github.com/repos/hachikuji/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/hachikuji/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/hachikuji/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/hachikuji/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/hachikuji/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/hachikuji/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/hachikuji/kafka/merges","archive_url":"https://api.github.com/repos/hachikuji/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/hachikuji/kafka/downloads","issues_url":"https://api.github.com/repos/hachikuji/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/hachikuji/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/hachikuji/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/hachikuji/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/hachikuji/kafka/labels{/name}","releases_url":"https://api.github.com/repos/hachikuji/kafka/releases{/id}","created_at":"2015-05-18T20:54:26Z","updated_at":"2015-05-18T20:54:28Z","pushed_at":"2015-10-21T03:37:19Z","git_url":"git://github.com/hachikuji/kafka.git","ssh_url":"git@github.com:hachikuji/kafka.git","clone_url":"https://github.com/hachikuji/kafka.git","svn_url":"https://github.com/hachikuji/kafka","homepage":null,"size":23917,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"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":"0.8.2"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"44f6c4b946511ce4663d41bf40f2960d2faee198","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.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://avatars.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}","created_at":"2011-08-15T18:06:16Z","updated_at":"2015-10-21T17:02:43Z","pushed_at":"2015-10-21T17:00:10Z","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":38677,"stargazers_count":2086,"watchers_count":2086,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1234,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":100,"forks":1234,"open_issues":100,"watchers":2086,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/165"},"html":{"href":"https://github.com/apache/kafka/pull/165"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/165"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/165/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/165/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/165/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/fd6c593b6a7144c0d693f17bab9b39d5fe265ec7"}}}}