{"action":"created","comment":{"url":"https://api.github.com/repos/pravega/pravega/pulls/comments/190123359","pull_request_review_id":122433589,"id":190123359,"diff_hunk":"@@ -0,0 +1,174 @@\n+/**\n+ * Copyright (c) 2017 Dell Inc., or its subsidiaries. All Rights Reserved.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\");\n+ * you may not use this file except in compliance with the License.\n+ * You may obtain a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ */\n+package io.pravega.controller.server.eventProcessor.requesthandlers;\n+\n+import com.google.common.annotations.VisibleForTesting;\n+import com.google.common.base.Preconditions;\n+import io.pravega.common.concurrent.Futures;\n+import io.pravega.controller.store.stream.OperationContext;\n+import io.pravega.controller.store.stream.StoreException;\n+import io.pravega.controller.store.stream.StreamMetadataStore;\n+import io.pravega.controller.store.stream.TxnStatus;\n+import io.pravega.controller.task.Stream.StreamMetadataTasks;\n+import io.pravega.shared.controller.event.CommitEvent;\n+import lombok.extern.slf4j.Slf4j;\n+\n+import java.util.List;\n+import java.util.Map;\n+import java.util.UUID;\n+import java.util.concurrent.BlockingQueue;\n+import java.util.concurrent.CompletableFuture;\n+import java.util.concurrent.ScheduledExecutorService;\n+import java.util.stream.Collectors;\n+\n+/**\n+ * Request handler for processing commit events in request-stream.\n+ */\n+@Slf4j\n+public class CommitTransactionTask implements StreamTask<CommitEvent> {\n+    private final StreamMetadataTasks streamMetadataTasks;\n+    private final StreamMetadataStore streamMetadataStore;\n+    private final ScheduledExecutorService executor;\n+    private final BlockingQueue<CommitEvent> processedEvents;\n+\n+    @VisibleForTesting\n+    public CommitTransactionTask(final StreamMetadataStore streamMetadataStore,\n+                                final StreamMetadataTasks streamMetadataTasks,\n+                                final ScheduledExecutorService executor,\n+                                final BlockingQueue<CommitEvent> queue) {\n+        this.streamMetadataStore = streamMetadataStore;\n+        this.streamMetadataTasks = streamMetadataTasks;\n+        this.executor = executor;\n+        this.processedEvents = queue;\n+    }\n+\n+    public CommitTransactionTask(final StreamMetadataTasks streamMetadataTasks,\n+                                 final StreamMetadataStore streamMetadataStore,\n+                                 final ScheduledExecutorService executor) {\n+        Preconditions.checkNotNull(streamMetadataStore);\n+        Preconditions.checkNotNull(streamMetadataTasks);\n+        Preconditions.checkNotNull(executor);\n+        this.streamMetadataTasks = streamMetadataTasks;\n+        this.streamMetadataStore = streamMetadataStore;\n+        this.executor = executor;\n+        this.processedEvents = null;\n+    }\n+\n+    @Override\n+    public CompletableFuture<Void> execute(CommitEvent event) {\n+        // 1. get epoch.. check if the epoch matches the epoch in commit event.. if true, we can proceed.. else throw OperationNotAllowedException\n+        // 2. check if txn-commit-node exists.\n+        //      if node exists and has same epoch, process the node.\n+        // 3. if node doesnt exist, collect all committing txn in the epoch and create txn-commit-node.\n+        // 4. loop over the transactions and commit one transaction at a time until all such transactions are committed.\n+        // 5. delete txn-commit-node\n+        // 6. try complete scale.\n+        // Note: Once scale is completed, all events for this epoch will become a no-op as all transactions would have been committed already.\n+        String scope = event.getScope();\n+        String stream = event.getStream();\n+        int epoch = event.getEpoch();\n+        OperationContext context = streamMetadataStore.createContext(scope, stream);\n+        log.debug(\"Attempting to commit available transactions on epoch {} on stream {}/{}\", event.getEpoch(), event.getScope(), event.getStream());\n+\n+        return streamMetadataStore.getActiveEpoch(scope, stream, context, false, executor).thenComposeAsync(pair -> {","path":"controller/src/main/java/io/pravega/controller/server/eventProcessor/requesthandlers/CommitTransactionTask.java","position":80,"original_position":80,"commit_id":"a615c25fca4999a9911db6257a93b3c81abd6162","original_commit_id":"a615c25fca4999a9911db6257a93b3c81abd6162","user":{"login":"shiveshr","id":13659022,"avatar_url":"https://avatars2.githubusercontent.com/u/13659022?v=4","gravatar_id":"","url":"https://api.github.com/users/shiveshr","html_url":"https://github.com/shiveshr","followers_url":"https://api.github.com/users/shiveshr/followers","following_url":"https://api.github.com/users/shiveshr/following{/other_user}","gists_url":"https://api.github.com/users/shiveshr/gists{/gist_id}","starred_url":"https://api.github.com/users/shiveshr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shiveshr/subscriptions","organizations_url":"https://api.github.com/users/shiveshr/orgs","repos_url":"https://api.github.com/users/shiveshr/repos","events_url":"https://api.github.com/users/shiveshr/events{/privacy}","received_events_url":"https://api.github.com/users/shiveshr/received_events","type":"User","site_admin":false},"body":"replaced thenComposeAsync with thenCompose.\r\nThenComposeAsync came because of copy pasting that part of the code from the original class (CommitEventProcessor.java) which has been replaced by this new class. \r\n","created_at":"2018-05-23T05:01:54Z","updated_at":"2018-05-23T05:01:55Z","html_url":"https://github.com/pravega/pravega/pull/2611#discussion_r190123359","pull_request_url":"https://api.github.com/repos/pravega/pravega/pulls/2611","author_association":"COLLABORATOR","_links":{"self":{"href":"https://api.github.com/repos/pravega/pravega/pulls/comments/190123359"},"html":{"href":"https://github.com/pravega/pravega/pull/2611#discussion_r190123359"},"pull_request":{"href":"https://api.github.com/repos/pravega/pravega/pulls/2611"}},"in_reply_to_id":190060625},"pull_request":{"url":"https://api.github.com/repos/pravega/pravega/pulls/2611","id":189416144,"html_url":"https://github.com/pravega/pravega/pull/2611","diff_url":"https://github.com/pravega/pravega/pull/2611.diff","patch_url":"https://github.com/pravega/pravega/pull/2611.patch","issue_url":"https://api.github.com/repos/pravega/pravega/issues/2611","number":2611,"state":"open","locked":false,"title":"Issue 2324: Linearizing commit transaction and scale workflows by processing them on requeststream","user":{"login":"shiveshr","id":13659022,"avatar_url":"https://avatars2.githubusercontent.com/u/13659022?v=4","gravatar_id":"","url":"https://api.github.com/users/shiveshr","html_url":"https://github.com/shiveshr","followers_url":"https://api.github.com/users/shiveshr/followers","following_url":"https://api.github.com/users/shiveshr/following{/other_user}","gists_url":"https://api.github.com/users/shiveshr/gists{/gist_id}","starred_url":"https://api.github.com/users/shiveshr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shiveshr/subscriptions","organizations_url":"https://api.github.com/users/shiveshr/orgs","repos_url":"https://api.github.com/users/shiveshr/repos","events_url":"https://api.github.com/users/shiveshr/events{/privacy}","received_events_url":"https://api.github.com/users/shiveshr/received_events","type":"User","site_admin":false},"body":"**Change log description**\r\nWe have linearized commit transaction and scale by processing both on requeststream.\r\nThe commit transaction workflow is also modified to look for multiple transactions that are in committing state under the same epoch and merge them as part of a single workflow. \r\n\r\n**Purpose of the change**\r\nFixes #2324 \r\n\r\n**What the code does**\r\nWe have removed CommitEventProcessor and replaced it with CommitTransactionTask.java. \r\nCommitEventProcessor was a serial event processor with one event at a time. Moving the commit transaction logic to `ConcurrentEventProcessor` based task processing on requeststream means we can have transaction commit requests from older epochs also being replaced. And upon failover recovery we may recover at an older checkpoint whereby we need to identify if there is an ongoing transaction commit or not so that we continue the guarantee that exactly one transaction per stream is getting committed at any given time. \r\nFor this purpose we have created a new znode called CommitTxnList. This is implemented in our metadata store and metadata stream interfaces. This znode is transient and one znode per stream can exist and it captures the list of transaction ids and epoch against which a commit workflow is running currently. This node ensures that exactly one set of transaction commit can happen at any time without any conflicts that could arise from failover recovery. \r\n\r\nWe have also enhanced the commit workflow to not just do an individual transaction commit, rather it opportunistically collects all transactions marked for commit (state=committing) under the epoch and stores them in the CommitTxnList and fixes the order in which their commit workflows will be executed. \r\n\r\nTo ensure that we have the ability to commit multiple transactions on the same epoch against a single event, we have removed the transaction id from CommitEvent.java. Instead we pass the epoch. The CommitTransactionTask takes the epoch and compares it with active epoch. If the epoch matches active epoch then transactions under the epoch are collected for processing. \r\nIf its an older epoch, the message is discarded. If its higher epoch the message is reposted in the request stream and only processed when the scale completes.\r\n\r\nSo behaviour wise nothing changes from transactions perspective and its interplay with scale stays same. \r\n\r\n**How to verify it**\r\nunit tests added.","created_at":"2018-05-21T15:45:28Z","updated_at":"2018-05-23T05:01:55Z","closed_at":null,"merged_at":null,"merge_commit_sha":"604a57838bd61f72a79bdc13d4c2050f204e732e","assignee":null,"assignees":[],"requested_reviewers":[{"login":"fpj","id":572920,"avatar_url":"https://avatars3.githubusercontent.com/u/572920?v=4","gravatar_id":"","url":"https://api.github.com/users/fpj","html_url":"https://github.com/fpj","followers_url":"https://api.github.com/users/fpj/followers","following_url":"https://api.github.com/users/fpj/following{/other_user}","gists_url":"https://api.github.com/users/fpj/gists{/gist_id}","starred_url":"https://api.github.com/users/fpj/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fpj/subscriptions","organizations_url":"https://api.github.com/users/fpj/orgs","repos_url":"https://api.github.com/users/fpj/repos","events_url":"https://api.github.com/users/fpj/events{/privacy}","received_events_url":"https://api.github.com/users/fpj/received_events","type":"User","site_admin":false}],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/pravega/pravega/pulls/2611/commits","review_comments_url":"https://api.github.com/repos/pravega/pravega/pulls/2611/comments","review_comment_url":"https://api.github.com/repos/pravega/pravega/pulls/comments{/number}","comments_url":"https://api.github.com/repos/pravega/pravega/issues/2611/comments","statuses_url":"https://api.github.com/repos/pravega/pravega/statuses/a615c25fca4999a9911db6257a93b3c81abd6162","head":{"label":"shiveshr:issue-2324","ref":"issue-2324","sha":"a615c25fca4999a9911db6257a93b3c81abd6162","user":{"login":"shiveshr","id":13659022,"avatar_url":"https://avatars2.githubusercontent.com/u/13659022?v=4","gravatar_id":"","url":"https://api.github.com/users/shiveshr","html_url":"https://github.com/shiveshr","followers_url":"https://api.github.com/users/shiveshr/followers","following_url":"https://api.github.com/users/shiveshr/following{/other_user}","gists_url":"https://api.github.com/users/shiveshr/gists{/gist_id}","starred_url":"https://api.github.com/users/shiveshr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shiveshr/subscriptions","organizations_url":"https://api.github.com/users/shiveshr/orgs","repos_url":"https://api.github.com/users/shiveshr/repos","events_url":"https://api.github.com/users/shiveshr/events{/privacy}","received_events_url":"https://api.github.com/users/shiveshr/received_events","type":"User","site_admin":false},"repo":{"id":92372740,"name":"pravega-1","full_name":"shiveshr/pravega-1","owner":{"login":"shiveshr","id":13659022,"avatar_url":"https://avatars2.githubusercontent.com/u/13659022?v=4","gravatar_id":"","url":"https://api.github.com/users/shiveshr","html_url":"https://github.com/shiveshr","followers_url":"https://api.github.com/users/shiveshr/followers","following_url":"https://api.github.com/users/shiveshr/following{/other_user}","gists_url":"https://api.github.com/users/shiveshr/gists{/gist_id}","starred_url":"https://api.github.com/users/shiveshr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shiveshr/subscriptions","organizations_url":"https://api.github.com/users/shiveshr/orgs","repos_url":"https://api.github.com/users/shiveshr/repos","events_url":"https://api.github.com/users/shiveshr/events{/privacy}","received_events_url":"https://api.github.com/users/shiveshr/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/shiveshr/pravega-1","description":"Pravega - Streaming as a new software defined storage primitive","fork":true,"url":"https://api.github.com/repos/shiveshr/pravega-1","forks_url":"https://api.github.com/repos/shiveshr/pravega-1/forks","keys_url":"https://api.github.com/repos/shiveshr/pravega-1/keys{/key_id}","collaborators_url":"https://api.github.com/repos/shiveshr/pravega-1/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/shiveshr/pravega-1/teams","hooks_url":"https://api.github.com/repos/shiveshr/pravega-1/hooks","issue_events_url":"https://api.github.com/repos/shiveshr/pravega-1/issues/events{/number}","events_url":"https://api.github.com/repos/shiveshr/pravega-1/events","assignees_url":"https://api.github.com/repos/shiveshr/pravega-1/assignees{/user}","branches_url":"https://api.github.com/repos/shiveshr/pravega-1/branches{/branch}","tags_url":"https://api.github.com/repos/shiveshr/pravega-1/tags","blobs_url":"https://api.github.com/repos/shiveshr/pravega-1/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/shiveshr/pravega-1/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/shiveshr/pravega-1/git/refs{/sha}","trees_url":"https://api.github.com/repos/shiveshr/pravega-1/git/trees{/sha}","statuses_url":"https://api.github.com/repos/shiveshr/pravega-1/statuses/{sha}","languages_url":"https://api.github.com/repos/shiveshr/pravega-1/languages","stargazers_url":"https://api.github.com/repos/shiveshr/pravega-1/stargazers","contributors_url":"https://api.github.com/repos/shiveshr/pravega-1/contributors","subscribers_url":"https://api.github.com/repos/shiveshr/pravega-1/subscribers","subscription_url":"https://api.github.com/repos/shiveshr/pravega-1/subscription","commits_url":"https://api.github.com/repos/shiveshr/pravega-1/commits{/sha}","git_commits_url":"https://api.github.com/repos/shiveshr/pravega-1/git/commits{/sha}","comments_url":"https://api.github.com/repos/shiveshr/pravega-1/comments{/number}","issue_comment_url":"https://api.github.com/repos/shiveshr/pravega-1/issues/comments{/number}","contents_url":"https://api.github.com/repos/shiveshr/pravega-1/contents/{+path}","compare_url":"https://api.github.com/repos/shiveshr/pravega-1/compare/{base}...{head}","merges_url":"https://api.github.com/repos/shiveshr/pravega-1/merges","archive_url":"https://api.github.com/repos/shiveshr/pravega-1/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/shiveshr/pravega-1/downloads","issues_url":"https://api.github.com/repos/shiveshr/pravega-1/issues{/number}","pulls_url":"https://api.github.com/repos/shiveshr/pravega-1/pulls{/number}","milestones_url":"https://api.github.com/repos/shiveshr/pravega-1/milestones{/number}","notifications_url":"https://api.github.com/repos/shiveshr/pravega-1/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/shiveshr/pravega-1/labels{/name}","releases_url":"https://api.github.com/repos/shiveshr/pravega-1/releases{/id}","deployments_url":"https://api.github.com/repos/shiveshr/pravega-1/deployments","created_at":"2017-05-25T06:31:49Z","updated_at":"2018-05-18T06:45:05Z","pushed_at":"2018-05-23T04:48:56Z","git_url":"git://github.com/shiveshr/pravega-1.git","ssh_url":"git@github.com:shiveshr/pravega-1.git","clone_url":"https://github.com/shiveshr/pravega-1.git","svn_url":"https://github.com/shiveshr/pravega-1","homepage":"http://pravega.io","size":25316,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"pravega:master","ref":"master","sha":"aa78d07b7dfcc5baf4345ca6f7c696a9e4673e5a","user":{"login":"pravega","id":25698199,"avatar_url":"https://avatars1.githubusercontent.com/u/25698199?v=4","gravatar_id":"","url":"https://api.github.com/users/pravega","html_url":"https://github.com/pravega","followers_url":"https://api.github.com/users/pravega/followers","following_url":"https://api.github.com/users/pravega/following{/other_user}","gists_url":"https://api.github.com/users/pravega/gists{/gist_id}","starred_url":"https://api.github.com/users/pravega/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pravega/subscriptions","organizations_url":"https://api.github.com/users/pravega/orgs","repos_url":"https://api.github.com/users/pravega/repos","events_url":"https://api.github.com/users/pravega/events{/privacy}","received_events_url":"https://api.github.com/users/pravega/received_events","type":"Organization","site_admin":false},"repo":{"id":63096822,"name":"pravega","full_name":"pravega/pravega","owner":{"login":"pravega","id":25698199,"avatar_url":"https://avatars1.githubusercontent.com/u/25698199?v=4","gravatar_id":"","url":"https://api.github.com/users/pravega","html_url":"https://github.com/pravega","followers_url":"https://api.github.com/users/pravega/followers","following_url":"https://api.github.com/users/pravega/following{/other_user}","gists_url":"https://api.github.com/users/pravega/gists{/gist_id}","starred_url":"https://api.github.com/users/pravega/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pravega/subscriptions","organizations_url":"https://api.github.com/users/pravega/orgs","repos_url":"https://api.github.com/users/pravega/repos","events_url":"https://api.github.com/users/pravega/events{/privacy}","received_events_url":"https://api.github.com/users/pravega/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/pravega/pravega","description":"Pravega - Streaming as a new software defined storage primitive","fork":false,"url":"https://api.github.com/repos/pravega/pravega","forks_url":"https://api.github.com/repos/pravega/pravega/forks","keys_url":"https://api.github.com/repos/pravega/pravega/keys{/key_id}","collaborators_url":"https://api.github.com/repos/pravega/pravega/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/pravega/pravega/teams","hooks_url":"https://api.github.com/repos/pravega/pravega/hooks","issue_events_url":"https://api.github.com/repos/pravega/pravega/issues/events{/number}","events_url":"https://api.github.com/repos/pravega/pravega/events","assignees_url":"https://api.github.com/repos/pravega/pravega/assignees{/user}","branches_url":"https://api.github.com/repos/pravega/pravega/branches{/branch}","tags_url":"https://api.github.com/repos/pravega/pravega/tags","blobs_url":"https://api.github.com/repos/pravega/pravega/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/pravega/pravega/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/pravega/pravega/git/refs{/sha}","trees_url":"https://api.github.com/repos/pravega/pravega/git/trees{/sha}","statuses_url":"https://api.github.com/repos/pravega/pravega/statuses/{sha}","languages_url":"https://api.github.com/repos/pravega/pravega/languages","stargazers_url":"https://api.github.com/repos/pravega/pravega/stargazers","contributors_url":"https://api.github.com/repos/pravega/pravega/contributors","subscribers_url":"https://api.github.com/repos/pravega/pravega/subscribers","subscription_url":"https://api.github.com/repos/pravega/pravega/subscription","commits_url":"https://api.github.com/repos/pravega/pravega/commits{/sha}","git_commits_url":"https://api.github.com/repos/pravega/pravega/git/commits{/sha}","comments_url":"https://api.github.com/repos/pravega/pravega/comments{/number}","issue_comment_url":"https://api.github.com/repos/pravega/pravega/issues/comments{/number}","contents_url":"https://api.github.com/repos/pravega/pravega/contents/{+path}","compare_url":"https://api.github.com/repos/pravega/pravega/compare/{base}...{head}","merges_url":"https://api.github.com/repos/pravega/pravega/merges","archive_url":"https://api.github.com/repos/pravega/pravega/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/pravega/pravega/downloads","issues_url":"https://api.github.com/repos/pravega/pravega/issues{/number}","pulls_url":"https://api.github.com/repos/pravega/pravega/pulls{/number}","milestones_url":"https://api.github.com/repos/pravega/pravega/milestones{/number}","notifications_url":"https://api.github.com/repos/pravega/pravega/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/pravega/pravega/labels{/name}","releases_url":"https://api.github.com/repos/pravega/pravega/releases{/id}","deployments_url":"https://api.github.com/repos/pravega/pravega/deployments","created_at":"2016-07-11T19:41:03Z","updated_at":"2018-05-22T22:34:33Z","pushed_at":"2018-05-23T04:56:13Z","git_url":"git://github.com/pravega/pravega.git","ssh_url":"git@github.com:pravega/pravega.git","clone_url":"https://github.com/pravega/pravega.git","svn_url":"https://github.com/pravega/pravega","homepage":"http://pravega.io","size":25137,"stargazers_count":210,"watchers_count":210,"language":"Java","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":71,"mirror_url":null,"archived":false,"open_issues_count":378,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":71,"open_issues":378,"watchers":210,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/pravega/pravega/pulls/2611"},"html":{"href":"https://github.com/pravega/pravega/pull/2611"},"issue":{"href":"https://api.github.com/repos/pravega/pravega/issues/2611"},"comments":{"href":"https://api.github.com/repos/pravega/pravega/issues/2611/comments"},"review_comments":{"href":"https://api.github.com/repos/pravega/pravega/pulls/2611/comments"},"review_comment":{"href":"https://api.github.com/repos/pravega/pravega/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/pravega/pravega/pulls/2611/commits"},"statuses":{"href":"https://api.github.com/repos/pravega/pravega/statuses/a615c25fca4999a9911db6257a93b3c81abd6162"}},"author_association":"COLLABORATOR"}}
{"actor":{"display_login":"shiveshr"}}