{"action":"created","comment":{"url":"https://api.github.com/repos/apache/storm/pulls/comments/22678017","id":22678017,"diff_hunk":"@@ -0,0 +1,154 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the 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\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 storm.kafka;\n+\n+import java.util.PriorityQueue;\n+import java.util.Queue;\n+import java.util.SortedMap;\n+import java.util.TreeMap;\n+\n+public class ExponentialBackoffMsgRetryManager implements FailedMsgRetryManager {\n+\n+    private final long retryInitialDelayMs;\n+    private final double retryDelayMultiplier;\n+    private final long retryDelayMaxMs;\n+\n+    private Queue<MessageRetryRecord> waiting = new PriorityQueue<MessageRetryRecord>();\n+    private SortedMap<Long,MessageRetryRecord> records = new TreeMap<Long,MessageRetryRecord>();\n+\n+    public ExponentialBackoffMsgRetryManager(long retryInitialDelayMs, double retryDelayMultiplier, long retryDelayMaxMs) {\n+        this.retryInitialDelayMs = retryInitialDelayMs;\n+        this.retryDelayMultiplier = retryDelayMultiplier;\n+        this.retryDelayMaxMs = retryDelayMaxMs;\n+    }\n+\n+    @Override\n+    public void failed(Long offset) {\n+        MessageRetryRecord oldRecord = this.records.get(offset);\n+        MessageRetryRecord newRecord = oldRecord == null ?\n+                                       new MessageRetryRecord(offset) :\n+                                       oldRecord.createNextRetryRecord();\n+        this.records.put(offset, newRecord);\n+        this.waiting.add(newRecord);\n+    }\n+\n+    @Override\n+    public void acked(Long offset) {\n+        MessageRetryRecord record = this.records.remove(offset);\n+        if (record != null) {\n+            this.waiting.remove(record);\n+        }\n+    }\n+\n+    @Override\n+    public void retryStarted(Long offset) {\n+        MessageRetryRecord record = this.records.get(offset);\n+        if (record == null || !this.waiting.contains(record)) {\n+            throw new IllegalStateException(\"cannot retry a message that has not failed\");\n+        } else {\n+            this.waiting.remove(record);\n+        }\n+    }\n+\n+    @Override\n+    public Long nextFailedMessageToRetry() {\n+        if (this.waiting.size() > 0) {\n+            MessageRetryRecord first = this.waiting.peek();\n+            if (System.currentTimeMillis() >= first.retryTimeUTC) {\n+                if (this.records.containsKey(first.offset)) {\n+                    return first.offset;\n+                } else {\n+                    // defensive programming - should be impossible\n+                    this.waiting.remove(first);\n+                    return nextFailedMessageToRetry();\n+                }\n+            }\n+        }\n+        return null;\n+    }\n+\n+    @Override\n+    public boolean shouldRetryMsg(Long offset) {\n+        MessageRetryRecord record = this.records.get(offset);\n+        return record != null &&\n+                this.waiting.contains(record) &&\n+                System.currentTimeMillis() >= record.retryTimeUTC;\n+    }\n+\n+    /**\n+     * A MessageRetryRecord holds the data of how many times a message has\n+     * failed and been retried, and when the last failure occurred.  It can\n+     * determine whether it is ready to be retried by employing an exponential\n+     * back-off calculation using config values stored in SpoutConfig:\n+     * <ul>\n+     *  <li>retryInitialDelayMs - time to delay before the first retry</li>\n+     *  <li>retryDelayMultiplier - multiplier by which to increase the delay for each subsequent retry</li>\n+     *  <li>retryDelayMaxMs - maximum retry delay (once this delay time is reached, subsequent retries will\n+     *                        delay for this amount of time every time)\n+     *  </li>\n+     * </ul>\n+     */\n+    class MessageRetryRecord implements Comparable<MessageRetryRecord> {\n+        private final long offset;\n+        private final int retryNum;\n+        private final long retryTimeUTC;\n+\n+        public MessageRetryRecord(long offset) {\n+            this(offset, 1);\n+        }\n+\n+        private MessageRetryRecord(long offset, int retryNum) {\n+            this.offset = offset;\n+            this.retryNum = retryNum;\n+            this.retryTimeUTC = System.currentTimeMillis() + calculateRetryDelay();\n+        }\n+\n+        /**\n+         * Create a MessageRetryRecord for the next retry that should occur after this one.\n+         * @return MessageRetryRecord with the next retry time, or null to indicate that another\n+         *         retry should not be performed.  The latter case can happen if we are about to\n+         *         run into the backtype.storm.Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS in the Storm\n+         *         configuration.\n+         */\n+        public MessageRetryRecord createNextRetryRecord() {\n+            return new MessageRetryRecord(this.offset, this.retryNum + 1);\n+        }\n+\n+        private long calculateRetryDelay() {\n+            double delayMultiplier = Math.pow(retryDelayMultiplier, this.retryNum - 1);","path":"external/storm-kafka/src/jvm/storm/kafka/ExponentialBackoffMsgRetryManager.java","position":133,"original_position":133,"commit_id":"f99fe8ee38249951688d11c8dc93d39d24c6b0ac","original_commit_id":"f99fe8ee38249951688d11c8dc93d39d24c6b0ac","user":{"login":"revans2","id":3441321,"avatar_url":"https://avatars.githubusercontent.com/u/3441321?v=3","gravatar_id":"","url":"https://api.github.com/users/revans2","html_url":"https://github.com/revans2","followers_url":"https://api.github.com/users/revans2/followers","following_url":"https://api.github.com/users/revans2/following{/other_user}","gists_url":"https://api.github.com/users/revans2/gists{/gist_id}","starred_url":"https://api.github.com/users/revans2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/revans2/subscriptions","organizations_url":"https://api.github.com/users/revans2/orgs","repos_url":"https://api.github.com/users/revans2/repos","events_url":"https://api.github.com/users/revans2/events{/privacy}","received_events_url":"https://api.github.com/users/revans2/received_events","type":"User","site_admin":false},"body":"We have had issues with other exponential backoff code in the past where the retry count would overflow the number of bits which would result in a negative delay.  This code looks like it is also susceptible to that.  Because the math is done in longs and doubles I think it is OK for just about all use cases, but it would be nice to either fix it, or file a JIRA to follow up on it in the future.","created_at":"2015-01-08T20:14:14Z","updated_at":"2015-01-08T20:14:14Z","html_url":"https://github.com/apache/storm/pull/254#discussion_r22678017","pull_request_url":"https://api.github.com/repos/apache/storm/pulls/254","_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/comments/22678017"},"html":{"href":"https://github.com/apache/storm/pull/254#discussion_r22678017"},"pull_request":{"href":"https://api.github.com/repos/apache/storm/pulls/254"}}},"pull_request":{"url":"https://api.github.com/repos/apache/storm/pulls/254","id":21052433,"html_url":"https://github.com/apache/storm/pull/254","diff_url":"https://github.com/apache/storm/pull/254.diff","patch_url":"https://github.com/apache/storm/pull/254.patch","issue_url":"https://api.github.com/repos/apache/storm/issues/254","number":254,"state":"open","locked":false,"title":"STORM-495: KafkaSpout retries with exponential backoff","user":{"login":"rick-kilgore","id":4665096,"avatar_url":"https://avatars.githubusercontent.com/u/4665096?v=3","gravatar_id":"","url":"https://api.github.com/users/rick-kilgore","html_url":"https://github.com/rick-kilgore","followers_url":"https://api.github.com/users/rick-kilgore/followers","following_url":"https://api.github.com/users/rick-kilgore/following{/other_user}","gists_url":"https://api.github.com/users/rick-kilgore/gists{/gist_id}","starred_url":"https://api.github.com/users/rick-kilgore/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rick-kilgore/subscriptions","organizations_url":"https://api.github.com/users/rick-kilgore/orgs","repos_url":"https://api.github.com/users/rick-kilgore/repos","events_url":"https://api.github.com/users/rick-kilgore/events{/privacy}","received_events_url":"https://api.github.com/users/rick-kilgore/received_events","type":"User","site_admin":false},"body":"This is an implementation of the improvement request STORM-495:\r\n    https://issues.apache.org/jira/browse/STORM-495\r\n\r\nThe KafkaSpout keeps track of failed messages through multiple retries, and delays their resubmission to the topology with an exponentially increasing delay time.  The spout does not terminate the retry cycle (it is my conviction that it should not do so, because it cannot report context about the failure that happened to abort the reqeust), it only handles delaying the retries.  A bolt in the topology is still expected to eventually call ack() instead of fail() to stop the cycle.","created_at":"2014-09-10T21:41:05Z","updated_at":"2015-01-08T20:14:14Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8a5215a383e2094cb29a9b062c0a763d2929be1d","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/storm/pulls/254/commits","review_comments_url":"https://api.github.com/repos/apache/storm/pulls/254/comments","review_comment_url":"https://api.github.com/repos/apache/storm/pulls/comments/{number}","comments_url":"https://api.github.com/repos/apache/storm/issues/254/comments","statuses_url":"https://api.github.com/repos/apache/storm/statuses/f99fe8ee38249951688d11c8dc93d39d24c6b0ac","head":{"label":"HBOCodeLabs:retries","ref":"retries","sha":"f99fe8ee38249951688d11c8dc93d39d24c6b0ac","user":{"login":"HBOCodeLabs","id":2342721,"avatar_url":"https://avatars.githubusercontent.com/u/2342721?v=3","gravatar_id":"","url":"https://api.github.com/users/HBOCodeLabs","html_url":"https://github.com/HBOCodeLabs","followers_url":"https://api.github.com/users/HBOCodeLabs/followers","following_url":"https://api.github.com/users/HBOCodeLabs/following{/other_user}","gists_url":"https://api.github.com/users/HBOCodeLabs/gists{/gist_id}","starred_url":"https://api.github.com/users/HBOCodeLabs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HBOCodeLabs/subscriptions","organizations_url":"https://api.github.com/users/HBOCodeLabs/orgs","repos_url":"https://api.github.com/users/HBOCodeLabs/repos","events_url":"https://api.github.com/users/HBOCodeLabs/events{/privacy}","received_events_url":"https://api.github.com/users/HBOCodeLabs/received_events","type":"Organization","site_admin":false},"repo":{"id":23728923,"name":"incubator-storm","full_name":"HBOCodeLabs/incubator-storm","owner":{"login":"HBOCodeLabs","id":2342721,"avatar_url":"https://avatars.githubusercontent.com/u/2342721?v=3","gravatar_id":"","url":"https://api.github.com/users/HBOCodeLabs","html_url":"https://github.com/HBOCodeLabs","followers_url":"https://api.github.com/users/HBOCodeLabs/followers","following_url":"https://api.github.com/users/HBOCodeLabs/following{/other_user}","gists_url":"https://api.github.com/users/HBOCodeLabs/gists{/gist_id}","starred_url":"https://api.github.com/users/HBOCodeLabs/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HBOCodeLabs/subscriptions","organizations_url":"https://api.github.com/users/HBOCodeLabs/orgs","repos_url":"https://api.github.com/users/HBOCodeLabs/repos","events_url":"https://api.github.com/users/HBOCodeLabs/events{/privacy}","received_events_url":"https://api.github.com/users/HBOCodeLabs/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/HBOCodeLabs/incubator-storm","description":"Mirror of Apache Storm","fork":true,"url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm","forks_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/forks","keys_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/teams","hooks_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/hooks","issue_events_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/issues/events{/number}","events_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/events","assignees_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/assignees{/user}","branches_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/branches{/branch}","tags_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/tags","blobs_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/statuses/{sha}","languages_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/languages","stargazers_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/stargazers","contributors_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/contributors","subscribers_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/subscribers","subscription_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/subscription","commits_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/issues/comments/{number}","contents_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/contents/{+path}","compare_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/merges","archive_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/downloads","issues_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/issues{/number}","pulls_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/pulls{/number}","milestones_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/milestones{/number}","notifications_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/labels{/name}","releases_url":"https://api.github.com/repos/HBOCodeLabs/incubator-storm/releases{/id}","created_at":"2014-09-06T06:50:08Z","updated_at":"2014-12-03T19:59:32Z","pushed_at":"2014-12-03T22:12:31Z","git_url":"git://github.com/HBOCodeLabs/incubator-storm.git","ssh_url":"git@github.com:HBOCodeLabs/incubator-storm.git","clone_url":"https://github.com/HBOCodeLabs/incubator-storm.git","svn_url":"https://github.com/HBOCodeLabs/incubator-storm","homepage":null,"size":12174,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":0,"mirror_url":null,"open_issues_count":2,"forks":0,"open_issues":2,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"c064dbaa83897ba81de835256935c5b7cde2784d","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":14135470,"name":"storm","full_name":"apache/storm","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/storm","description":"Mirror of Apache Storm","fork":false,"url":"https://api.github.com/repos/apache/storm","forks_url":"https://api.github.com/repos/apache/storm/forks","keys_url":"https://api.github.com/repos/apache/storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/storm/teams","hooks_url":"https://api.github.com/repos/apache/storm/hooks","issue_events_url":"https://api.github.com/repos/apache/storm/issues/events{/number}","events_url":"https://api.github.com/repos/apache/storm/events","assignees_url":"https://api.github.com/repos/apache/storm/assignees{/user}","branches_url":"https://api.github.com/repos/apache/storm/branches{/branch}","tags_url":"https://api.github.com/repos/apache/storm/tags","blobs_url":"https://api.github.com/repos/apache/storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/storm/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/storm/languages","stargazers_url":"https://api.github.com/repos/apache/storm/stargazers","contributors_url":"https://api.github.com/repos/apache/storm/contributors","subscribers_url":"https://api.github.com/repos/apache/storm/subscribers","subscription_url":"https://api.github.com/repos/apache/storm/subscription","commits_url":"https://api.github.com/repos/apache/storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/storm/issues/comments/{number}","contents_url":"https://api.github.com/repos/apache/storm/contents/{+path}","compare_url":"https://api.github.com/repos/apache/storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/storm/merges","archive_url":"https://api.github.com/repos/apache/storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/storm/downloads","issues_url":"https://api.github.com/repos/apache/storm/issues{/number}","pulls_url":"https://api.github.com/repos/apache/storm/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/storm/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/storm/labels{/name}","releases_url":"https://api.github.com/repos/apache/storm/releases{/id}","created_at":"2013-11-05T08:00:14Z","updated_at":"2015-01-08T17:35:40Z","pushed_at":"2015-01-06T22:44:21Z","git_url":"git://github.com/apache/storm.git","ssh_url":"git@github.com:apache/storm.git","clone_url":"https://github.com/apache/storm.git","svn_url":"https://github.com/apache/storm","homepage":null,"size":767495,"stargazers_count":1387,"watchers_count":1387,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":871,"mirror_url":"git://git.apache.org/incubator-storm.git","open_issues_count":53,"forks":871,"open_issues":53,"watchers":1387,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/254"},"html":{"href":"https://github.com/apache/storm/pull/254"},"issue":{"href":"https://api.github.com/repos/apache/storm/issues/254"},"comments":{"href":"https://api.github.com/repos/apache/storm/issues/254/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/storm/pulls/254/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/storm/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/apache/storm/pulls/254/commits"},"statuses":{"href":"https://api.github.com/repos/apache/storm/statuses/f99fe8ee38249951688d11c8dc93d39d24c6b0ac"}}}}