{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/27437230","id":27437230,"diff_hunk":"@@ -80,87 +79,122 @@ public Object convertSchema(Object inputSchema, WorkUnitState workUnit)\n   }\n \n   /**\n-   * A type of {@link java.util.Iterator} to be used with {@link MultiConverter}.\n+   * A type of {@link java.util.Iterator} to be used with {@link MultiConverter}. The Converter uses the\n+   * {@link ChainedConverterIterator} to chain iterators together. The first ChainedConverterIterator created contains\n+   * an iterator with only the inputRecord and the first converter in the converters list. Each subsequent\n+   * ChainedConverterIterator is created using the previous ChainedConverterIterator along with the next converter in\n+   * the converters list. By chaining the converters and iterators in this fashion, a reference to the last\n+   * ChainedConverterIterator will be sufficient to iterate through all the data.\n    */\n   private class MultiConverterIterator implements Iterator<Object> {\n \n     private final WorkUnitState workUnitState;\n-    private final Deque<Iterator<Object>> converterIteratorStack = Lists.newLinkedList();\n-    private Iterator<Object> outputIterator;\n+\n+    private Iterator<Object> chainedConverterIterator;\n \n     public MultiConverterIterator(Object inputRecord, WorkUnitState workUnitState) throws DataConversionException {\n       this.workUnitState = workUnitState;\n-\n-      // Construct the initial stack of converter iterators\n-      if (converters.isEmpty()) {\n-        this.converterIteratorStack.push(new SingleRecordIterable<Object>(inputRecord).iterator());\n-      } else {\n-        Object record = inputRecord;\n-        for (Converter converter : converters) {\n-          if (!this.converterIteratorStack.isEmpty()) {\n-            record = this.converterIteratorStack.peek().next();\n-          }\n-          Iterator<Object> iterator =\n-              converter.convertRecord(convertedSchemaMap.get(converter), record, workUnitState).iterator();\n-          this.converterIteratorStack.push(iterator);\n-          // Do not continue when encountering an empty iterator\n-          if (!iterator.hasNext()) {\n-            break;\n-          }\n+      this.chainedConverterIterator = new SingleRecordIterable<Object>(inputRecord).iterator();\n+\n+      for (int i = 0; i < converters.size(); i++) {\n+        if (i == 0) {\n+          this.chainedConverterIterator =\n+              new ChainedConverterIterator(new SingleRecordIterable<Object>(inputRecord).iterator(), converters.get(i));\n+        } else {\n+          this.chainedConverterIterator =\n+              new ChainedConverterIterator(this.chainedConverterIterator, converters.get(i));\n         }\n       }\n-\n-      this.outputIterator = this.converterIteratorStack.peek();\n     }\n \n     @Override\n     public boolean hasNext() {\n-      if (this.outputIterator.hasNext()) {\n-        return true;\n-      }\n+      return this.chainedConverterIterator.hasNext();\n+    }\n \n-      // Pop out all empty converter iterators from the stack\n-      while (!this.converterIteratorStack.isEmpty() && !this.converterIteratorStack.peek().hasNext()) {\n-        this.converterIteratorStack.pop();\n+    @Override\n+    public Object next() {\n+      return this.chainedConverterIterator.next();\n+    }\n+\n+    @Override\n+    public void remove() {\n+      throw new UnsupportedOperationException();\n+    }\n+\n+    /**\n+     * A helper class that implements {@link Iterator}. It is constructed with a {@link Iterator} and a {@link Converter}.\n+     * The class iterates through the results of each converted record from prevIterator. It iterates through each\n+     * element in prevIterator, converts the result, and then stores the result in the currentIterator object. It\n+     * returns every element in currentIterator until it is empty, and then it gets the next element from prevIterator,\n+     * converts the object, and stores the result in currentIterator. This pattern continues until there are no more\n+     * elements left in prevIterator.\n+     */\n+    private class ChainedConverterIterator implements Iterator<Object> {\n+\n+      private final Converter converter;\n+      private final Iterator<Object> prevIterator;\n+\n+      private Iterator<Object> currentIterator;\n+\n+      public ChainedConverterIterator(Iterator<Object> prevIterator, Converter converter)\n+          throws DataConversionException {\n+        this.converter = converter;\n+        this.prevIterator = prevIterator;\n+\n+        if (this.prevIterator.hasNext()) {\n+          this.currentIterator =\n+              converter.convertRecord(convertedSchemaMap.get(converter), this.prevIterator.next(), workUnitState)\n+                  .iterator();\n+        } else {\n+          this.currentIterator = new EmptyIterable<Object>().iterator();\n+        }\n       }\n \n-      if (this.converterIteratorStack.isEmpty()) {\n+      @Override\n+      public boolean hasNext() {\n+        if (this.currentIterator.hasNext()) {\n+          return true;\n+        }\n+        while (this.prevIterator.hasNext()) {\n+          try {\n+            this.currentIterator =\n+                converter.convertRecord(convertedSchemaMap.get(converter), this.prevIterator.next(), workUnitState)\n+                    .iterator();\n+          } catch (DataConversionException e) {\n+            Throwables.propagate(e);\n+          }\n+          if (this.currentIterator.hasNext()) {\n+            return true;\n+          }\n+        }\n         return false;\n       }\n \n-      // Get the next record from the iterator (that still has more records) at the top of the stack\n-      // and reconstruct the stack of converter iterators from that record\n-      for (int i = this.converterIteratorStack.size(); i < converters.size(); i++) {\n-        Object record = this.converterIteratorStack.peek().next();\n-        Converter converter = converters.get(i);\n-        try {\n-          Iterator<Object> iterator =\n-              converter.convertRecord(convertedSchemaMap.get(converter), record, this.workUnitState).iterator();\n-          this.converterIteratorStack.push(iterator);\n-          // Do not continue when encountering an empty iterator\n-          if (!iterator.hasNext()) {\n-            break;\n+      @Override\n+      public Object next() {\n+        if (this.currentIterator.hasNext()) {","path":"gobblin-runtime/src/main/java/gobblin/runtime/MultiConverter.java","position":170,"original_position":170,"commit_id":"5d55cc0740a81710fe81bd4e7b05d3e958d82731","original_commit_id":"5d55cc0740a81710fe81bd4e7b05d3e958d82731","user":{"login":"liyinan926","id":3709137,"avatar_url":"https://avatars.githubusercontent.com/u/3709137?v=3","gravatar_id":"","url":"https://api.github.com/users/liyinan926","html_url":"https://github.com/liyinan926","followers_url":"https://api.github.com/users/liyinan926/followers","following_url":"https://api.github.com/users/liyinan926/following{/other_user}","gists_url":"https://api.github.com/users/liyinan926/gists{/gist_id}","starred_url":"https://api.github.com/users/liyinan926/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liyinan926/subscriptions","organizations_url":"https://api.github.com/users/liyinan926/orgs","repos_url":"https://api.github.com/users/liyinan926/repos","events_url":"https://api.github.com/users/liyinan926/events{/privacy}","received_events_url":"https://api.github.com/users/liyinan926/received_events","type":"User","site_admin":false},"body":"You can replace the duplicate code here with a call to `hasNext()`, and throw a `NoSuchElementException` if `hasNext` returns false. ","created_at":"2015-03-30T22:15:29Z","updated_at":"2015-03-30T22:15:29Z","html_url":"https://github.com/linkedin/gobblin/pull/63#discussion_r27437230","pull_request_url":"https://api.github.com/repos/linkedin/gobblin/pulls/63","_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/27437230"},"html":{"href":"https://github.com/linkedin/gobblin/pull/63#discussion_r27437230"},"pull_request":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/63"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/63","id":32072147,"html_url":"https://github.com/linkedin/gobblin/pull/63","diff_url":"https://github.com/linkedin/gobblin/pull/63.diff","patch_url":"https://github.com/linkedin/gobblin/pull/63.patch","issue_url":"https://api.github.com/repos/linkedin/gobblin/issues/63","number":63,"state":"open","locked":false,"title":"Initial pull request for the Gobblin Log Parser","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"body":"This pull request contains the changes for MutliConverter and MultiConverterTest.\r\n\r\nThis pull request adds a series of Converters and a Source class.\r\n\r\nConverters:\r\n\r\nObjectToStringConverter.java - converts an Object to its String representation\r\nStringFilterConverter.java - filters out an inputRecord based on whether or not it matches a regex\r\nStringSplitterConverter.java - splits a string based on some configurable value\r\n\r\nSource:\r\n\r\nDatePartitionedSource.java - implementation of the Source interface that reads data from a date-partitioned directory\r\nMultiWorkUnitWeightedQueue.java - users a PriorityQueue to pair WorkUnits together based on some weight","created_at":"2015-03-26T23:38:42Z","updated_at":"2015-03-30T22:15:29Z","closed_at":null,"merged_at":null,"merge_commit_sha":"9f367a05158a5e2863869c3fdc213811b71a01e3","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/gobblin/pulls/63/commits","review_comments_url":"https://api.github.com/repos/linkedin/gobblin/pulls/63/comments","review_comment_url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/gobblin/issues/63/comments","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/5d55cc0740a81710fe81bd4e7b05d3e958d82731","head":{"label":"sahilTakiar:log-parser","ref":"log-parser","sha":"5d55cc0740a81710fe81bd4e7b05d3e958d82731","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"repo":{"id":31343048,"name":"gobblin","full_name":"sahilTakiar/gobblin","owner":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/sahilTakiar/gobblin","description":"","fork":true,"url":"https://api.github.com/repos/sahilTakiar/gobblin","forks_url":"https://api.github.com/repos/sahilTakiar/gobblin/forks","keys_url":"https://api.github.com/repos/sahilTakiar/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/sahilTakiar/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/sahilTakiar/gobblin/teams","hooks_url":"https://api.github.com/repos/sahilTakiar/gobblin/hooks","issue_events_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/sahilTakiar/gobblin/events","assignees_url":"https://api.github.com/repos/sahilTakiar/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/sahilTakiar/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/sahilTakiar/gobblin/tags","blobs_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/sahilTakiar/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/sahilTakiar/gobblin/languages","stargazers_url":"https://api.github.com/repos/sahilTakiar/gobblin/stargazers","contributors_url":"https://api.github.com/repos/sahilTakiar/gobblin/contributors","subscribers_url":"https://api.github.com/repos/sahilTakiar/gobblin/subscribers","subscription_url":"https://api.github.com/repos/sahilTakiar/gobblin/subscription","commits_url":"https://api.github.com/repos/sahilTakiar/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/sahilTakiar/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/sahilTakiar/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/sahilTakiar/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/sahilTakiar/gobblin/merges","archive_url":"https://api.github.com/repos/sahilTakiar/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/sahilTakiar/gobblin/downloads","issues_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/sahilTakiar/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/sahilTakiar/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/sahilTakiar/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/sahilTakiar/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/sahilTakiar/gobblin/releases{/id}","created_at":"2015-02-26T00:28:16Z","updated_at":"2015-03-26T17:30:17Z","pushed_at":"2015-03-30T21:19:14Z","git_url":"git://github.com/sahilTakiar/gobblin.git","ssh_url":"git@github.com:sahilTakiar/gobblin.git","clone_url":"https://github.com/sahilTakiar/gobblin.git","svn_url":"https://github.com/sahilTakiar/gobblin","homepage":null,"size":76032,"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":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"linkedin:master","ref":"master","sha":"2e7edf740c9031de4f029816c533672c0d4c7d27","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-03-29T17:49:34Z","pushed_at":"2015-03-30T21:19:15Z","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":151557,"stargazers_count":23,"watchers_count":23,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":15,"mirror_url":null,"open_issues_count":5,"forks":15,"open_issues":5,"watchers":23,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/63"},"html":{"href":"https://github.com/linkedin/gobblin/pull/63"},"issue":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/63"},"comments":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/63/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/63/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/63/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/gobblin/statuses/5d55cc0740a81710fe81bd4e7b05d3e958d82731"}}}}