{"action":"created","comment":{"url":"https://api.github.com/repos/apache/storm/pulls/comments/37024551","id":37024551,"diff_hunk":"@@ -69,63 +78,92 @@\n \n         abstract void doPrepare(Map conf, int partitionIndex, int numPartitions) throws IOException;\n \n-        protected void rotateOutputFile() throws IOException {\n+        abstract long getCurrentOffset() throws  IOException;\n+\n+        abstract void doCommit(Long txId) throws IOException;\n+\n+        abstract void doRecover(Path srcPath, long nBytes) throws Exception;\n+\n+        protected void rotateOutputFile(boolean doRotateAction) throws IOException {\n             LOG.info(\"Rotating output file...\");\n             long start = System.currentTimeMillis();\n-            synchronized (this.writeLock) {\n-                closeOutputFile();\n-                this.rotation++;\n-\n-                Path newFile = createOutputFile();\n+            closeOutputFile();\n+            this.rotation++;\n+            Path newFile = createOutputFile();\n+            if (doRotateAction) {\n                 LOG.info(\"Performing {} file rotation actions.\", this.rotationActions.size());\n                 for (RotationAction action : this.rotationActions) {\n                     action.execute(this.fs, this.currentFile);\n                 }\n-                this.currentFile = newFile;\n             }\n+            this.currentFile = newFile;\n             long time = System.currentTimeMillis() - start;\n             LOG.info(\"File rotation took {} ms.\", time);\n+        }\n \n-\n+        protected void rotateOutputFile() throws IOException {\n+            rotateOutputFile(true);\n         }\n \n-        void prepare(Map conf, int partitionIndex, int numPartitions){\n-            this.writeLock = new Object();\n-            if (this.rotationPolicy == null) throw new IllegalStateException(\"RotationPolicy must be specified.\");\n+\n+        void prepare(Map conf, int partitionIndex, int numPartitions) {\n+            if (this.rotationPolicy == null) {\n+                throw new IllegalStateException(\"RotationPolicy must be specified.\");\n+            } else if (this.rotationPolicy instanceof FileSizeRotationPolicy) {\n+                long rotationBytes = ((FileSizeRotationPolicy) rotationPolicy).getMaxBytes();\n+                LOG.warn(\"FileSizeRotationPolicy specified with {} bytes.\", rotationBytes);\n+                LOG.warn(\"Recovery will fail if data files cannot be copied within topology.message.timeout.secs.\");\n+                LOG.warn(\"Ensure that the data files does not grow too big with the FileSizeRotationPolicy.\");\n+            } else if (this.rotationPolicy instanceof TimedRotationPolicy) {\n+                LOG.warn(\"TimedRotationPolicy specified with interval {} ms.\", ((TimedRotationPolicy) rotationPolicy).getInterval());\n+                LOG.warn(\"Recovery will fail if data files cannot be copied within topology.message.timeout.secs.\");\n+                LOG.warn(\"Ensure that the data files does not grow too big with the TimedRotationPolicy.\");\n+            }","path":"external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java","position":111,"original_position":111,"commit_id":"4d8c51b4503110f03a27b401f096ff2477bbb7cf","original_commit_id":"4d8c51b4503110f03a27b401f096ff2477bbb7cf","user":{"login":"d2r","id":905298,"avatar_url":"https://avatars.githubusercontent.com/u/905298?v=3","gravatar_id":"","url":"https://api.github.com/users/d2r","html_url":"https://github.com/d2r","followers_url":"https://api.github.com/users/d2r/followers","following_url":"https://api.github.com/users/d2r/following{/other_user}","gists_url":"https://api.github.com/users/d2r/gists{/gist_id}","starred_url":"https://api.github.com/users/d2r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/d2r/subscriptions","organizations_url":"https://api.github.com/users/d2r/orgs","repos_url":"https://api.github.com/users/d2r/repos","events_url":"https://api.github.com/users/d2r/events{/privacy}","received_events_url":"https://api.github.com/users/d2r/received_events","type":"User","site_admin":false},"body":"It might be nicer to move this code into the subclasses own methods: FileSizeRotationPolicy#prepare and TimedRotationPolicy#prepare.  These prepare methods can each call `super.prepare(conf, partitionIndex, numPartitions)`.","created_at":"2015-08-13T20:58:44Z","updated_at":"2015-08-13T20:58:44Z","html_url":"https://github.com/apache/storm/pull/644#discussion_r37024551","pull_request_url":"https://api.github.com/repos/apache/storm/pulls/644","_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/comments/37024551"},"html":{"href":"https://github.com/apache/storm/pull/644#discussion_r37024551"},"pull_request":{"href":"https://api.github.com/repos/apache/storm/pulls/644"}}},"pull_request":{"url":"https://api.github.com/repos/apache/storm/pulls/644","id":40495108,"html_url":"https://github.com/apache/storm/pull/644","diff_url":"https://github.com/apache/storm/pull/644.diff","patch_url":"https://github.com/apache/storm/pull/644.patch","issue_url":"https://api.github.com/repos/apache/storm/issues/644","number":644,"state":"open","locked":false,"title":"[STORM-837] Support for exactly once semantics in HdfsState","user":{"login":"arunmahadevan","id":6792890,"avatar_url":"https://avatars.githubusercontent.com/u/6792890?v=3","gravatar_id":"","url":"https://api.github.com/users/arunmahadevan","html_url":"https://github.com/arunmahadevan","followers_url":"https://api.github.com/users/arunmahadevan/followers","following_url":"https://api.github.com/users/arunmahadevan/following{/other_user}","gists_url":"https://api.github.com/users/arunmahadevan/gists{/gist_id}","starred_url":"https://api.github.com/users/arunmahadevan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/arunmahadevan/subscriptions","organizations_url":"https://api.github.com/users/arunmahadevan/orgs","repos_url":"https://api.github.com/users/arunmahadevan/repos","events_url":"https://api.github.com/users/arunmahadevan/events{/privacy}","received_events_url":"https://api.github.com/users/arunmahadevan/received_events","type":"User","site_admin":false},"body":"Changes to support exactly once semantics in HdfsState.\r\n\r\n1. Moved the file rotation and sync to commit()\r\n2. In pre-commit, if we have previously seen the txnid, recover the data up to that point by copying to a new file and discard the current data file.\r\n3. In pre-commit atomically update [current txid, the datafile path and the current offset] in a (per partition) index file.\r\n4. To keep it simple, automatically turn off exactly once semantics if TimedRotation policy is in use.\r\n\r\nHave tested with the normal flow and simulating the recovery scenario, with both regular Hdfs file and sequence files. Appears to work fine.","created_at":"2015-07-21T19:30:22Z","updated_at":"2015-08-13T20:58:44Z","closed_at":null,"merged_at":null,"merge_commit_sha":"32d51e4aa69a4ba085c686879e772539f65fb669","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/storm/pulls/644/commits","review_comments_url":"https://api.github.com/repos/apache/storm/pulls/644/comments","review_comment_url":"https://api.github.com/repos/apache/storm/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/storm/issues/644/comments","statuses_url":"https://api.github.com/repos/apache/storm/statuses/4d8c51b4503110f03a27b401f096ff2477bbb7cf","head":{"label":"arunmahadevan:master","ref":"master","sha":"4d8c51b4503110f03a27b401f096ff2477bbb7cf","user":{"login":"arunmahadevan","id":6792890,"avatar_url":"https://avatars.githubusercontent.com/u/6792890?v=3","gravatar_id":"","url":"https://api.github.com/users/arunmahadevan","html_url":"https://github.com/arunmahadevan","followers_url":"https://api.github.com/users/arunmahadevan/followers","following_url":"https://api.github.com/users/arunmahadevan/following{/other_user}","gists_url":"https://api.github.com/users/arunmahadevan/gists{/gist_id}","starred_url":"https://api.github.com/users/arunmahadevan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/arunmahadevan/subscriptions","organizations_url":"https://api.github.com/users/arunmahadevan/orgs","repos_url":"https://api.github.com/users/arunmahadevan/repos","events_url":"https://api.github.com/users/arunmahadevan/events{/privacy}","received_events_url":"https://api.github.com/users/arunmahadevan/received_events","type":"User","site_admin":false},"repo":{"id":38767161,"name":"storm","full_name":"arunmahadevan/storm","owner":{"login":"arunmahadevan","id":6792890,"avatar_url":"https://avatars.githubusercontent.com/u/6792890?v=3","gravatar_id":"","url":"https://api.github.com/users/arunmahadevan","html_url":"https://github.com/arunmahadevan","followers_url":"https://api.github.com/users/arunmahadevan/followers","following_url":"https://api.github.com/users/arunmahadevan/following{/other_user}","gists_url":"https://api.github.com/users/arunmahadevan/gists{/gist_id}","starred_url":"https://api.github.com/users/arunmahadevan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/arunmahadevan/subscriptions","organizations_url":"https://api.github.com/users/arunmahadevan/orgs","repos_url":"https://api.github.com/users/arunmahadevan/repos","events_url":"https://api.github.com/users/arunmahadevan/events{/privacy}","received_events_url":"https://api.github.com/users/arunmahadevan/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/arunmahadevan/storm","description":"Mirror of Apache Storm","fork":true,"url":"https://api.github.com/repos/arunmahadevan/storm","forks_url":"https://api.github.com/repos/arunmahadevan/storm/forks","keys_url":"https://api.github.com/repos/arunmahadevan/storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/arunmahadevan/storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/arunmahadevan/storm/teams","hooks_url":"https://api.github.com/repos/arunmahadevan/storm/hooks","issue_events_url":"https://api.github.com/repos/arunmahadevan/storm/issues/events{/number}","events_url":"https://api.github.com/repos/arunmahadevan/storm/events","assignees_url":"https://api.github.com/repos/arunmahadevan/storm/assignees{/user}","branches_url":"https://api.github.com/repos/arunmahadevan/storm/branches{/branch}","tags_url":"https://api.github.com/repos/arunmahadevan/storm/tags","blobs_url":"https://api.github.com/repos/arunmahadevan/storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/arunmahadevan/storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/arunmahadevan/storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/arunmahadevan/storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/arunmahadevan/storm/statuses/{sha}","languages_url":"https://api.github.com/repos/arunmahadevan/storm/languages","stargazers_url":"https://api.github.com/repos/arunmahadevan/storm/stargazers","contributors_url":"https://api.github.com/repos/arunmahadevan/storm/contributors","subscribers_url":"https://api.github.com/repos/arunmahadevan/storm/subscribers","subscription_url":"https://api.github.com/repos/arunmahadevan/storm/subscription","commits_url":"https://api.github.com/repos/arunmahadevan/storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/arunmahadevan/storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/arunmahadevan/storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/arunmahadevan/storm/issues/comments{/number}","contents_url":"https://api.github.com/repos/arunmahadevan/storm/contents/{+path}","compare_url":"https://api.github.com/repos/arunmahadevan/storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/arunmahadevan/storm/merges","archive_url":"https://api.github.com/repos/arunmahadevan/storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/arunmahadevan/storm/downloads","issues_url":"https://api.github.com/repos/arunmahadevan/storm/issues{/number}","pulls_url":"https://api.github.com/repos/arunmahadevan/storm/pulls{/number}","milestones_url":"https://api.github.com/repos/arunmahadevan/storm/milestones{/number}","notifications_url":"https://api.github.com/repos/arunmahadevan/storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/arunmahadevan/storm/labels{/name}","releases_url":"https://api.github.com/repos/arunmahadevan/storm/releases{/id}","created_at":"2015-07-08T16:48:04Z","updated_at":"2015-07-08T16:48:07Z","pushed_at":"2015-08-11T06:56:58Z","git_url":"git://github.com/arunmahadevan/storm.git","ssh_url":"git@github.com:arunmahadevan/storm.git","clone_url":"https://github.com/arunmahadevan/storm.git","svn_url":"https://github.com/arunmahadevan/storm","homepage":null,"size":21495,"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":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"6de597a6d004e56d1889df9b6a913c7391aff126","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-08-13T16:16:58Z","pushed_at":"2015-08-13T20:29:46Z","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":850989,"stargazers_count":2240,"watchers_count":2240,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":1574,"mirror_url":"git://git.apache.org/incubator-storm.git","open_issues_count":83,"forks":1574,"open_issues":83,"watchers":2240,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/644"},"html":{"href":"https://github.com/apache/storm/pull/644"},"issue":{"href":"https://api.github.com/repos/apache/storm/issues/644"},"comments":{"href":"https://api.github.com/repos/apache/storm/issues/644/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/storm/pulls/644/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/storm/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/storm/pulls/644/commits"},"statuses":{"href":"https://api.github.com/repos/apache/storm/statuses/4d8c51b4503110f03a27b401f096ff2477bbb7cf"}}}}