{"action":"created","comment":{"url":"https://api.github.com/repos/caskdata/cdap/pulls/comments/26261983","id":26261983,"diff_hunk":"@@ -29,37 +34,36 @@\n import java.io.Closeable;\n import java.io.IOException;\n import java.util.List;\n+import java.util.Set;\n \n /**\n  * A {@link co.cask.cdap.data2.queue.QueueProducer} that uses HBase as the storage for queue entries and consumers\n- * states. The row key has the following format:\n- *\n- * <pre>\n- * {@code\n- *\n- * row_key = <salt_prefix> <row_key_base>\n- * salt_prefix = 1 byte hash of <row_key_base>\n- * row_key_base = <queue_prefix> <write_point> <counter>\n- * queue_prefix = <name_hash> <queue_name>\n- * name_hash = First byte of MD5 of <queue_name>\n- * queue_name = flowlet_name + \"/\" + output_name\n- * write_pointer = 8 bytes long value of the write pointer of the transaction\n- * counter = 4 bytes int value of a monotonic increasing number assigned for each entry written in the same transaction\n- * }\n- * </pre>\n+ * states.\n  */\n public class HBaseQueueProducer extends AbstractQueueProducer implements Closeable {\n \n   private final HBaseQueueStrategy queueStrategy;\n+  private final List<ConsumerGroupConfig> consumerGroupConfigs;\n   private final byte[] queueRowPrefix;\n   private final HTable hTable;\n   private final List<byte[]> rollbackKeys;\n \n   public HBaseQueueProducer(HTable hTable, QueueName queueName,\n-                            QueueMetrics queueMetrics, HBaseQueueStrategy queueStrategy) {\n+                            QueueMetrics queueMetrics, HBaseQueueStrategy queueStrategy,\n+                            Iterable<? extends ConsumerGroupConfig> consumerGroupConfigs) {\n     super(queueMetrics, queueName);\n     this.queueStrategy = queueStrategy;\n-    // Base row key = queue_name + writePointer + counter\n+    // Make sure only one config per consumer group\n+    this.consumerGroupConfigs = ImmutableList.copyOf(\n+      Iterables.filter(consumerGroupConfigs, new Predicate<ConsumerGroupConfig>() {\n+        private final Set<Long> seenGroups = Sets.newHashSet();\n+          @Override\n+          public boolean apply(ConsumerGroupConfig config) {\n+            return seenGroups.add(config.getGroupId());","path":"cdap-data-fabric/src/main/java/co/cask/cdap/data2/transaction/queue/hbase/HBaseQueueProducer.java","position":64,"original_position":64,"commit_id":"1e03b70cfa716d51e895e6ff978c83886b9cb6b2","original_commit_id":"1e03b70cfa716d51e895e6ff978c83886b9cb6b2","user":{"login":"abaranau","id":434133,"avatar_url":"https://avatars.githubusercontent.com/u/434133?v=3","gravatar_id":"","url":"https://api.github.com/users/abaranau","html_url":"https://github.com/abaranau","followers_url":"https://api.github.com/users/abaranau/followers","following_url":"https://api.github.com/users/abaranau/following{/other_user}","gists_url":"https://api.github.com/users/abaranau/gists{/gist_id}","starred_url":"https://api.github.com/users/abaranau/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/abaranau/subscriptions","organizations_url":"https://api.github.com/users/abaranau/orgs","repos_url":"https://api.github.com/users/abaranau/repos","events_url":"https://api.github.com/users/abaranau/events{/privacy}","received_events_url":"https://api.github.com/users/abaranau/received_events","type":"User","site_admin":false},"body":"would it indicated an error if there's more than one?","created_at":"2015-03-11T22:19:36Z","updated_at":"2015-03-11T22:19:36Z","html_url":"https://github.com/caskdata/cdap/pull/1583#discussion_r26261983","pull_request_url":"https://api.github.com/repos/caskdata/cdap/pulls/1583","_links":{"self":{"href":"https://api.github.com/repos/caskdata/cdap/pulls/comments/26261983"},"html":{"href":"https://github.com/caskdata/cdap/pull/1583#discussion_r26261983"},"pull_request":{"href":"https://api.github.com/repos/caskdata/cdap/pulls/1583"}}},"pull_request":{"url":"https://api.github.com/repos/caskdata/cdap/pulls/1583","id":30975280,"html_url":"https://github.com/caskdata/cdap/pull/1583","diff_url":"https://github.com/caskdata/cdap/pull/1583.diff","patch_url":"https://github.com/caskdata/cdap/pull/1583.patch","issue_url":"https://api.github.com/repos/caskdata/cdap/issues/1583","number":1583,"state":"open","locked":false,"title":"(CDAP-1204) Adds ShardedHBaseQueueStrategy","user":{"login":"chtyim","id":885032,"avatar_url":"https://avatars.githubusercontent.com/u/885032?v=3","gravatar_id":"","url":"https://api.github.com/users/chtyim","html_url":"https://github.com/chtyim","followers_url":"https://api.github.com/users/chtyim/followers","following_url":"https://api.github.com/users/chtyim/following{/other_user}","gists_url":"https://api.github.com/users/chtyim/gists{/gist_id}","starred_url":"https://api.github.com/users/chtyim/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chtyim/subscriptions","organizations_url":"https://api.github.com/users/chtyim/orgs","repos_url":"https://api.github.com/users/chtyim/repos","events_url":"https://api.github.com/users/chtyim/events{/privacy}","received_events_url":"https://api.github.com/users/chtyim/received_events","type":"User","site_admin":false},"body":"- Also includes refactoring on HBaseQueueProducer for future changes that includes ConsumerConfigGroup information","created_at":"2015-03-11T19:10:01Z","updated_at":"2015-03-11T22:19:36Z","closed_at":null,"merged_at":null,"merge_commit_sha":"cf601d05c9bcb1719f63e6e78ea782aa0ab67183","assignee":{"login":"abaranau","id":434133,"avatar_url":"https://avatars.githubusercontent.com/u/434133?v=3","gravatar_id":"","url":"https://api.github.com/users/abaranau","html_url":"https://github.com/abaranau","followers_url":"https://api.github.com/users/abaranau/followers","following_url":"https://api.github.com/users/abaranau/following{/other_user}","gists_url":"https://api.github.com/users/abaranau/gists{/gist_id}","starred_url":"https://api.github.com/users/abaranau/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/abaranau/subscriptions","organizations_url":"https://api.github.com/users/abaranau/orgs","repos_url":"https://api.github.com/users/abaranau/repos","events_url":"https://api.github.com/users/abaranau/events{/privacy}","received_events_url":"https://api.github.com/users/abaranau/received_events","type":"User","site_admin":false},"milestone":{"url":"https://api.github.com/repos/caskdata/cdap/milestones/3","html_url":"https://github.com/caskdata/cdap/milestones/2.8.0","labels_url":"https://api.github.com/repos/caskdata/cdap/milestones/3/labels","id":944584,"number":3,"title":"2.8.0","description":null,"creator":{"login":"jwang47","id":1084704,"avatar_url":"https://avatars.githubusercontent.com/u/1084704?v=3","gravatar_id":"","url":"https://api.github.com/users/jwang47","html_url":"https://github.com/jwang47","followers_url":"https://api.github.com/users/jwang47/followers","following_url":"https://api.github.com/users/jwang47/following{/other_user}","gists_url":"https://api.github.com/users/jwang47/gists{/gist_id}","starred_url":"https://api.github.com/users/jwang47/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jwang47/subscriptions","organizations_url":"https://api.github.com/users/jwang47/orgs","repos_url":"https://api.github.com/users/jwang47/repos","events_url":"https://api.github.com/users/jwang47/events{/privacy}","received_events_url":"https://api.github.com/users/jwang47/received_events","type":"User","site_admin":false},"open_issues":8,"closed_issues":184,"state":"open","created_at":"2015-01-21T01:15:14Z","updated_at":"2015-03-11T21:42:35Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/caskdata/cdap/pulls/1583/commits","review_comments_url":"https://api.github.com/repos/caskdata/cdap/pulls/1583/comments","review_comment_url":"https://api.github.com/repos/caskdata/cdap/pulls/comments{/number}","comments_url":"https://api.github.com/repos/caskdata/cdap/issues/1583/comments","statuses_url":"https://api.github.com/repos/caskdata/cdap/statuses/1e03b70cfa716d51e895e6ff978c83886b9cb6b2","head":{"label":"caskdata:feature/sharded-queue-strategy","ref":"feature/sharded-queue-strategy","sha":"1e03b70cfa716d51e895e6ff978c83886b9cb6b2","user":{"login":"caskdata","id":8123023,"avatar_url":"https://avatars.githubusercontent.com/u/8123023?v=3","gravatar_id":"","url":"https://api.github.com/users/caskdata","html_url":"https://github.com/caskdata","followers_url":"https://api.github.com/users/caskdata/followers","following_url":"https://api.github.com/users/caskdata/following{/other_user}","gists_url":"https://api.github.com/users/caskdata/gists{/gist_id}","starred_url":"https://api.github.com/users/caskdata/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/caskdata/subscriptions","organizations_url":"https://api.github.com/users/caskdata/orgs","repos_url":"https://api.github.com/users/caskdata/repos","events_url":"https://api.github.com/users/caskdata/events{/privacy}","received_events_url":"https://api.github.com/users/caskdata/received_events","type":"Organization","site_admin":false},"repo":{"id":22542759,"name":"cdap","full_name":"caskdata/cdap","owner":{"login":"caskdata","id":8123023,"avatar_url":"https://avatars.githubusercontent.com/u/8123023?v=3","gravatar_id":"","url":"https://api.github.com/users/caskdata","html_url":"https://github.com/caskdata","followers_url":"https://api.github.com/users/caskdata/followers","following_url":"https://api.github.com/users/caskdata/following{/other_user}","gists_url":"https://api.github.com/users/caskdata/gists{/gist_id}","starred_url":"https://api.github.com/users/caskdata/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/caskdata/subscriptions","organizations_url":"https://api.github.com/users/caskdata/orgs","repos_url":"https://api.github.com/users/caskdata/repos","events_url":"https://api.github.com/users/caskdata/events{/privacy}","received_events_url":"https://api.github.com/users/caskdata/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/caskdata/cdap","description":"Cask Data Application Platform (CDAP)","fork":false,"url":"https://api.github.com/repos/caskdata/cdap","forks_url":"https://api.github.com/repos/caskdata/cdap/forks","keys_url":"https://api.github.com/repos/caskdata/cdap/keys{/key_id}","collaborators_url":"https://api.github.com/repos/caskdata/cdap/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/caskdata/cdap/teams","hooks_url":"https://api.github.com/repos/caskdata/cdap/hooks","issue_events_url":"https://api.github.com/repos/caskdata/cdap/issues/events{/number}","events_url":"https://api.github.com/repos/caskdata/cdap/events","assignees_url":"https://api.github.com/repos/caskdata/cdap/assignees{/user}","branches_url":"https://api.github.com/repos/caskdata/cdap/branches{/branch}","tags_url":"https://api.github.com/repos/caskdata/cdap/tags","blobs_url":"https://api.github.com/repos/caskdata/cdap/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/caskdata/cdap/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/caskdata/cdap/git/refs{/sha}","trees_url":"https://api.github.com/repos/caskdata/cdap/git/trees{/sha}","statuses_url":"https://api.github.com/repos/caskdata/cdap/statuses/{sha}","languages_url":"https://api.github.com/repos/caskdata/cdap/languages","stargazers_url":"https://api.github.com/repos/caskdata/cdap/stargazers","contributors_url":"https://api.github.com/repos/caskdata/cdap/contributors","subscribers_url":"https://api.github.com/repos/caskdata/cdap/subscribers","subscription_url":"https://api.github.com/repos/caskdata/cdap/subscription","commits_url":"https://api.github.com/repos/caskdata/cdap/commits{/sha}","git_commits_url":"https://api.github.com/repos/caskdata/cdap/git/commits{/sha}","comments_url":"https://api.github.com/repos/caskdata/cdap/comments{/number}","issue_comment_url":"https://api.github.com/repos/caskdata/cdap/issues/comments{/number}","contents_url":"https://api.github.com/repos/caskdata/cdap/contents/{+path}","compare_url":"https://api.github.com/repos/caskdata/cdap/compare/{base}...{head}","merges_url":"https://api.github.com/repos/caskdata/cdap/merges","archive_url":"https://api.github.com/repos/caskdata/cdap/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/caskdata/cdap/downloads","issues_url":"https://api.github.com/repos/caskdata/cdap/issues{/number}","pulls_url":"https://api.github.com/repos/caskdata/cdap/pulls{/number}","milestones_url":"https://api.github.com/repos/caskdata/cdap/milestones{/number}","notifications_url":"https://api.github.com/repos/caskdata/cdap/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/caskdata/cdap/labels{/name}","releases_url":"https://api.github.com/repos/caskdata/cdap/releases{/id}","created_at":"2014-08-02T09:00:10Z","updated_at":"2015-03-11T21:48:52Z","pushed_at":"2015-03-11T22:12:58Z","git_url":"git://github.com/caskdata/cdap.git","ssh_url":"git@github.com:caskdata/cdap.git","clone_url":"https://github.com/caskdata/cdap.git","svn_url":"https://github.com/caskdata/cdap","homepage":"http://cdap.io","size":812083,"stargazers_count":33,"watchers_count":33,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":9,"mirror_url":null,"open_issues_count":19,"forks":9,"open_issues":19,"watchers":33,"default_branch":"develop"}},"base":{"label":"caskdata:develop","ref":"develop","sha":"610ff8b7add4e899b1412150949f2ded24dd8c4a","user":{"login":"caskdata","id":8123023,"avatar_url":"https://avatars.githubusercontent.com/u/8123023?v=3","gravatar_id":"","url":"https://api.github.com/users/caskdata","html_url":"https://github.com/caskdata","followers_url":"https://api.github.com/users/caskdata/followers","following_url":"https://api.github.com/users/caskdata/following{/other_user}","gists_url":"https://api.github.com/users/caskdata/gists{/gist_id}","starred_url":"https://api.github.com/users/caskdata/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/caskdata/subscriptions","organizations_url":"https://api.github.com/users/caskdata/orgs","repos_url":"https://api.github.com/users/caskdata/repos","events_url":"https://api.github.com/users/caskdata/events{/privacy}","received_events_url":"https://api.github.com/users/caskdata/received_events","type":"Organization","site_admin":false},"repo":{"id":22542759,"name":"cdap","full_name":"caskdata/cdap","owner":{"login":"caskdata","id":8123023,"avatar_url":"https://avatars.githubusercontent.com/u/8123023?v=3","gravatar_id":"","url":"https://api.github.com/users/caskdata","html_url":"https://github.com/caskdata","followers_url":"https://api.github.com/users/caskdata/followers","following_url":"https://api.github.com/users/caskdata/following{/other_user}","gists_url":"https://api.github.com/users/caskdata/gists{/gist_id}","starred_url":"https://api.github.com/users/caskdata/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/caskdata/subscriptions","organizations_url":"https://api.github.com/users/caskdata/orgs","repos_url":"https://api.github.com/users/caskdata/repos","events_url":"https://api.github.com/users/caskdata/events{/privacy}","received_events_url":"https://api.github.com/users/caskdata/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/caskdata/cdap","description":"Cask Data Application Platform (CDAP)","fork":false,"url":"https://api.github.com/repos/caskdata/cdap","forks_url":"https://api.github.com/repos/caskdata/cdap/forks","keys_url":"https://api.github.com/repos/caskdata/cdap/keys{/key_id}","collaborators_url":"https://api.github.com/repos/caskdata/cdap/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/caskdata/cdap/teams","hooks_url":"https://api.github.com/repos/caskdata/cdap/hooks","issue_events_url":"https://api.github.com/repos/caskdata/cdap/issues/events{/number}","events_url":"https://api.github.com/repos/caskdata/cdap/events","assignees_url":"https://api.github.com/repos/caskdata/cdap/assignees{/user}","branches_url":"https://api.github.com/repos/caskdata/cdap/branches{/branch}","tags_url":"https://api.github.com/repos/caskdata/cdap/tags","blobs_url":"https://api.github.com/repos/caskdata/cdap/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/caskdata/cdap/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/caskdata/cdap/git/refs{/sha}","trees_url":"https://api.github.com/repos/caskdata/cdap/git/trees{/sha}","statuses_url":"https://api.github.com/repos/caskdata/cdap/statuses/{sha}","languages_url":"https://api.github.com/repos/caskdata/cdap/languages","stargazers_url":"https://api.github.com/repos/caskdata/cdap/stargazers","contributors_url":"https://api.github.com/repos/caskdata/cdap/contributors","subscribers_url":"https://api.github.com/repos/caskdata/cdap/subscribers","subscription_url":"https://api.github.com/repos/caskdata/cdap/subscription","commits_url":"https://api.github.com/repos/caskdata/cdap/commits{/sha}","git_commits_url":"https://api.github.com/repos/caskdata/cdap/git/commits{/sha}","comments_url":"https://api.github.com/repos/caskdata/cdap/comments{/number}","issue_comment_url":"https://api.github.com/repos/caskdata/cdap/issues/comments{/number}","contents_url":"https://api.github.com/repos/caskdata/cdap/contents/{+path}","compare_url":"https://api.github.com/repos/caskdata/cdap/compare/{base}...{head}","merges_url":"https://api.github.com/repos/caskdata/cdap/merges","archive_url":"https://api.github.com/repos/caskdata/cdap/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/caskdata/cdap/downloads","issues_url":"https://api.github.com/repos/caskdata/cdap/issues{/number}","pulls_url":"https://api.github.com/repos/caskdata/cdap/pulls{/number}","milestones_url":"https://api.github.com/repos/caskdata/cdap/milestones{/number}","notifications_url":"https://api.github.com/repos/caskdata/cdap/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/caskdata/cdap/labels{/name}","releases_url":"https://api.github.com/repos/caskdata/cdap/releases{/id}","created_at":"2014-08-02T09:00:10Z","updated_at":"2015-03-11T21:48:52Z","pushed_at":"2015-03-11T22:12:58Z","git_url":"git://github.com/caskdata/cdap.git","ssh_url":"git@github.com:caskdata/cdap.git","clone_url":"https://github.com/caskdata/cdap.git","svn_url":"https://github.com/caskdata/cdap","homepage":"http://cdap.io","size":812083,"stargazers_count":33,"watchers_count":33,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":9,"mirror_url":null,"open_issues_count":19,"forks":9,"open_issues":19,"watchers":33,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/caskdata/cdap/pulls/1583"},"html":{"href":"https://github.com/caskdata/cdap/pull/1583"},"issue":{"href":"https://api.github.com/repos/caskdata/cdap/issues/1583"},"comments":{"href":"https://api.github.com/repos/caskdata/cdap/issues/1583/comments"},"review_comments":{"href":"https://api.github.com/repos/caskdata/cdap/pulls/1583/comments"},"review_comment":{"href":"https://api.github.com/repos/caskdata/cdap/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/caskdata/cdap/pulls/1583/commits"},"statuses":{"href":"https://api.github.com/repos/caskdata/cdap/statuses/1e03b70cfa716d51e895e6ff978c83886b9cb6b2"}}}}