GitHub PullRequestReviewCommentEvent at 2014-11-27 17:25:21
Repository: Malhar
Actor: chandnisingh
Payload: {"action":"created","comment":{"url":"https://api.github.com/repos/DataTorrent/Malhar/pulls/comments/21006034","id":21006034,"diff_hunk":"@@ -0,0 +1,276 @@\n+/*\n+ * Copyright (c) 2014 DataTorrent, Inc. 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+ * 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 com.datatorrent.lib.io.jms;\n+\n+import com.datatorrent.api.annotation.Stateless;\n+import java.io.IOException;\n+import javax.jms.JMSException;\n+import javax.validation.constraints.NotNull;\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.fs.*;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+/**\n+ * This is a JMS store which stores committed window ids in a file. This is not a true\n+ * transactionable store because there is a chance that a failure may occur in between storing the\n+ * committed window and committing the JMS transaction. A failure in such a scenario could cause JMS Messages\n+ * to be missed or duplicated. However, the chance of this is very small. If you need 100% reliability use\n+ * the {@link JMSTransactionableStore}.\n+ */\n+public class FSPsuedoTransactionableStore extends JMSBaseTransactionableStore\n+{\n+  private static final Logger logger = LoggerFactory.getLogger(FSPsuedoTransactionableStore.class);\n+\n+  /**\n+   * The default directory in which recovery occurs.\n+   */\n+  public static final String DEFAULT_RECOVERY_DIRECTORY = \"recovery\";\n+  /**\n+   * The name of the committed window file.\n+   */\n+  public static final String COMMITTED_WINDOW_FILE = \"DT_CMT\";\n+  /**\n+   * The temporary file extension for committed windows.\n+   */\n+  public static final String TEMP_FILE_EXTENSION = \".tmp\";\n+\n+  /**\n+   * Indicates whether the store is connected or not.\n+   */\n+  private transient boolean connected = false;\n+  /**\n+   * Indicates whether the store is in a transaction or not.\n+   */\n+  private transient boolean inTransaction = false;\n+\n+  /**\n+   * The path of the directory to where files are written.\n+   */\n+  @NotNull\n+  protected String recoveryDirectory = DEFAULT_RECOVERY_DIRECTORY;\n+\n+  /**\n+   * File system for storing committed window Ids.\n+   */\n+  private transient FileSystem fs;\n+\n+  public FSPsuedoTransactionableStore()\n+  {\n+  }\n+\n+  /**\n+   * Override this method to change the FileSystem instance that is used by the operator.\n+   * This method is mainly helpful for unit testing.\n+   * @return A FileSystem object.\n+   * @throws IOException\n+   */\n+  protected FileSystem getFSInstance() throws IOException\n+  {\n+    FileSystem tempFS = FileSystem.newInstance(new Path(recoveryDirectory).toUri(), new Configuration());\n+\n+    if(tempFS instanceof LocalFileSystem)\n+    {\n+      tempFS = ((LocalFileSystem) tempFS).getRaw();\n+    }\n+\n+    return tempFS;\n+  }\n+\n+  /**\n+   * Sets the directory path in which committed windows are stored.\n+   * @param recoveryDirectory The directory path in which committed windows are stored.\n+   */\n+  public void setRecoveryDirectory(String recoveryDirectory)\n+  {\n+    this.recoveryDirectory = recoveryDirectory;\n+  }\n+\n+  /**\n+   * Gets the directory path in which committed windows are stored.\n+   * @return The directory path in which committed windows are stored.\n+   */\n+  public String getRecoveryDirectory()\n+  {\n+    return recoveryDirectory;\n+  }\n+\n+  @Override\n+  public long getCommittedWindowId(String appId, int operatorId)\n+  {\n+    Path path = getOperatorRecoveryPath(appId, operatorId);\n+\n+    if(logger.isDebugEnabled()) {\n+      logger.debug(\"Working directory path {}\", fs.getWorkingDirectory());\n+      logger.debug(\"Recovery directory path {}\", path);\n+    }\n+\n+    try {\n+      //No committed window stored, return negative invalid window.\n+      if(!fs.exists(path))\n+      {\n+        return Stateless.WINDOW_ID;\n+      }\n+    }\n+    catch (IOException ex) {\n+      throw new RuntimeException(ex);\n+    }\n+\n+    try {\n+      //Read committed winodw from file.\n+      FSDataInputStream inputStream = fs.open(path);\n+      return inputStream.readLong();\n+    }\n+    catch (IOException ex) {\n+      throw new RuntimeException(ex);\n+    }\n+  }\n+\n+  @Override\n+  public void storeCommittedWindowId(String appId, int operatorId, long windowId)\n+  {\n+    Path path = getOperatorRecoveryPath(appId, operatorId);\n+    Path pathTMP = getOperatorRecoveryTMPPath(appId, operatorId);\n+\n+    try {\n+      FSDataOutputStream output = fs.create(pathTMP, true);","path":"library/src/main/java/com/datatorrent/lib/io/jms/FSPsuedoTransactionableStore.java","position":149,"original_position":149,"commit_id":"5d8e92d31ddeefb2fb23b58c62a179717a717596","original_commit_id":"5d8e92d31ddeefb2fb23b58c62a179717a717596","user":{"login":"chandnisingh","id":4690923,"avatar_url":"https://avatars.githubusercontent.com/u/4690923?v=3","gravatar_id":"","url":"https://api.github.com/users/chandnisingh","html_url":"https://github.com/chandnisingh","followers_url":"https://api.github.com/users/chandnisingh/followers","following_url":"https://api.github.com/users/chandnisingh/following{/other_user}","gists_url":"https://api.github.com/users/chandnisingh/gists{/gist_id}","starred_url":"https://api.github.com/users/chandnisingh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chandnisingh/subscriptions","organizations_url":"https://api.github.com/users/chandnisingh/orgs","repos_url":"https://api.github.com/users/chandnisingh/repos","events_url":"https://api.github.com/users/chandnisingh/events{/privacy}","received_events_url":"https://api.github.com/users/chandnisingh/received_events","type":"User","site_admin":false},"body":"I think it is better to create empty committed files with name as the window id. That can avoid opening the stream for reading/writing.  ","created_at":"2014-11-27T17:25:21Z","updated_at":"2014-11-27T17:25:21Z","html_url":"https://github.com/DataTorrent/Malhar/pull/1179#discussion_r21006034","pull_request_url":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179","_links":{"self":{"href":"https://api.github.com/repos/DataTorrent/Malhar/pulls/comments/21006034"},"html":{"href":"https://github.com/DataTorrent/Malhar/pull/1179#discussion_r21006034"},"pull_request":{"href":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179"}}},"pull_request":{"url":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179","id":25108281,"html_url":"https://github.com/DataTorrent/Malhar/pull/1179","diff_url":"https://github.com/DataTorrent/Malhar/pull/1179.diff","patch_url":"https://github.com/DataTorrent/Malhar/pull/1179.patch","issue_url":"https://api.github.com/repos/DataTorrent/Malhar/issues/1179","number":1179,"state":"open","locked":false,"title":"   * Made JMS output operator fault tolerant","user":{"login":"ilooner","id":8020165,"avatar_url":"https://avatars.githubusercontent.com/u/8020165?v=3","gravatar_id":"","url":"https://api.github.com/users/ilooner","html_url":"https://github.com/ilooner","followers_url":"https://api.github.com/users/ilooner/followers","following_url":"https://api.github.com/users/ilooner/following{/other_user}","gists_url":"https://api.github.com/users/ilooner/gists{/gist_id}","starred_url":"https://api.github.com/users/ilooner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilooner/subscriptions","organizations_url":"https://api.github.com/users/ilooner/orgs","repos_url":"https://api.github.com/users/ilooner/repos","events_url":"https://api.github.com/users/ilooner/events{/privacy}","received_events_url":"https://api.github.com/users/ilooner/received_events","type":"User","site_admin":false},"body":"   * JMSTransactionableStore is used for exactly once output\r\n   * FSPsuedoTransactionableStore should be used for Atleast Once and At most once output.","created_at":"2014-11-26T19:29:34Z","updated_at":"2014-11-27T17:25:21Z","closed_at":null,"merged_at":null,"merge_commit_sha":"811b6944a884cb998b535ff1337fedd33534e2ed","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179/commits","review_comments_url":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179/comments","review_comment_url":"https://api.github.com/repos/DataTorrent/Malhar/pulls/comments/{number}","comments_url":"https://api.github.com/repos/DataTorrent/Malhar/issues/1179/comments","statuses_url":"https://api.github.com/repos/DataTorrent/Malhar/statuses/5d8e92d31ddeefb2fb23b58c62a179717a717596","head":{"label":"ilooner:MLHR-1546_pull2","ref":"MLHR-1546_pull2","sha":"5d8e92d31ddeefb2fb23b58c62a179717a717596","user":{"login":"ilooner","id":8020165,"avatar_url":"https://avatars.githubusercontent.com/u/8020165?v=3","gravatar_id":"","url":"https://api.github.com/users/ilooner","html_url":"https://github.com/ilooner","followers_url":"https://api.github.com/users/ilooner/followers","following_url":"https://api.github.com/users/ilooner/following{/other_user}","gists_url":"https://api.github.com/users/ilooner/gists{/gist_id}","starred_url":"https://api.github.com/users/ilooner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilooner/subscriptions","organizations_url":"https://api.github.com/users/ilooner/orgs","repos_url":"https://api.github.com/users/ilooner/repos","events_url":"https://api.github.com/users/ilooner/events{/privacy}","received_events_url":"https://api.github.com/users/ilooner/received_events","type":"User","site_admin":false},"repo":{"id":23372785,"name":"Malhar","full_name":"ilooner/Malhar","owner":{"login":"ilooner","id":8020165,"avatar_url":"https://avatars.githubusercontent.com/u/8020165?v=3","gravatar_id":"","url":"https://api.github.com/users/ilooner","html_url":"https://github.com/ilooner","followers_url":"https://api.github.com/users/ilooner/followers","following_url":"https://api.github.com/users/ilooner/following{/other_user}","gists_url":"https://api.github.com/users/ilooner/gists{/gist_id}","starred_url":"https://api.github.com/users/ilooner/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilooner/subscriptions","organizations_url":"https://api.github.com/users/ilooner/orgs","repos_url":"https://api.github.com/users/ilooner/repos","events_url":"https://api.github.com/users/ilooner/events{/privacy}","received_events_url":"https://api.github.com/users/ilooner/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/ilooner/Malhar","description":"DataTorrent open source code.","fork":true,"url":"https://api.github.com/repos/ilooner/Malhar","forks_url":"https://api.github.com/repos/ilooner/Malhar/forks","keys_url":"https://api.github.com/repos/ilooner/Malhar/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ilooner/Malhar/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ilooner/Malhar/teams","hooks_url":"https://api.github.com/repos/ilooner/Malhar/hooks","issue_events_url":"https://api.github.com/repos/ilooner/Malhar/issues/events{/number}","events_url":"https://api.github.com/repos/ilooner/Malhar/events","assignees_url":"https://api.github.com/repos/ilooner/Malhar/assignees{/user}","branches_url":"https://api.github.com/repos/ilooner/Malhar/branches{/branch}","tags_url":"https://api.github.com/repos/ilooner/Malhar/tags","blobs_url":"https://api.github.com/repos/ilooner/Malhar/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ilooner/Malhar/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ilooner/Malhar/git/refs{/sha}","trees_url":"https://api.github.com/repos/ilooner/Malhar/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ilooner/Malhar/statuses/{sha}","languages_url":"https://api.github.com/repos/ilooner/Malhar/languages","stargazers_url":"https://api.github.com/repos/ilooner/Malhar/stargazers","contributors_url":"https://api.github.com/repos/ilooner/Malhar/contributors","subscribers_url":"https://api.github.com/repos/ilooner/Malhar/subscribers","subscription_url":"https://api.github.com/repos/ilooner/Malhar/subscription","commits_url":"https://api.github.com/repos/ilooner/Malhar/commits{/sha}","git_commits_url":"https://api.github.com/repos/ilooner/Malhar/git/commits{/sha}","comments_url":"https://api.github.com/repos/ilooner/Malhar/comments{/number}","issue_comment_url":"https://api.github.com/repos/ilooner/Malhar/issues/comments/{number}","contents_url":"https://api.github.com/repos/ilooner/Malhar/contents/{+path}","compare_url":"https://api.github.com/repos/ilooner/Malhar/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ilooner/Malhar/merges","archive_url":"https://api.github.com/repos/ilooner/Malhar/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ilooner/Malhar/downloads","issues_url":"https://api.github.com/repos/ilooner/Malhar/issues{/number}","pulls_url":"https://api.github.com/repos/ilooner/Malhar/pulls{/number}","milestones_url":"https://api.github.com/repos/ilooner/Malhar/milestones{/number}","notifications_url":"https://api.github.com/repos/ilooner/Malhar/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ilooner/Malhar/labels{/name}","releases_url":"https://api.github.com/repos/ilooner/Malhar/releases{/id}","created_at":"2014-08-27T01:32:02Z","updated_at":"2014-07-15T18:55:01Z","pushed_at":"2014-11-27T06:49:00Z","git_url":"git://github.com/ilooner/Malhar.git","ssh_url":"git@github.com:ilooner/Malhar.git","clone_url":"https://github.com/ilooner/Malhar.git","svn_url":"https://github.com/ilooner/Malhar","homepage":null,"size":39244,"stargazers_count":0,"watchers_count":0,"language":null,"has_issues":false,"has_downloads":true,"has_wiki":true,"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":"DataTorrent:dt-dev","ref":"dt-dev","sha":"e3f2bef14ca8516092ea070dae4bf67aa9023174","user":{"login":"DataTorrent","id":4400063,"avatar_url":"https://avatars.githubusercontent.com/u/4400063?v=3","gravatar_id":"","url":"https://api.github.com/users/DataTorrent","html_url":"https://github.com/DataTorrent","followers_url":"https://api.github.com/users/DataTorrent/followers","following_url":"https://api.github.com/users/DataTorrent/following{/other_user}","gists_url":"https://api.github.com/users/DataTorrent/gists{/gist_id}","starred_url":"https://api.github.com/users/DataTorrent/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/DataTorrent/subscriptions","organizations_url":"https://api.github.com/users/DataTorrent/orgs","repos_url":"https://api.github.com/users/DataTorrent/repos","events_url":"https://api.github.com/users/DataTorrent/events{/privacy}","received_events_url":"https://api.github.com/users/DataTorrent/received_events","type":"Organization","site_admin":false},"repo":{"id":11749698,"name":"Malhar","full_name":"DataTorrent/Malhar","owner":{"login":"DataTorrent","id":4400063,"avatar_url":"https://avatars.githubusercontent.com/u/4400063?v=3","gravatar_id":"","url":"https://api.github.com/users/DataTorrent","html_url":"https://github.com/DataTorrent","followers_url":"https://api.github.com/users/DataTorrent/followers","following_url":"https://api.github.com/users/DataTorrent/following{/other_user}","gists_url":"https://api.github.com/users/DataTorrent/gists{/gist_id}","starred_url":"https://api.github.com/users/DataTorrent/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/DataTorrent/subscriptions","organizations_url":"https://api.github.com/users/DataTorrent/orgs","repos_url":"https://api.github.com/users/DataTorrent/repos","events_url":"https://api.github.com/users/DataTorrent/events{/privacy}","received_events_url":"https://api.github.com/users/DataTorrent/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/DataTorrent/Malhar","description":"DataTorrent open source code.","fork":false,"url":"https://api.github.com/repos/DataTorrent/Malhar","forks_url":"https://api.github.com/repos/DataTorrent/Malhar/forks","keys_url":"https://api.github.com/repos/DataTorrent/Malhar/keys{/key_id}","collaborators_url":"https://api.github.com/repos/DataTorrent/Malhar/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/DataTorrent/Malhar/teams","hooks_url":"https://api.github.com/repos/DataTorrent/Malhar/hooks","issue_events_url":"https://api.github.com/repos/DataTorrent/Malhar/issues/events{/number}","events_url":"https://api.github.com/repos/DataTorrent/Malhar/events","assignees_url":"https://api.github.com/repos/DataTorrent/Malhar/assignees{/user}","branches_url":"https://api.github.com/repos/DataTorrent/Malhar/branches{/branch}","tags_url":"https://api.github.com/repos/DataTorrent/Malhar/tags","blobs_url":"https://api.github.com/repos/DataTorrent/Malhar/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/DataTorrent/Malhar/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/DataTorrent/Malhar/git/refs{/sha}","trees_url":"https://api.github.com/repos/DataTorrent/Malhar/git/trees{/sha}","statuses_url":"https://api.github.com/repos/DataTorrent/Malhar/statuses/{sha}","languages_url":"https://api.github.com/repos/DataTorrent/Malhar/languages","stargazers_url":"https://api.github.com/repos/DataTorrent/Malhar/stargazers","contributors_url":"https://api.github.com/repos/DataTorrent/Malhar/contributors","subscribers_url":"https://api.github.com/repos/DataTorrent/Malhar/subscribers","subscription_url":"https://api.github.com/repos/DataTorrent/Malhar/subscription","commits_url":"https://api.github.com/repos/DataTorrent/Malhar/commits{/sha}","git_commits_url":"https://api.github.com/repos/DataTorrent/Malhar/git/commits{/sha}","comments_url":"https://api.github.com/repos/DataTorrent/Malhar/comments{/number}","issue_comment_url":"https://api.github.com/repos/DataTorrent/Malhar/issues/comments/{number}","contents_url":"https://api.github.com/repos/DataTorrent/Malhar/contents/{+path}","compare_url":"https://api.github.com/repos/DataTorrent/Malhar/compare/{base}...{head}","merges_url":"https://api.github.com/repos/DataTorrent/Malhar/merges","archive_url":"https://api.github.com/repos/DataTorrent/Malhar/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/DataTorrent/Malhar/downloads","issues_url":"https://api.github.com/repos/DataTorrent/Malhar/issues{/number}","pulls_url":"https://api.github.com/repos/DataTorrent/Malhar/pulls{/number}","milestones_url":"https://api.github.com/repos/DataTorrent/Malhar/milestones{/number}","notifications_url":"https://api.github.com/repos/DataTorrent/Malhar/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/DataTorrent/Malhar/labels{/name}","releases_url":"https://api.github.com/repos/DataTorrent/Malhar/releases{/id}","created_at":"2013-07-29T21:00:05Z","updated_at":"2014-11-27T10:31:24Z","pushed_at":"2014-11-27T04:19:34Z","git_url":"git://github.com/DataTorrent/Malhar.git","ssh_url":"git@github.com:DataTorrent/Malhar.git","clone_url":"https://github.com/DataTorrent/Malhar.git","svn_url":"https://github.com/DataTorrent/Malhar","homepage":null,"size":113333,"stargazers_count":44,"watchers_count":44,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":48,"mirror_url":null,"open_issues_count":137,"forks":48,"open_issues":137,"watchers":44,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179"},"html":{"href":"https://github.com/DataTorrent/Malhar/pull/1179"},"issue":{"href":"https://api.github.com/repos/DataTorrent/Malhar/issues/1179"},"comments":{"href":"https://api.github.com/repos/DataTorrent/Malhar/issues/1179/comments"},"review_comments":{"href":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179/comments"},"review_comment":{"href":"https://api.github.com/repos/DataTorrent/Malhar/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/DataTorrent/Malhar/pulls/1179/commits"},"statuses":{"href":"https://api.github.com/repos/DataTorrent/Malhar/statuses/5d8e92d31ddeefb2fb23b58c62a179717a717596"}}}}