{"action":"created","comment":{"url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/comments/41353217","id":41353217,"diff_hunk":"@@ -62,106 +64,119 @@ public HdfsWriter(HdfsSinkConnectorConfig connectorConfig) {\n       if (!fs.exists(path)) {\n         fs.mkdirs(path);\n       }\n-      offsets = readOffsets();\n+      offsets = new HashMap<>();\n       writerProviderClass = (Class<RecordWriterProvider>)\n-          Class.forName(connectorConfig.getString(HdfsSinkConnectorConfig.RECORD_WRITER_PROVIDER_CLASS_CONFIG));\n+          Class.forName(connectorConfig.getString(\n+              HdfsSinkConnectorConfig.RECORD_WRITER_PROVIDER_CLASS_CONFIG));\n+\n+      this.assignment = assignment;\n     } catch (IOException e) {\n-      throw new CopycatRuntimeException(e);\n+      throw new CopycatException(e);\n     } catch (ClassNotFoundException e) {\n-      throw new CopycatRuntimeException(\"RecordWriterProvider class not found\", e);\n+      throw new CopycatException(\"RecordWriterProvider class not found\", e);\n     }\n   }\n \n-\n-  public Map<TopicPartition, Long> getPreviousOffsets() {\n+  /**\n+   * @return get committed offsets of previous task runs.\n+   */\n+  public Map<TopicPartition, Long> getCommittedOffsets() {\n     return offsets;\n   }\n \n-  public void writeRecord(TopicPartition topicPart, SinkRecord record)\n-      throws CopycatException{\n+  public void writeRecord(final TopicPartition topicPart, final SinkRecord record)\n+      throws CopycatException {\n     try {\n       RecordWriter<Long, SinkRecord> writer = getWriter(topicPart, record);\n       writer.write(System.currentTimeMillis(), record);\n+\n       if (!offsets.containsKey(topicPart)) {\n-        offsets.put(topicPart, record.getOffset() - 1);\n+        offsets.put(topicPart, record.kafkaOffset() - 1);\n       }\n       updateRecordCounter(topicPart);\n       if (shouldRotate(topicPart)) {\n         rotate(topicPart);\n       }\n     } catch (IOException e) {\n-      throw new CopycatRuntimeException(e);\n+      throw new CopycatException(e);\n     }\n   }\n \n-  public void close() throws IOException, CopycatException {\n-    for (TopicPartition topicPart: writers.keySet()) {\n+  public void close() throws CopycatException {\n+\n+    for (TopicPartition topicPart : assignment) {\n       rotate(topicPart);\n     }\n     writers.clear();\n+\n+    for (WAL wal : wals.values()) {\n+      wal.close();\n+    }\n+    wals.clear();\n+\n+    tempFileNames.clear();\n+    recordCounters.clear();\n+    offsets.clear();\n     if (fs != null) {\n-      fs.close();\n+      try {\n+        fs.close();\n+      } catch (IOException e) {\n+        throw new CopycatException(e);\n+      }\n     }\n   }\n \n-  private Map<TopicPartition, Long> readOffsets() throws IOException {\n-    Map<TopicPartition, Long> offsets = new HashMap<TopicPartition, Long>();\n-    Path path = new Path(url + \"/\" + topicsDir);\n+  public void recover(TopicPartition topicPart) throws CopycatException {\n+    if (!wals.containsKey(topicPart)) {\n+        WAL wal = new WAL(conf, url, topicsDir, topicPart);\n+        wal.apply();\n+        readOffsets(topicPart);\n+        wals.put(topicPart, wal);\n+    }\n+  }\n+\n+  private void readOffsets(TopicPartition topicPart) throws CopycatException{\n+    String topic = topicPart.topic();\n+    int partition = topicPart.partition();\n+    Path path = new Path(url + \"/\" + topicsDir + \"/\" + topic + \"/\" + partition);","path":"src/main/java/io/confluent/copycat/hdfs/HdfsWriter.java","position":168,"original_position":168,"commit_id":"6004ff846bfba4795d0691d2b55b9efa255cb3ec","original_commit_id":"6004ff846bfba4795d0691d2b55b9efa255cb3ec","user":{"login":"ewencp","id":47366,"avatar_url":"https://avatars.githubusercontent.com/u/47366?v=3","gravatar_id":"","url":"https://api.github.com/users/ewencp","html_url":"https://github.com/ewencp","followers_url":"https://api.github.com/users/ewencp/followers","following_url":"https://api.github.com/users/ewencp/following{/other_user}","gists_url":"https://api.github.com/users/ewencp/gists{/gist_id}","starred_url":"https://api.github.com/users/ewencp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ewencp/subscriptions","organizations_url":"https://api.github.com/users/ewencp/orgs","repos_url":"https://api.github.com/users/ewencp/repos","events_url":"https://api.github.com/users/ewencp/events{/privacy}","received_events_url":"https://api.github.com/users/ewencp/received_events","type":"User","site_admin":false},"body":"This could use one of the `FileUtils` methods.","created_at":"2015-10-07T05:32:24Z","updated_at":"2015-10-07T05:32:24Z","html_url":"https://github.com/confluentinc/copycat-hdfs/pull/6#discussion_r41353217","pull_request_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6","_links":{"self":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/comments/41353217"},"html":{"href":"https://github.com/confluentinc/copycat-hdfs/pull/6#discussion_r41353217"},"pull_request":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6"}}},"pull_request":{"url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6","id":43351247,"html_url":"https://github.com/confluentinc/copycat-hdfs/pull/6","diff_url":"https://github.com/confluentinc/copycat-hdfs/pull/6.diff","patch_url":"https://github.com/confluentinc/copycat-hdfs/pull/6.patch","issue_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/6","number":6,"state":"open","locked":false,"title":"Exactly once delivery with WAL","user":{"login":"Ishiihara","id":1189848,"avatar_url":"https://avatars.githubusercontent.com/u/1189848?v=3","gravatar_id":"","url":"https://api.github.com/users/Ishiihara","html_url":"https://github.com/Ishiihara","followers_url":"https://api.github.com/users/Ishiihara/followers","following_url":"https://api.github.com/users/Ishiihara/following{/other_user}","gists_url":"https://api.github.com/users/Ishiihara/gists{/gist_id}","starred_url":"https://api.github.com/users/Ishiihara/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Ishiihara/subscriptions","organizations_url":"https://api.github.com/users/Ishiihara/orgs","repos_url":"https://api.github.com/users/Ishiihara/repos","events_url":"https://api.github.com/users/Ishiihara/events{/privacy}","received_events_url":"https://api.github.com/users/Ishiihara/received_events","type":"User","site_admin":false},"body":"@ewencp  Can you review it and provide some feedback? Thanks!\r\nThis PR adds the following functionality:\r\n1. Time based file rotate using ScheduledExecutorService. \r\n2. WAL implementation to ensure exactly once delivery in case of failure and zombie writes.\r\n\r\nTODO:\r\n1. Replace SequenceFile class with a customized format. This PR uses the SequenceFile  implementation in hadoop trunk. \r\n2. WAL truncation. Currently, to make sure that we don't commit a file in WAL with a certain starting offset if there is already a committed file with the same start offset to ensure that data is not written to HDFS more than once.  This operation is relatively expensive if there are lots of committed files under a topic partition directory.  Truncating the WAL can reduce the number of such operations.  \r\n3. Create WAL factory class to enable different implementation of WAL and some refactoring of WAL api. \r\n4. Currently, HBase uses disruptor to perform sync to HDFS asynchronously. And we call fsync appending each entry to WAL? Given that our frequency in writing to WAL is much smaller than HBase, do you think it is worth exploring this option? ","created_at":"2015-08-26T00:05:17Z","updated_at":"2015-10-07T05:32:24Z","closed_at":null,"merged_at":null,"merge_commit_sha":"5a41b975afa74c1e76e95daab49a2c7240f5fd94","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6/commits","review_comments_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6/comments","review_comment_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/comments{/number}","comments_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/6/comments","statuses_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/statuses/6004ff846bfba4795d0691d2b55b9efa255cb3ec","head":{"label":"confluentinc:exactly-once","ref":"exactly-once","sha":"6004ff846bfba4795d0691d2b55b9efa255cb3ec","user":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":35967803,"name":"copycat-hdfs","full_name":"confluentinc/copycat-hdfs","owner":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/copycat-hdfs","description":"Copycat HDFS connector","fork":false,"url":"https://api.github.com/repos/confluentinc/copycat-hdfs","forks_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/forks","keys_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/teams","hooks_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/events","assignees_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/tags","blobs_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/languages","stargazers_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/subscription","commits_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/merges","archive_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/downloads","issues_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/releases{/id}","created_at":"2015-05-20T18:54:34Z","updated_at":"2015-09-30T20:12:47Z","pushed_at":"2015-10-06T05:09:20Z","git_url":"git://github.com/confluentinc/copycat-hdfs.git","ssh_url":"git@github.com:confluentinc/copycat-hdfs.git","clone_url":"https://github.com/confluentinc/copycat-hdfs.git","svn_url":"https://github.com/confluentinc/copycat-hdfs","homepage":null,"size":252,"stargazers_count":2,"watchers_count":2,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":4,"forks":0,"open_issues":4,"watchers":2,"default_branch":"master"}},"base":{"label":"confluentinc:master","ref":"master","sha":"e802567234b9b934548fba17a6addb555b0317a6","user":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":35967803,"name":"copycat-hdfs","full_name":"confluentinc/copycat-hdfs","owner":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/copycat-hdfs","description":"Copycat HDFS connector","fork":false,"url":"https://api.github.com/repos/confluentinc/copycat-hdfs","forks_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/forks","keys_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/teams","hooks_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/events","assignees_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/tags","blobs_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/languages","stargazers_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/subscription","commits_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/merges","archive_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/downloads","issues_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/copycat-hdfs/releases{/id}","created_at":"2015-05-20T18:54:34Z","updated_at":"2015-09-30T20:12:47Z","pushed_at":"2015-10-06T05:09:20Z","git_url":"git://github.com/confluentinc/copycat-hdfs.git","ssh_url":"git@github.com:confluentinc/copycat-hdfs.git","clone_url":"https://github.com/confluentinc/copycat-hdfs.git","svn_url":"https://github.com/confluentinc/copycat-hdfs","homepage":null,"size":252,"stargazers_count":2,"watchers_count":2,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":4,"forks":0,"open_issues":4,"watchers":2,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6"},"html":{"href":"https://github.com/confluentinc/copycat-hdfs/pull/6"},"issue":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/6"},"comments":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/issues/6/comments"},"review_comments":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6/comments"},"review_comment":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/pulls/6/commits"},"statuses":{"href":"https://api.github.com/repos/confluentinc/copycat-hdfs/statuses/6004ff846bfba4795d0691d2b55b9efa255cb3ec"}}}}