{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/30286375","id":30286375,"diff_hunk":"@@ -0,0 +1,175 @@\n+/* (c) 2015 NerdWallet All rights reserved.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\"); you may not use\n+ * this file except in compliance with the License. You may obtain a copy of the\n+ * License at  http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software distributed\n+ * under the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR\n+ * CONDITIONS OF ANY KIND, either express or implied.\n+ */\n+\n+package gobblin.writer;\n+\n+import com.google.common.base.Preconditions;\n+\n+import gobblin.configuration.ConfigurationKeys;\n+import gobblin.configuration.State;\n+import gobblin.util.ForkOperatorUtils;\n+\n+import org.apache.commons.lang.ArrayUtils;\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.fs.FSDataOutputStream;\n+import org.apache.hadoop.fs.FileSystem;\n+import org.apache.hadoop.fs.Path;\n+\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.io.IOException;\n+import java.net.URI;\n+import java.nio.ByteBuffer;\n+import java.util.Arrays;\n+\n+/**\n+ * An implementation of {@link DataWriter} that writes bytes directly to HDFS.\n+ *\n+ * @author akshay@nerdwallet.com\n+ */\n+public class SimpleDataWriter extends BaseDataWriter<byte[]> {\n+\n+  private static final Logger LOG = LoggerFactory.getLogger(SimpleDataWriter.class);\n+\n+  private final FSDataOutputStream outputStream; // the output stream to the staging file\n+  private final Byte recordDelimiter; // optional byte to place between each record write\n+  private final boolean prependSize;\n+\n+  private int recordsWritten;\n+  private int bytesWritten;\n+  private boolean closed;\n+\n+  public SimpleDataWriter(State properties, String fileName, int numBranches, int branchId) throws IOException {\n+    super(properties, fileName, numBranches, branchId);\n+    String delim;\n+    if ((delim = properties.getProp(ConfigurationKeys.SIMPLE_WRITER_DELIMITER, null)) == null || delim.length() == 0) {\n+      this.recordDelimiter = null;\n+    } else {\n+      this.recordDelimiter = delim.getBytes()[0];\n+    }\n+\n+    this.outputStream = this.fs.create(this.stagingFile, true);\n+    this.prependSize = properties.getPropAsBoolean(ConfigurationKeys.SIMPLE_WRITER_PREPEND_SIZE, true);\n+    this.recordsWritten = 0;\n+    this.bytesWritten = 0;\n+    this.closed = false;\n+  }\n+  /**\n+   * Write a source record to the staging file\n+   *\n+   * @param record data record to write\n+   * @throws java.io.IOException if there is anything wrong writing the record\n+   */\n+  @Override\n+  public void write(byte[] record) throws IOException {\n+    byte[] toWrite = record;\n+    Preconditions.checkNotNull(record);\n+    if (recordDelimiter != null) {\n+      toWrite = Arrays.copyOf(record, record.length + 1);\n+      toWrite[toWrite.length - 1] = recordDelimiter;\n+    }\n+    if (prependSize) {\n+      Long recordSize = new Long(toWrite.length);\n+      ByteBuffer buf = ByteBuffer.allocate(Long.SIZE / 8);\n+      buf.putLong(recordSize);\n+      toWrite = ArrayUtils.addAll(buf.array(), toWrite);\n+    }\n+    this.outputStream.write(toWrite);\n+    bytesWritten += (toWrite.length);\n+    recordsWritten++;\n+  }\n+\n+  /**\n+   * Commit the data written to the final output file.\n+   *\n+   * @throws java.io.IOException if there is anything wrong committing the output\n+   */\n+  @Override\n+  public void commit() throws IOException {\n+    this.close();\n+    if (!this.fs.exists(this.stagingFile)) {\n+      throw new IOException(String.format(\"File %s does not exist\", this.stagingFile));\n+    }\n+\n+    LOG.info(String.format(\"Moving data from %s to %s\", this.stagingFile, this.outputFile));\n+    // For the same reason as deleting the staging file if it already exists, deleting\n+    // the output file if it already exists prevents task retry from being blocked.\n+    if (this.fs.exists(this.outputFile)) {\n+      LOG.warn(String.format(\"Task output file %s already exists\", this.outputFile));\n+      this.deletePath(this.outputFile, false);\n+    }\n+    this.fs.rename(this.stagingFile, this.outputFile);\n+  }\n+\n+  /**\n+   * Cleanup context/resources.\n+   *\n+   * @throws java.io.IOException if there is anything wrong doing cleanup.\n+   */\n+  @Override\n+  public void cleanup() throws IOException {\n+    if (this.fs.exists(this.stagingFile)) {\n+      this.deletePath(this.stagingFile, false);\n+    }\n+  }\n+\n+  /**\n+   * Get the number of records written.\n+   *\n+   * @return number of records written\n+   */\n+  @Override\n+  public long recordsWritten() {\n+    return this.recordsWritten;\n+  }\n+\n+  /**\n+   * Get the number of bytes written.\n+   * <p/>\n+   * <p>\n+   * This method should ONLY be called after {@link gobblin.writer.DataWriter#commit()}\n+   * is called.\n+   * </p>\n+   *\n+   * @return number of bytes written\n+   */\n+  @Override\n+  public long bytesWritten() throws IOException {\n+    if (this.closed) {\n+      return this.bytesWritten;\n+    } else {\n+      return 0;","path":"gobblin-core/src/main/java/gobblin/writer/SimpleDataWriter.java","position":150,"original_position":150,"commit_id":"3d35a60ae124b9b1c4da98f0cdaa4a900f21ecb1","original_commit_id":"3d35a60ae124b9b1c4da98f0cdaa4a900f21ecb1","user":{"login":"zliu41","id":6342538,"avatar_url":"https://avatars.githubusercontent.com/u/6342538?v=3","gravatar_id":"","url":"https://api.github.com/users/zliu41","html_url":"https://github.com/zliu41","followers_url":"https://api.github.com/users/zliu41/followers","following_url":"https://api.github.com/users/zliu41/following{/other_user}","gists_url":"https://api.github.com/users/zliu41/gists{/gist_id}","starred_url":"https://api.github.com/users/zliu41/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zliu41/subscriptions","organizations_url":"https://api.github.com/users/zliu41/orgs","repos_url":"https://api.github.com/users/zliu41/repos","events_url":"https://api.github.com/users/zliu41/events{/privacy}","received_events_url":"https://api.github.com/users/zliu41/received_events","type":"User","site_admin":false},"body":"I see. Then it should be fine for now.","created_at":"2015-05-13T23:29:37Z","updated_at":"2015-05-13T23:29:37Z","html_url":"https://github.com/linkedin/gobblin/pull/123#discussion_r30286375","pull_request_url":"https://api.github.com/repos/linkedin/gobblin/pulls/123","_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/30286375"},"html":{"href":"https://github.com/linkedin/gobblin/pull/123#discussion_r30286375"},"pull_request":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/123"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/123","id":35182646,"html_url":"https://github.com/linkedin/gobblin/pull/123","diff_url":"https://github.com/linkedin/gobblin/pull/123.diff","patch_url":"https://github.com/linkedin/gobblin/pull/123.patch","issue_url":"https://api.github.com/repos/linkedin/gobblin/issues/123","number":123,"state":"open","locked":false,"title":"Feature/simple data writer","user":{"login":"akshayn1107","id":3410061,"avatar_url":"https://avatars.githubusercontent.com/u/3410061?v=3","gravatar_id":"","url":"https://api.github.com/users/akshayn1107","html_url":"https://github.com/akshayn1107","followers_url":"https://api.github.com/users/akshayn1107/followers","following_url":"https://api.github.com/users/akshayn1107/following{/other_user}","gists_url":"https://api.github.com/users/akshayn1107/gists{/gist_id}","starred_url":"https://api.github.com/users/akshayn1107/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akshayn1107/subscriptions","organizations_url":"https://api.github.com/users/akshayn1107/orgs","repos_url":"https://api.github.com/users/akshayn1107/repos","events_url":"https://api.github.com/users/akshayn1107/events{/privacy}","received_events_url":"https://api.github.com/users/akshayn1107/received_events","type":"User","site_admin":false},"body":"A simple data writer which writes records as byte[] to a file. This also has some refactors around some of the common elements between writers. Adds a new configuration option to add the writer's final output path to the task properties. This can be used by a publisher to read the write file for a task.","created_at":"2015-05-12T00:13:08Z","updated_at":"2015-05-13T23:29:37Z","closed_at":null,"merged_at":null,"merge_commit_sha":"743bd8341cb357e8484c208276b2a1fe46e5605a","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/gobblin/pulls/123/commits","review_comments_url":"https://api.github.com/repos/linkedin/gobblin/pulls/123/comments","review_comment_url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/gobblin/issues/123/comments","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/3d35a60ae124b9b1c4da98f0cdaa4a900f21ecb1","head":{"label":"NerdWallet:feature/simple_data_writer","ref":"feature/simple_data_writer","sha":"3d35a60ae124b9b1c4da98f0cdaa4a900f21ecb1","user":{"login":"NerdWallet","id":5913289,"avatar_url":"https://avatars.githubusercontent.com/u/5913289?v=3","gravatar_id":"","url":"https://api.github.com/users/NerdWallet","html_url":"https://github.com/NerdWallet","followers_url":"https://api.github.com/users/NerdWallet/followers","following_url":"https://api.github.com/users/NerdWallet/following{/other_user}","gists_url":"https://api.github.com/users/NerdWallet/gists{/gist_id}","starred_url":"https://api.github.com/users/NerdWallet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NerdWallet/subscriptions","organizations_url":"https://api.github.com/users/NerdWallet/orgs","repos_url":"https://api.github.com/users/NerdWallet/repos","events_url":"https://api.github.com/users/NerdWallet/events{/privacy}","received_events_url":"https://api.github.com/users/NerdWallet/received_events","type":"Organization","site_admin":false},"repo":{"id":31392357,"name":"gobblin","full_name":"NerdWallet/gobblin","owner":{"login":"NerdWallet","id":5913289,"avatar_url":"https://avatars.githubusercontent.com/u/5913289?v=3","gravatar_id":"","url":"https://api.github.com/users/NerdWallet","html_url":"https://github.com/NerdWallet","followers_url":"https://api.github.com/users/NerdWallet/followers","following_url":"https://api.github.com/users/NerdWallet/following{/other_user}","gists_url":"https://api.github.com/users/NerdWallet/gists{/gist_id}","starred_url":"https://api.github.com/users/NerdWallet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NerdWallet/subscriptions","organizations_url":"https://api.github.com/users/NerdWallet/orgs","repos_url":"https://api.github.com/users/NerdWallet/repos","events_url":"https://api.github.com/users/NerdWallet/events{/privacy}","received_events_url":"https://api.github.com/users/NerdWallet/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/NerdWallet/gobblin","description":"NerdWallet fork of gobblin.  (Kobold??)","fork":true,"url":"https://api.github.com/repos/NerdWallet/gobblin","forks_url":"https://api.github.com/repos/NerdWallet/gobblin/forks","keys_url":"https://api.github.com/repos/NerdWallet/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/NerdWallet/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/NerdWallet/gobblin/teams","hooks_url":"https://api.github.com/repos/NerdWallet/gobblin/hooks","issue_events_url":"https://api.github.com/repos/NerdWallet/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/NerdWallet/gobblin/events","assignees_url":"https://api.github.com/repos/NerdWallet/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/NerdWallet/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/NerdWallet/gobblin/tags","blobs_url":"https://api.github.com/repos/NerdWallet/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/NerdWallet/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/NerdWallet/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/NerdWallet/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/NerdWallet/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/NerdWallet/gobblin/languages","stargazers_url":"https://api.github.com/repos/NerdWallet/gobblin/stargazers","contributors_url":"https://api.github.com/repos/NerdWallet/gobblin/contributors","subscribers_url":"https://api.github.com/repos/NerdWallet/gobblin/subscribers","subscription_url":"https://api.github.com/repos/NerdWallet/gobblin/subscription","commits_url":"https://api.github.com/repos/NerdWallet/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/NerdWallet/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/NerdWallet/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/NerdWallet/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/NerdWallet/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/NerdWallet/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/NerdWallet/gobblin/merges","archive_url":"https://api.github.com/repos/NerdWallet/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/NerdWallet/gobblin/downloads","issues_url":"https://api.github.com/repos/NerdWallet/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/NerdWallet/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/NerdWallet/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/NerdWallet/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/NerdWallet/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/NerdWallet/gobblin/releases{/id}","created_at":"2015-02-26T22:32:37Z","updated_at":"2015-05-01T17:21:28Z","pushed_at":"2015-05-13T22:49:33Z","git_url":"git://github.com/NerdWallet/gobblin.git","ssh_url":"git@github.com:NerdWallet/gobblin.git","clone_url":"https://github.com/NerdWallet/gobblin.git","svn_url":"https://github.com/NerdWallet/gobblin","homepage":"","size":77394,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"linkedin:master","ref":"master","sha":"f5bebda6ba3b624f9c451d9fcb162cdccad37633","user":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"repo":{"id":27392543,"name":"gobblin","full_name":"linkedin/gobblin","owner":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/linkedin/gobblin","description":"","fork":false,"url":"https://api.github.com/repos/linkedin/gobblin","forks_url":"https://api.github.com/repos/linkedin/gobblin/forks","keys_url":"https://api.github.com/repos/linkedin/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/linkedin/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/linkedin/gobblin/teams","hooks_url":"https://api.github.com/repos/linkedin/gobblin/hooks","issue_events_url":"https://api.github.com/repos/linkedin/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/linkedin/gobblin/events","assignees_url":"https://api.github.com/repos/linkedin/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/linkedin/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/linkedin/gobblin/tags","blobs_url":"https://api.github.com/repos/linkedin/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/linkedin/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/linkedin/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/linkedin/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/linkedin/gobblin/languages","stargazers_url":"https://api.github.com/repos/linkedin/gobblin/stargazers","contributors_url":"https://api.github.com/repos/linkedin/gobblin/contributors","subscribers_url":"https://api.github.com/repos/linkedin/gobblin/subscribers","subscription_url":"https://api.github.com/repos/linkedin/gobblin/subscription","commits_url":"https://api.github.com/repos/linkedin/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/linkedin/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/linkedin/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/linkedin/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/linkedin/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/linkedin/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/linkedin/gobblin/merges","archive_url":"https://api.github.com/repos/linkedin/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/linkedin/gobblin/downloads","issues_url":"https://api.github.com/repos/linkedin/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/linkedin/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/linkedin/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/linkedin/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/linkedin/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/linkedin/gobblin/releases{/id}","created_at":"2014-12-01T18:10:50Z","updated_at":"2015-05-07T20:00:41Z","pushed_at":"2015-05-13T23:24:50Z","git_url":"git://github.com/linkedin/gobblin.git","ssh_url":"git@github.com:linkedin/gobblin.git","clone_url":"https://github.com/linkedin/gobblin.git","svn_url":"https://github.com/linkedin/gobblin","homepage":null,"size":157360,"stargazers_count":37,"watchers_count":37,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":24,"mirror_url":null,"open_issues_count":5,"forks":24,"open_issues":5,"watchers":37,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/123"},"html":{"href":"https://github.com/linkedin/gobblin/pull/123"},"issue":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/123"},"comments":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/123/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/123/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/123/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/gobblin/statuses/3d35a60ae124b9b1c4da98f0cdaa4a900f21ecb1"}}}}