{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/102655110","pull_request_review_id":23414806,"id":102655110,"diff_hunk":"@@ -0,0 +1,184 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ * <p>\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ * <p>\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.apache.kafka.streams.processor.internals;\n+\n+import org.apache.kafka.clients.consumer.Consumer;\n+import org.apache.kafka.clients.consumer.ConsumerRecord;\n+import org.apache.kafka.clients.consumer.ConsumerRecords;\n+import org.apache.kafka.common.PartitionInfo;\n+import org.apache.kafka.common.TopicPartition;\n+import org.apache.kafka.common.errors.TimeoutException;\n+import org.apache.kafka.common.utils.Time;\n+import org.apache.kafka.streams.errors.StreamsException;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.util.Collections;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.Iterator;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Set;\n+\n+public class StoreChangelogReader implements ChangelogReader {\n+    private static final Logger log = LoggerFactory.getLogger(StoreChangelogReader.class);\n+\n+    private final Consumer<byte[], byte[]> consumer;\n+    private final Time time;\n+    private final long partitionValidationTimeoutMs;\n+    private final Map<String, List<PartitionInfo>> partitionInfo = new HashMap<>();\n+    private final Map<TopicPartition, StateRestorer> stateRestorers = new HashMap<>();\n+\n+\n+    public StoreChangelogReader(final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs) {\n+        this.consumer = consumer;\n+        this.time = time;\n+        this.partitionValidationTimeoutMs = partitionValidationTimeoutMs;\n+    }\n+\n+\n+    @Override\n+    public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) {\n+        final long start = time.milliseconds();\n+        // fetch all info on all topics to avoid multiple remote calls\n+        if (partitionInfo.isEmpty()) {\n+            try {\n+                partitionInfo.putAll(consumer.listTopics());\n+            } catch (final TimeoutException e) {\n+                log.warn(\"Could not list topics so will fall back to partition by partition fetching\");\n+            }\n+        }\n+\n+        final long endTime = time.milliseconds() + partitionValidationTimeoutMs;\n+        while (!hasPartition(topicPartition) && time.milliseconds() < endTime) {\n+            try {\n+                final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic());\n+                if (partitions != null) {\n+                    partitionInfo.put(topicPartition.topic(), partitions);\n+                }\n+            } catch (final TimeoutException e) {\n+                throw new StreamsException(String.format(\"Could not fetch partition info for topic: %s before expiration of the configured request timeout\",\n+                                                         topicPartition.topic()));\n+            }\n+        }\n+\n+        if (!hasPartition(topicPartition)) {\n+            throw new StreamsException(String.format(\"Store %s's change log (%s) does not contain partition %s\",\n+                                                     storeName, topicPartition.topic(), topicPartition.partition()));\n+        }\n+        log.debug(\"Took {} ms to validate that partition {} exists\", time.milliseconds() - start, topicPartition);\n+    }\n+\n+    @Override\n+    public void register(final StateRestorer restorer) {\n+        stateRestorers.put(restorer.partition(), restorer);\n+    }\n+\n+    public void restore() {\n+        final long start = time.milliseconds();\n+        try {\n+            if (!consumer.subscription().isEmpty()) {\n+                throw new IllegalStateException(String.format(\"Restore consumer should have not subscribed to any partitions (%s) beforehand\", consumer.subscription()));\n+            }\n+            final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet());\n+            consumer.assign(stateRestorers.keySet());\n+\n+            for (final StateRestorer restorer : stateRestorers.values()) {\n+                if (restorer.checkpoint() != null) {\n+                    consumer.seek(restorer.partition(), restorer.checkpoint());\n+                } else {\n+                    consumer.seekToBeginning(Collections.singletonList(restorer.partition()));\n+                }\n+            }\n+\n+            final Set<TopicPartition> partitions = new HashSet<>(stateRestorers.keySet());\n+            while (!partitions.isEmpty()) {\n+                final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10);\n+                final Iterator<TopicPartition> partitionIterator = partitions.iterator();\n+                while (partitionIterator.hasNext()) {\n+                    restorePartition(endOffsets, allRecords, partitionIterator);\n+                }\n+            }\n+        } finally {\n+            consumer.assign(Collections.<TopicPartition>emptyList());\n+            log.debug(\"Took {} ms to restore active state\", time.milliseconds() - start);\n+        }\n+    }\n+\n+    @Override\n+    public Map<TopicPartition, Long> restoredOffsets() {\n+        final Map<TopicPartition, Long> restoredOffsets = new HashMap<>();\n+        for (final Map.Entry<TopicPartition, StateRestorer> entry : stateRestorers.entrySet()) {\n+            final StateRestorer restorer = entry.getValue();\n+            if (restorer.isPersistent()) {\n+                restoredOffsets.put(entry.getKey(), restorer.restoredOffset());\n+            }\n+        }\n+        return restoredOffsets;\n+    }\n+\n+    private void restorePartition(final Map<TopicPartition, Long> endOffsets,\n+                                  final ConsumerRecords<byte[], byte[]> allRecords,\n+                                  final Iterator<TopicPartition> partitionIterator) {\n+        final TopicPartition topicPartition = partitionIterator.next();\n+        final StateRestorer restorer = stateRestorers.get(topicPartition);\n+        final Long endOffset = endOffsets.get(topicPartition);\n+        final long pos = processNext(allRecords.records(topicPartition), restorer, endOffset);\n+        if (restorer.hasCompleted(pos + 1, endOffset)) {","path":"streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java","position":142,"original_position":142,"commit_id":"8bcf395014553448d2a3846a15f79d2f1d20de8a","original_commit_id":"8bcf395014553448d2a3846a15f79d2f1d20de8a","user":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"body":"it is because in some cases there is no data to fetch, i.e., there is no data to restore, so the endOffset and the current position are the same. In these cases this would never terminate.","created_at":"2017-02-23T07:46:20Z","updated_at":"2017-02-23T07:46:20Z","html_url":"https://github.com/apache/kafka/pull/2560#discussion_r102655110","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/2560","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/102655110"},"html":{"href":"https://github.com/apache/kafka/pull/2560#discussion_r102655110"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/2560"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/2560","id":106641885,"html_url":"https://github.com/apache/kafka/pull/2560","diff_url":"https://github.com/apache/kafka/pull/2560.diff","patch_url":"https://github.com/apache/kafka/pull/2560.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/2560","number":2560,"state":"open","locked":false,"title":"KAFKA-4494: Reduce startup and rebalance time","user":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"body":"Replace one-by-one initialization of state stores with bulk initialization.","created_at":"2017-02-17T00:04:15Z","updated_at":"2017-02-23T07:46:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"55094297181b3a467bb85cb3f652dd76d42ab0d5","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/2560/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/2560/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/2560/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/8bcf395014553448d2a3846a15f79d2f1d20de8a","head":{"label":"dguy:kafka-4494","ref":"kafka-4494","sha":"8bcf395014553448d2a3846a15f79d2f1d20de8a","user":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"repo":{"id":56393114,"name":"kafka","full_name":"dguy/kafka","owner":{"login":"dguy","id":54829,"avatar_url":"https://avatars.githubusercontent.com/u/54829?v=3","gravatar_id":"","url":"https://api.github.com/users/dguy","html_url":"https://github.com/dguy","followers_url":"https://api.github.com/users/dguy/followers","following_url":"https://api.github.com/users/dguy/following{/other_user}","gists_url":"https://api.github.com/users/dguy/gists{/gist_id}","starred_url":"https://api.github.com/users/dguy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dguy/subscriptions","organizations_url":"https://api.github.com/users/dguy/orgs","repos_url":"https://api.github.com/users/dguy/repos","events_url":"https://api.github.com/users/dguy/events{/privacy}","received_events_url":"https://api.github.com/users/dguy/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/dguy/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/dguy/kafka","forks_url":"https://api.github.com/repos/dguy/kafka/forks","keys_url":"https://api.github.com/repos/dguy/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/dguy/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/dguy/kafka/teams","hooks_url":"https://api.github.com/repos/dguy/kafka/hooks","issue_events_url":"https://api.github.com/repos/dguy/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/dguy/kafka/events","assignees_url":"https://api.github.com/repos/dguy/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/dguy/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/dguy/kafka/tags","blobs_url":"https://api.github.com/repos/dguy/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/dguy/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/dguy/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/dguy/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/dguy/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/dguy/kafka/languages","stargazers_url":"https://api.github.com/repos/dguy/kafka/stargazers","contributors_url":"https://api.github.com/repos/dguy/kafka/contributors","subscribers_url":"https://api.github.com/repos/dguy/kafka/subscribers","subscription_url":"https://api.github.com/repos/dguy/kafka/subscription","commits_url":"https://api.github.com/repos/dguy/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/dguy/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/dguy/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/dguy/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/dguy/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/dguy/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/dguy/kafka/merges","archive_url":"https://api.github.com/repos/dguy/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/dguy/kafka/downloads","issues_url":"https://api.github.com/repos/dguy/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/dguy/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/dguy/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/dguy/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/dguy/kafka/labels{/name}","releases_url":"https://api.github.com/repos/dguy/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/dguy/kafka/deployments","created_at":"2016-04-16T16:12:53Z","updated_at":"2016-04-16T16:12:57Z","pushed_at":"2017-02-22T14:23:11Z","git_url":"git://github.com/dguy/kafka.git","ssh_url":"git@github.com:dguy/kafka.git","clone_url":"https://github.com/dguy/kafka.git","svn_url":"https://github.com/dguy/kafka","homepage":null,"size":43536,"stargazers_count":0,"watchers_count":0,"language":"Java","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":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"1f1e794ad0025bb2a33b7d8378a481f224b3bccc","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}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2017-02-23T03:14:56Z","pushed_at":"2017-02-22T23:30:28Z","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":43787,"stargazers_count":4631,"watchers_count":4631,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2894,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":282,"forks":2894,"open_issues":282,"watchers":4631,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/2560"},"html":{"href":"https://github.com/apache/kafka/pull/2560"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/2560"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/2560/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/2560/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/2560/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/8bcf395014553448d2a3846a15f79d2f1d20de8a"}}}}
{"actor":{"display_login":"dguy"}}