{"action":"created","comment":{"url":"https://api.github.com/repos/druid-io/druid/pulls/comments/35034439","id":35034439,"diff_hunk":"@@ -325,6 +299,134 @@ public void run()\n       }\n     }\n \n+    private void runFirehoseV2(FirehoseV2 firehose)\n+    {\n+      final Period intermediatePersistPeriod = config.getIntermediatePersistPeriod();\n+      try {\n+        firehose.start();\n+      }\n+      catch (Exception e) {\n+      \tlog.error(e, \"Failed to start firehoseV2\");\n+      \treturn;\n+      }\n+      long nextFlush = new DateTime().plus(intermediatePersistPeriod).getMillis();\n+      log.info(\"Firehose started with nextFlush [%s]\", nextFlush);\n+      boolean haveRow = true;\n+      while (haveRow) {\n+        InputRow inputRow = null;\n+        int numRows = 0;\n+        try {\n+          inputRow = firehose.currRow();\n+          if (inputRow != null) {\n+            try {\n+              numRows = plumber.add(inputRow);\n+            }\n+            catch (IndexSizeExceededException e) {\n+              log.info(\"Index limit exceeded: %s\", e.getMessage());\n+              nextFlush = doIncrementalPersist(firehose.makeCommitter(), intermediatePersistPeriod);\n+              continue;\n+            }\n+\n+            if (numRows < 0) {\n+              metrics.incrementThrownAway();\n+              log.debug(\"Throwing away event[%s]\", inputRow);\n+            } else {\n+              metrics.incrementProcessed();\n+            }\n+          }\n+        }\n+        catch (Exception e) {","path":"server/src/main/java/io/druid/segment/realtime/RealtimeManager.java","position":160,"original_position":161,"commit_id":"037d5743b388dda81219bf340d0b7bdae5b2fe45","original_commit_id":"910bcceb07e152c1c53d1455390dcfebfcc3e7ab","user":{"login":"gianm","id":1214075,"avatar_url":"https://avatars.githubusercontent.com/u/1214075?v=3","gravatar_id":"","url":"https://api.github.com/users/gianm","html_url":"https://github.com/gianm","followers_url":"https://api.github.com/users/gianm/followers","following_url":"https://api.github.com/users/gianm/following{/other_user}","gists_url":"https://api.github.com/users/gianm/gists{/gist_id}","starred_url":"https://api.github.com/users/gianm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/gianm/subscriptions","organizations_url":"https://api.github.com/users/gianm/orgs","repos_url":"https://api.github.com/users/gianm/repos","events_url":"https://api.github.com/users/gianm/events{/privacy}","received_events_url":"https://api.github.com/users/gianm/received_events","type":"User","site_admin":false},"body":"I don't think this is a good idea. I don't think we want exceptions in firehose.makeCommitter, plumber.add, or plumber.persist to cause us to skip over whatever InputRow happened to trigger it, especially in a design that is trying to keep Druid consistent with what is in the Firehose.\r\n\r\nMy reasoning is that the Plumbers shouldn't be throwing exceptions except in case of bugs in the indexing code or running out of space on disk. In those situations, as someone running Druid, I'd actually want the indexer to send a really loud alert and shut down. I would probably set up some external supervision that tries to re-start the task when that happens, since that way I can try to get it running again from a checkpoint, which is preferred since it doesn't sacrifice the guarantees we're trying to provide.\r\n\r\nIf people are really demanding this behavior then I think it should be an option that is off by default.","created_at":"2015-07-20T19:36:48Z","updated_at":"2015-07-20T19:36:48Z","html_url":"https://github.com/druid-io/druid/pull/1482#discussion_r35034439","pull_request_url":"https://api.github.com/repos/druid-io/druid/pulls/1482","_links":{"self":{"href":"https://api.github.com/repos/druid-io/druid/pulls/comments/35034439"},"html":{"href":"https://github.com/druid-io/druid/pull/1482#discussion_r35034439"},"pull_request":{"href":"https://api.github.com/repos/druid-io/druid/pulls/1482"}}},"pull_request":{"url":"https://api.github.com/repos/druid-io/druid/pulls/1482","id":38913364,"html_url":"https://github.com/druid-io/druid/pull/1482","diff_url":"https://github.com/druid-io/druid/pull/1482.diff","patch_url":"https://github.com/druid-io/druid/pull/1482.patch","issue_url":"https://api.github.com/repos/druid-io/druid/issues/1482","number":1482,"state":"open","locked":false,"title":"kafka 8 simple consumer firehose","user":{"login":"anwenxu","id":2835988,"avatar_url":"https://avatars.githubusercontent.com/u/2835988?v=3","gravatar_id":"","url":"https://api.github.com/users/anwenxu","html_url":"https://github.com/anwenxu","followers_url":"https://api.github.com/users/anwenxu/followers","following_url":"https://api.github.com/users/anwenxu/following{/other_user}","gists_url":"https://api.github.com/users/anwenxu/gists{/gist_id}","starred_url":"https://api.github.com/users/anwenxu/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/anwenxu/subscriptions","organizations_url":"https://api.github.com/users/anwenxu/orgs","repos_url":"https://api.github.com/users/anwenxu/repos","events_url":"https://api.github.com/users/anwenxu/events{/privacy}","received_events_url":"https://api.github.com/users/anwenxu/received_events","type":"User","site_admin":false},"body":"This feature introduces a simple consumer implementation for realtime firehose. It keeps track of current offset metadata by storing it in metadata.drd with smooshed files, and will be able to recover the previous offset position after restart.\r\nOn restart, it take down the offset from the sequentially \"last\" valid persisted file, and rename all incomplete persist directory to a path for corrupted data. e.g. there are sub dirs /1, /2, /3, ..../8 under directory mydatasource/20150630T10:00:00-11:00:00/, and /7 doesn't contain meta.smoosh, RealtimePlumber will rename both /7 and /8 to corrupted/mydatasource/20150630T10:00:00-11:00:00/* and use offset from /6 as starting point.\r\nWhen indexing, it follows the logic:\r\n1. Call start()\r\n2. Read currRow()\r\n3. Call advance()\r\n4. If index should be committed: commit()\r\n5. GOTO 2\r\nNote it will only advance when current row is successfully processed, which means saving the end offset of current message.\r\n@cheddar \r\n\r\nGoogle Group reference:\r\nhttps://groups.google.com/forum/#!topic/druid-development/9HB9hCcqvuI","created_at":"2015-06-30T19:10:41Z","updated_at":"2015-07-20T19:36:48Z","closed_at":null,"merged_at":null,"merge_commit_sha":"3891b6a7c76ed9904421a8b4ee7efb07a3ef0624","assignee":null,"milestone":{"url":"https://api.github.com/repos/druid-io/druid/milestones/15","html_url":"https://github.com/druid-io/druid/milestones/0.8.1","labels_url":"https://api.github.com/repos/druid-io/druid/milestones/15/labels","id":1148224,"number":15,"title":"0.8.1","description":"","creator":{"login":"xvrl","id":815147,"avatar_url":"https://avatars.githubusercontent.com/u/815147?v=3","gravatar_id":"","url":"https://api.github.com/users/xvrl","html_url":"https://github.com/xvrl","followers_url":"https://api.github.com/users/xvrl/followers","following_url":"https://api.github.com/users/xvrl/following{/other_user}","gists_url":"https://api.github.com/users/xvrl/gists{/gist_id}","starred_url":"https://api.github.com/users/xvrl/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/xvrl/subscriptions","organizations_url":"https://api.github.com/users/xvrl/orgs","repos_url":"https://api.github.com/users/xvrl/repos","events_url":"https://api.github.com/users/xvrl/events{/privacy}","received_events_url":"https://api.github.com/users/xvrl/received_events","type":"User","site_admin":false},"open_issues":4,"closed_issues":17,"state":"open","created_at":"2015-06-03T20:18:18Z","updated_at":"2015-07-17T23:24:42Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/druid-io/druid/pulls/1482/commits","review_comments_url":"https://api.github.com/repos/druid-io/druid/pulls/1482/comments","review_comment_url":"https://api.github.com/repos/druid-io/druid/pulls/comments{/number}","comments_url":"https://api.github.com/repos/druid-io/druid/issues/1482/comments","statuses_url":"https://api.github.com/repos/druid-io/druid/statuses/037d5743b388dda81219bf340d0b7bdae5b2fe45","head":{"label":"anwenxu:feature-kafka-simple-consumer","ref":"feature-kafka-simple-consumer","sha":"037d5743b388dda81219bf340d0b7bdae5b2fe45","user":{"login":"anwenxu","id":2835988,"avatar_url":"https://avatars.githubusercontent.com/u/2835988?v=3","gravatar_id":"","url":"https://api.github.com/users/anwenxu","html_url":"https://github.com/anwenxu","followers_url":"https://api.github.com/users/anwenxu/followers","following_url":"https://api.github.com/users/anwenxu/following{/other_user}","gists_url":"https://api.github.com/users/anwenxu/gists{/gist_id}","starred_url":"https://api.github.com/users/anwenxu/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/anwenxu/subscriptions","organizations_url":"https://api.github.com/users/anwenxu/orgs","repos_url":"https://api.github.com/users/anwenxu/repos","events_url":"https://api.github.com/users/anwenxu/events{/privacy}","received_events_url":"https://api.github.com/users/anwenxu/received_events","type":"User","site_admin":false},"repo":{"id":34215933,"name":"druid","full_name":"anwenxu/druid","owner":{"login":"anwenxu","id":2835988,"avatar_url":"https://avatars.githubusercontent.com/u/2835988?v=3","gravatar_id":"","url":"https://api.github.com/users/anwenxu","html_url":"https://github.com/anwenxu","followers_url":"https://api.github.com/users/anwenxu/followers","following_url":"https://api.github.com/users/anwenxu/following{/other_user}","gists_url":"https://api.github.com/users/anwenxu/gists{/gist_id}","starred_url":"https://api.github.com/users/anwenxu/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/anwenxu/subscriptions","organizations_url":"https://api.github.com/users/anwenxu/orgs","repos_url":"https://api.github.com/users/anwenxu/repos","events_url":"https://api.github.com/users/anwenxu/events{/privacy}","received_events_url":"https://api.github.com/users/anwenxu/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/anwenxu/druid","description":"Druid Data Store","fork":true,"url":"https://api.github.com/repos/anwenxu/druid","forks_url":"https://api.github.com/repos/anwenxu/druid/forks","keys_url":"https://api.github.com/repos/anwenxu/druid/keys{/key_id}","collaborators_url":"https://api.github.com/repos/anwenxu/druid/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/anwenxu/druid/teams","hooks_url":"https://api.github.com/repos/anwenxu/druid/hooks","issue_events_url":"https://api.github.com/repos/anwenxu/druid/issues/events{/number}","events_url":"https://api.github.com/repos/anwenxu/druid/events","assignees_url":"https://api.github.com/repos/anwenxu/druid/assignees{/user}","branches_url":"https://api.github.com/repos/anwenxu/druid/branches{/branch}","tags_url":"https://api.github.com/repos/anwenxu/druid/tags","blobs_url":"https://api.github.com/repos/anwenxu/druid/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/anwenxu/druid/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/anwenxu/druid/git/refs{/sha}","trees_url":"https://api.github.com/repos/anwenxu/druid/git/trees{/sha}","statuses_url":"https://api.github.com/repos/anwenxu/druid/statuses/{sha}","languages_url":"https://api.github.com/repos/anwenxu/druid/languages","stargazers_url":"https://api.github.com/repos/anwenxu/druid/stargazers","contributors_url":"https://api.github.com/repos/anwenxu/druid/contributors","subscribers_url":"https://api.github.com/repos/anwenxu/druid/subscribers","subscription_url":"https://api.github.com/repos/anwenxu/druid/subscription","commits_url":"https://api.github.com/repos/anwenxu/druid/commits{/sha}","git_commits_url":"https://api.github.com/repos/anwenxu/druid/git/commits{/sha}","comments_url":"https://api.github.com/repos/anwenxu/druid/comments{/number}","issue_comment_url":"https://api.github.com/repos/anwenxu/druid/issues/comments{/number}","contents_url":"https://api.github.com/repos/anwenxu/druid/contents/{+path}","compare_url":"https://api.github.com/repos/anwenxu/druid/compare/{base}...{head}","merges_url":"https://api.github.com/repos/anwenxu/druid/merges","archive_url":"https://api.github.com/repos/anwenxu/druid/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/anwenxu/druid/downloads","issues_url":"https://api.github.com/repos/anwenxu/druid/issues{/number}","pulls_url":"https://api.github.com/repos/anwenxu/druid/pulls{/number}","milestones_url":"https://api.github.com/repos/anwenxu/druid/milestones{/number}","notifications_url":"https://api.github.com/repos/anwenxu/druid/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/anwenxu/druid/labels{/name}","releases_url":"https://api.github.com/repos/anwenxu/druid/releases{/id}","created_at":"2015-04-19T17:13:14Z","updated_at":"2015-04-19T17:13:19Z","pushed_at":"2015-07-20T19:11:02Z","git_url":"git://github.com/anwenxu/druid.git","ssh_url":"git@github.com:anwenxu/druid.git","clone_url":"https://github.com/anwenxu/druid.git","svn_url":"https://github.com/anwenxu/druid","homepage":"","size":95982,"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":"druid-io:master","ref":"master","sha":"020827349e8a80e07dab38b2400dfe478a7420ef","user":{"login":"druid-io","id":4203474,"avatar_url":"https://avatars.githubusercontent.com/u/4203474?v=3","gravatar_id":"","url":"https://api.github.com/users/druid-io","html_url":"https://github.com/druid-io","followers_url":"https://api.github.com/users/druid-io/followers","following_url":"https://api.github.com/users/druid-io/following{/other_user}","gists_url":"https://api.github.com/users/druid-io/gists{/gist_id}","starred_url":"https://api.github.com/users/druid-io/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/druid-io/subscriptions","organizations_url":"https://api.github.com/users/druid-io/orgs","repos_url":"https://api.github.com/users/druid-io/repos","events_url":"https://api.github.com/users/druid-io/events{/privacy}","received_events_url":"https://api.github.com/users/druid-io/received_events","type":"Organization","site_admin":false},"repo":{"id":6358188,"name":"druid","full_name":"druid-io/druid","owner":{"login":"druid-io","id":4203474,"avatar_url":"https://avatars.githubusercontent.com/u/4203474?v=3","gravatar_id":"","url":"https://api.github.com/users/druid-io","html_url":"https://github.com/druid-io","followers_url":"https://api.github.com/users/druid-io/followers","following_url":"https://api.github.com/users/druid-io/following{/other_user}","gists_url":"https://api.github.com/users/druid-io/gists{/gist_id}","starred_url":"https://api.github.com/users/druid-io/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/druid-io/subscriptions","organizations_url":"https://api.github.com/users/druid-io/orgs","repos_url":"https://api.github.com/users/druid-io/repos","events_url":"https://api.github.com/users/druid-io/events{/privacy}","received_events_url":"https://api.github.com/users/druid-io/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/druid-io/druid","description":"Column oriented distributed data store ideal for powering interactive applications","fork":false,"url":"https://api.github.com/repos/druid-io/druid","forks_url":"https://api.github.com/repos/druid-io/druid/forks","keys_url":"https://api.github.com/repos/druid-io/druid/keys{/key_id}","collaborators_url":"https://api.github.com/repos/druid-io/druid/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/druid-io/druid/teams","hooks_url":"https://api.github.com/repos/druid-io/druid/hooks","issue_events_url":"https://api.github.com/repos/druid-io/druid/issues/events{/number}","events_url":"https://api.github.com/repos/druid-io/druid/events","assignees_url":"https://api.github.com/repos/druid-io/druid/assignees{/user}","branches_url":"https://api.github.com/repos/druid-io/druid/branches{/branch}","tags_url":"https://api.github.com/repos/druid-io/druid/tags","blobs_url":"https://api.github.com/repos/druid-io/druid/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/druid-io/druid/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/druid-io/druid/git/refs{/sha}","trees_url":"https://api.github.com/repos/druid-io/druid/git/trees{/sha}","statuses_url":"https://api.github.com/repos/druid-io/druid/statuses/{sha}","languages_url":"https://api.github.com/repos/druid-io/druid/languages","stargazers_url":"https://api.github.com/repos/druid-io/druid/stargazers","contributors_url":"https://api.github.com/repos/druid-io/druid/contributors","subscribers_url":"https://api.github.com/repos/druid-io/druid/subscribers","subscription_url":"https://api.github.com/repos/druid-io/druid/subscription","commits_url":"https://api.github.com/repos/druid-io/druid/commits{/sha}","git_commits_url":"https://api.github.com/repos/druid-io/druid/git/commits{/sha}","comments_url":"https://api.github.com/repos/druid-io/druid/comments{/number}","issue_comment_url":"https://api.github.com/repos/druid-io/druid/issues/comments{/number}","contents_url":"https://api.github.com/repos/druid-io/druid/contents/{+path}","compare_url":"https://api.github.com/repos/druid-io/druid/compare/{base}...{head}","merges_url":"https://api.github.com/repos/druid-io/druid/merges","archive_url":"https://api.github.com/repos/druid-io/druid/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/druid-io/druid/downloads","issues_url":"https://api.github.com/repos/druid-io/druid/issues{/number}","pulls_url":"https://api.github.com/repos/druid-io/druid/pulls{/number}","milestones_url":"https://api.github.com/repos/druid-io/druid/milestones{/number}","notifications_url":"https://api.github.com/repos/druid-io/druid/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/druid-io/druid/labels{/name}","releases_url":"https://api.github.com/repos/druid-io/druid/releases{/id}","created_at":"2012-10-23T19:08:07Z","updated_at":"2015-07-19T09:16:23Z","pushed_at":"2015-07-20T19:11:03Z","git_url":"git://github.com/druid-io/druid.git","ssh_url":"git@github.com:druid-io/druid.git","clone_url":"https://github.com/druid-io/druid.git","svn_url":"https://github.com/druid-io/druid","homepage":"http://druid.io","size":172240,"stargazers_count":1878,"watchers_count":1878,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":412,"mirror_url":null,"open_issues_count":137,"forks":412,"open_issues":137,"watchers":1878,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/druid-io/druid/pulls/1482"},"html":{"href":"https://github.com/druid-io/druid/pull/1482"},"issue":{"href":"https://api.github.com/repos/druid-io/druid/issues/1482"},"comments":{"href":"https://api.github.com/repos/druid-io/druid/issues/1482/comments"},"review_comments":{"href":"https://api.github.com/repos/druid-io/druid/pulls/1482/comments"},"review_comment":{"href":"https://api.github.com/repos/druid-io/druid/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/druid-io/druid/pulls/1482/commits"},"statuses":{"href":"https://api.github.com/repos/druid-io/druid/statuses/037d5743b388dda81219bf340d0b7bdae5b2fe45"}}}}