{"action":"created","issue":{"url":"https://api.github.com/repos/linkedin/gobblin/issues/135","labels_url":"https://api.github.com/repos/linkedin/gobblin/issues/135/labels{/name}","comments_url":"https://api.github.com/repos/linkedin/gobblin/issues/135/comments","events_url":"https://api.github.com/repos/linkedin/gobblin/issues/135/events","html_url":"https://github.com/linkedin/gobblin/pull/135","id":77909305,"number":135,"title":"Refactor Kafka adapter for complex watermark","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},"labels":[{"url":"https://api.github.com/repos/linkedin/gobblin/labels/enhancement","name":"enhancement","color":"84b6eb"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":5,"created_at":"2015-05-19T03:17:32Z","updated_at":"2015-05-20T19:20:20Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/135","html_url":"https://github.com/linkedin/gobblin/pull/135","diff_url":"https://github.com/linkedin/gobblin/pull/135.diff","patch_url":"https://github.com/linkedin/gobblin/pull/135.patch"},"body":"- Refactored `KafkaSource` and `KafkaExtractor` for the complex watermark:\r\n\r\nNow each WorkUnit in a MultiWorkUnit returned by `KafkaSource.getWorkUnits()` is\r\nresponsible for pulling multiple Kafka partitions of the same topic, so that\r\nthese partitions can share output files rather than creating its own output files.\r\n\r\nTo this end, a method `KafkaSource.squeezeMultiWorkUnits()` is added to combine all WorkUnits in a MultiWorkUnit into a single WorkUnit.\r\n\r\n`KafkaExtractor.readRecord()` is rewritten to pull from multiple partitions.\r\n\r\n\r\n- Small modification to complex watermark:\r\n\r\n  - Changed the implementation of `WorkUnit.getLowWatermark` and\r\n`WorkUnit.getExpectedHighWatermark`. The original implementation doesn't work and\r\nthrows \"Not a JsonObject\" exceptions. \r\n  - Changed return type of\r\n`WorkUnitState.getActualHighWatermark()` from `JsonElement` to `String`. In the\r\noriginal implementation, `GSON.toJsonTree(String)` will convert `\"` in the input\r\nstring to `\\\"` in the JsonElement, which will break if we pass this JsonElement to GSON.fromJson.\r\n  - Added method `setWatermarkInterval` to `SourceState`, since we need a way to assign watermarks to workunits.\r\n\r\n- Modified the best fit implementation in KafkaSource, the original one was incorrect.\r\n\r\n- Modified the worst fit implementation in KafkaSource, using a MinMaxPriorityQueue to improve efficiency.\r\n\r\n- Use `com.google.common.net.HostAndPort`\r\n\r\n- Other small refactoring for better readability."},"comment":{"url":"https://api.github.com/repos/linkedin/gobblin/issues/comments/104003396","html_url":"https://github.com/linkedin/gobblin/pull/135#issuecomment-104003396","issue_url":"https://api.github.com/repos/linkedin/gobblin/issues/135","id":104003396,"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},"created_at":"2015-05-20T19:20:20Z","updated_at":"2015-05-20T19:20:20Z","body":"Merged with #141 and addressed comments."}}