{"action":"created","comment":{"url":"https://api.github.com/repos/wvanbergen/kafka/pulls/comments/38364000","id":38364000,"diff_hunk":"@@ -0,0 +1,286 @@\n+package kafkaconsumer\n+\n+import (\n+\t\"fmt\"\n+\t\"sync/atomic\"\n+\t\"time\"\n+\n+\t\"github.com/Shopify/sarama\"\n+\t\"github.com/wvanbergen/kazoo-go\"\n+\t\"gopkg.in/tomb.v1\"\n+)\n+\n+// partitionManager manages the consumption of a single partition, and committing\n+// the processed messages to offset storage.\n+type partitionManager struct {\n+\tparent    *consumerManager\n+\tt         tomb.Tomb\n+\tpartition *kazoo.Partition\n+\n+\toffsetManager      sarama.PartitionOffsetManager\n+\tlastConsumedOffset int64\n+\tprocessingDone     chan struct{}\n+}\n+\n+// run implements the main partition manager loop.\n+// 1. Claim the partition in Zookeeper\n+// 2. Determine at what offset to start consuming\n+// 3. Start a consumer for the partition at the inital offset\n+// 4. Transfer messages and errors from the partition consumer to the consumer manager.\n+func (pm *partitionManager) run() {\n+\tdefer pm.t.Done()\n+\n+\tif err := pm.claimPartition(); err != nil {\n+\t\tpm.t.Kill(err)\n+\t\treturn\n+\t}\n+\tdefer pm.releasePartition()\n+\n+\toffsetManager, err := pm.startPartitionOffsetManager()\n+\tif err != nil {\n+\t\tpm.t.Kill(err)\n+\t\treturn\n+\t} else {\n+\t\tpm.offsetManager = offsetManager\n+\t\tdefer pm.closePartitionOffsetManager()\n+\t}\n+\n+\t// We are ignoring metadata for now.\n+\tinitialOffset, _ := offsetManager.NextOffset()\n+\tdefer pm.waitForProcessing()\n+\n+\tpc, err := pm.startPartitionConsumer(initialOffset)\n+\tif err != nil {\n+\t\tpm.t.Kill(err)\n+\t\treturn\n+\t}\n+\tdefer pm.closePartitionConsumer(pc)\n+\n+\tfor {\n+\t\tselect {\n+\t\tcase <-pm.t.Dying():\n+\t\t\treturn\n+\n+\t\tcase msg := <-pc.Messages():\n+\t\t\tselect {\n+\t\t\tcase pm.parent.messages <- msg:\n+\t\t\t\tatomic.StoreInt64(&pm.lastConsumedOffset, msg.Offset)\n+\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn\n+\t\t\t}\n+\n+\t\tcase err := <-offsetManager.Errors():\n+\t\t\tselect {\n+\t\t\tcase pm.parent.errors <- err:\n+\t\t\t\t// Noop?\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn\n+\t\t\t}\n+\n+\t\tcase err := <-pc.Errors():\n+\t\t\tselect {\n+\t\t\tcase pm.parent.errors <- err:\n+\t\t\t\t// Noop?\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn\n+\t\t\t}\n+\t\t}\n+\t}\n+}\n+\n+// interrupt initiates the shutdown procedure for the partition manager, and returns immediately.\n+func (pm *partitionManager) interrupt() {\n+\tpm.t.Kill(nil)\n+}\n+\n+// close starts the shutdown proecure, and waits for it to complete.\n+func (pm *partitionManager) close() error {\n+\tpm.interrupt()\n+\treturn pm.t.Wait()\n+}\n+\n+// ack sets the offset on the partition's offset manager, and signals that\n+// processing done if the offset is equal to the last consumed offset during shutdown.\n+func (pm *partitionManager) ack(offset int64) {\n+\tpm.offsetManager.MarkOffset(offset, \"\")\n+\n+\tif pm.t.Err() != tomb.ErrStillAlive && offset == atomic.LoadInt64(&pm.lastConsumedOffset) {\n+\t\tclose(pm.processingDone)\n+\t}\n+}\n+\n+// claimPartition claims a partition in Zookeeper for this instance.\n+// If the partition is already claimed by someone else, it will wait for the\n+// partition to become available. It will retry errors if they occur.\n+// This method should therefore only return with a nil error value, or\n+// tomb.ErrDying if the partitionManager was interrupted. Any other errors\n+// are not recoverable.\n+func (pm *partitionManager) claimPartition() error {\n+\tpm.logf(\"Trying to claim partition...\")\n+\n+\tfor {\n+\t\towner, changed, err := pm.parent.group.WatchPartitionOwner(pm.partition.Topic().Name, pm.partition.ID)\n+\t\tif err != nil {\n+\t\t\tpm.logf(\"Failed to get partition owner from Zookeeper: %s. Trying again in 1 second...\", err)\n+\t\t\tselect {\n+\t\t\tcase <-time.After(1 * time.Second):\n+\t\t\t\tcontinue\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn tomb.ErrDying\n+\t\t\t}\n+\t\t}\n+\n+\t\tif owner != nil {\n+\t\t\tif owner.ID == pm.parent.instance.ID {\n+\t\t\t\treturn fmt.Errorf(\"The current instance is already the owner of %s. This should not happen.\", pm.partition.Key())\n+\t\t\t}\n+\n+\t\t\tpm.logf(\"Partition is currently claimed by instance %s. Waiting for it to be released...\", owner.ID)\n+\t\t\tselect {\n+\t\t\tcase <-changed:\n+\t\t\t\tcontinue\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn tomb.ErrDying\n+\t\t\t}\n+\n+\t\t} else {\n+\n+\t\t\terr := pm.parent.instance.ClaimPartition(pm.partition.Topic().Name, pm.partition.ID)\n+\t\t\tif err != nil {\n+\t\t\t\tpm.logf(\"Fail to claim partition ownership: %s. Trying again...\", err)\n+\t\t\t\tcontinue\n+\t\t\t}\n+\n+\t\t\tpm.logf(\"Claimed partition ownership\")\n+\t\t\treturn nil\n+\t\t}\n+\t}\n+}\n+\n+// releasePartition releases this instance's claim on this partition in Zookeeper.\n+func (pm *partitionManager) releasePartition() {\n+\tif err := pm.parent.instance.ReleasePartition(pm.partition.Topic().Name, pm.partition.ID); err != nil {\n+\t\tpm.logf(\"FAILED to release partition: %s\", err)\n+\t} else {\n+\t\tpm.logf(\"Released partition.\")\n+\t}\n+}\n+\n+// startPartitionOffsetManager starts a PartitionOffsetManager for the partition, and will\n+// retry any errors. The only error value that can be returned is tomb.ErrDying, which is\n+// returned when the partition manager is interrupted. Any other error should be considered\n+// non-recoverable.\n+func (pm *partitionManager) startPartitionOffsetManager() (sarama.PartitionOffsetManager, error) {\n+\tfor {\n+\t\toffsetManager, err := pm.parent.offsetManager.ManagePartition(pm.partition.Topic().Name, pm.partition.ID)\n+\t\tif err != nil {\n+\t\t\tpm.logf(\"FAILED to start partition offset manager: %s. Trying again in 1 second...\\n\", err)\n+\n+\t\t\tselect {\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn nil, tomb.ErrDying\n+\t\t\tcase <-time.After(1 * time.Second):\n+\t\t\t\tcontinue\n+\t\t\t}\n+\t\t}\n+\n+\t\treturn offsetManager, nil\n+\t}\n+}\n+\n+// closePartitionOffsetManager stops the partition offset manager for this partitions, and will write\n+// any error that may happen to the logger.\n+func (pm *partitionManager) closePartitionOffsetManager() {\n+\tif err := pm.offsetManager.Close(); err != nil {\n+\t\tpm.logf(\"Failed to close partition offset manager: %s\\n\", err)\n+\t}\n+}\n+\n+// startPartitionConsumer starts a sarama consumer for the partition under management.\n+// This function will retry any error that may occur. The error return value is nil once\n+// it successfully has started the partition consumer, or tomb.ErrDying if the partition\n+// manager was interrupted. Any other error is not recoverable.\n+func (pm *partitionManager) startPartitionConsumer(initialOffset int64) (sarama.PartitionConsumer, error) {\n+\tvar (\n+\t\tpc  sarama.PartitionConsumer\n+\t\terr error\n+\t)\n+\n+\tfor {\n+\t\tpc, err = pm.parent.consumer.ConsumePartition(pm.partition.Topic().Name, pm.partition.ID, initialOffset)\n+\t\tswitch err {\n+\t\tcase nil:\n+\t\t\tswitch initialOffset {\n+\t\t\tcase sarama.OffsetNewest:\n+\t\t\t\tpm.logf(\"Started consumer for new messages only.\")\n+\t\t\tcase sarama.OffsetOldest:\n+\t\t\t\tpm.logf(\"Started consumer at the oldest available offset.\")\n+\t\t\tdefault:\n+\t\t\t\tpm.logf(\"Started consumer at offset %d.\", initialOffset)\n+\t\t\t}\n+\n+\t\t\t// We have a valid partition consumer so we can return\n+\t\t\treturn pc, nil\n+\n+\t\tcase sarama.ErrOffsetOutOfRange:\n+\t\t\t// The offset we had on file is too old. Restart with initial offset\n+\t\t\tif pm.parent.config.Offsets.Initial == sarama.OffsetNewest {\n+\t\t\t\tpm.logf(\"Offset %d is no longer available. Trying again with new messages only...\", initialOffset)\n+\t\t\t} else if pm.parent.config.Offsets.Initial == sarama.OffsetOldest {\n+\t\t\t\tpm.logf(\"Offset %d is no longer available. Trying again with he oldest available offset...\", initialOffset)\n+\t\t\t}\n+\t\t\tinitialOffset = pm.parent.config.Offsets.Initial\n+\n+\t\t\tcontinue\n+\n+\t\tdefault:\n+\t\t\t// Assume the problem is temporary; just try again.\n+\t\t\tpm.logf(\"Failed to start consuming partition: %s. Trying again in 1 second...\\n\", err)\n+\t\t\tselect {\n+\t\t\tcase <-pm.t.Dying():\n+\t\t\t\treturn nil, tomb.ErrDying\n+\t\t\tcase <-time.After(1 * time.Second):\n+\t\t\t\tcontinue\n+\t\t\t}\n+\t\t}\n+\n+\t}\n+}\n+\n+// closePartitionConsumer closes the sarama consumer for the partition under management.\n+func (pm *partitionManager) closePartitionConsumer(pc sarama.PartitionConsumer) {\n+\tif err := pc.Close(); err != nil {\n+\t\tpm.logf(\"Failed to close partition consumer: %s\\n\", err)\n+\t}\n+}\n+\n+// waitForProcessing waits for all the messages that were consumed for this partition to be processed.\n+// The processing can take at most MaxProcessingTime time. After that, those messages are consisered\n+// lost and will not be committed. Note that this may cause messages to be processed twice if another\n+// partition consumer resumes consuming from this partition later.\n+func (pm *partitionManager) waitForProcessing() {\n+\tnextOffset, _ := pm.offsetManager.NextOffset()\n+\tlastProcessedOffset := nextOffset - 1","path":"kafkaconsumer/partition_manager.go","position":264,"original_position":264,"commit_id":"9abf9ee238b111b92b8fe8689fbfea583be60952","original_commit_id":"98e6727cfbe25bd87f641c411789cd202953dfc9","user":{"login":"eapache","id":1706826,"avatar_url":"https://avatars.githubusercontent.com/u/1706826?v=3","gravatar_id":"","url":"https://api.github.com/users/eapache","html_url":"https://github.com/eapache","followers_url":"https://api.github.com/users/eapache/followers","following_url":"https://api.github.com/users/eapache/following{/other_user}","gists_url":"https://api.github.com/users/eapache/gists{/gist_id}","starred_url":"https://api.github.com/users/eapache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/eapache/subscriptions","organizations_url":"https://api.github.com/users/eapache/orgs","repos_url":"https://api.github.com/users/eapache/repos","events_url":"https://api.github.com/users/eapache/events{/privacy}","received_events_url":"https://api.github.com/users/eapache/received_events","type":"User","site_admin":false},"body":"seems reasonable to me... early returns would make it much prettier though, e.g. \r\n\r\n```if lastConsumedOffset == -1 {\r\n  return\r\n}\r\n// ...\r\n```","created_at":"2015-08-31T21:50:27Z","updated_at":"2015-08-31T21:50:27Z","html_url":"https://github.com/wvanbergen/kafka/pull/72#discussion_r38364000","pull_request_url":"https://api.github.com/repos/wvanbergen/kafka/pulls/72","_links":{"self":{"href":"https://api.github.com/repos/wvanbergen/kafka/pulls/comments/38364000"},"html":{"href":"https://github.com/wvanbergen/kafka/pull/72#discussion_r38364000"},"pull_request":{"href":"https://api.github.com/repos/wvanbergen/kafka/pulls/72"}}},"pull_request":{"url":"https://api.github.com/repos/wvanbergen/kafka/pulls/72","id":42536185,"html_url":"https://github.com/wvanbergen/kafka/pull/72","diff_url":"https://github.com/wvanbergen/kafka/pull/72.diff","patch_url":"https://github.com/wvanbergen/kafka/pull/72.patch","issue_url":"https://api.github.com/repos/wvanbergen/kafka/issues/72","number":72,"state":"open","locked":false,"title":"[WIP] Reimplementation","user":{"login":"wvanbergen","id":15870,"avatar_url":"https://avatars.githubusercontent.com/u/15870?v=3","gravatar_id":"","url":"https://api.github.com/users/wvanbergen","html_url":"https://github.com/wvanbergen","followers_url":"https://api.github.com/users/wvanbergen/followers","following_url":"https://api.github.com/users/wvanbergen/following{/other_user}","gists_url":"https://api.github.com/users/wvanbergen/gists{/gist_id}","starred_url":"https://api.github.com/users/wvanbergen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wvanbergen/subscriptions","organizations_url":"https://api.github.com/users/wvanbergen/orgs","repos_url":"https://api.github.com/users/wvanbergen/repos","events_url":"https://api.github.com/users/wvanbergen/events{/privacy}","received_events_url":"https://api.github.com/users/wvanbergen/received_events","type":"User","site_admin":false},"body":"This reimplements the kafka high-level consumer. This addresses a couple of limitations of the old implementation, and also makes the code and especially the multithreading model easier to grok (I was young an inexperienced when I wrote the initial implementation ;).\r\n\r\n#### Changes\r\n\r\n1. It now uses Zookeeper to discover all topic/partition metadata, instead of Kafka. The primary reason is that this allows us to set watches, so we can detect changes.\r\n2. Using 1., it now automatically starts to consume new partitions when they become available.\r\n3. When claiming a partition that is still being claimed by someone else, it uses a zookeeper watch to wait until it becomes available.\r\n4. It is smarter during a rebalance operation. Instead of stopping everything and starting everything, it now only stops partition consumers it no longer needs, and only starts partition consumers which are not already there. Because it does all of this in parallel, rebalance operations are a lot faster.\r\n5. Many things are more resilient by implementing retries.\r\n6. It uses a `Subscription` interface to describe what topics to consume. This would allow us to implement a regular expression based black list or white list approach, as well as a static list of topics.\r\n7. It uses an interface for the main `Consumer` type, so unit testing apps that use this library is now possible using dependency injection.\r\n\r\n#### Implementation notes\r\n\r\n- I have implemented it as a separate package, because the API is changed somewhat, and I want to keep the old one around for now.\r\n- We have three main types:\r\n  1. `consumerManager`: runs a goroutine that figures out what partitions to consume, and start/stop partition managers for them. Afterwards it waits for changes in the subscription or changes in this list of running instances to do it again. Implements the `Consumer` interface.\r\n  2. `partitionManager`: runs a goroutine that manages a single `sarama.PartitionConsumer`, claiming the partition in Zookeeper, and managing offsets.\r\n  3. `Subscription`: Describes what partitions the entire group should be consuming, and watches zookeeper for potential changes.\r\n- This depends on some Kazoo changes: https://github.com/wvanbergen/kazoo-go/pull/10\r\n- This depends on sarama's offset manager, which has not yet landed in master: https://github.com/Shopify/sarama/pull/461\r\n\r\n#### TODO\r\n\r\n- [x] Kafka offset management. \r\n- [ ] Zookeeper offset management?\r\n- [x] Add whitelist / blacklist Subscription types. Maybe move the `Subscription` type to Kazoo?\r\n- [ ] Unit tests - feasible now sarama has mock types.\r\n- [x] Functional tests","created_at":"2015-08-16T13:08:58Z","updated_at":"2015-08-31T21:50:27Z","closed_at":null,"merged_at":null,"merge_commit_sha":"c3154093423b66219e6a2617cbe84673edeef130","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/wvanbergen/kafka/pulls/72/commits","review_comments_url":"https://api.github.com/repos/wvanbergen/kafka/pulls/72/comments","review_comment_url":"https://api.github.com/repos/wvanbergen/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/wvanbergen/kafka/issues/72/comments","statuses_url":"https://api.github.com/repos/wvanbergen/kafka/statuses/9abf9ee238b111b92b8fe8689fbfea583be60952","head":{"label":"wvanbergen:refactor","ref":"refactor","sha":"9abf9ee238b111b92b8fe8689fbfea583be60952","user":{"login":"wvanbergen","id":15870,"avatar_url":"https://avatars.githubusercontent.com/u/15870?v=3","gravatar_id":"","url":"https://api.github.com/users/wvanbergen","html_url":"https://github.com/wvanbergen","followers_url":"https://api.github.com/users/wvanbergen/followers","following_url":"https://api.github.com/users/wvanbergen/following{/other_user}","gists_url":"https://api.github.com/users/wvanbergen/gists{/gist_id}","starred_url":"https://api.github.com/users/wvanbergen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wvanbergen/subscriptions","organizations_url":"https://api.github.com/users/wvanbergen/orgs","repos_url":"https://api.github.com/users/wvanbergen/repos","events_url":"https://api.github.com/users/wvanbergen/events{/privacy}","received_events_url":"https://api.github.com/users/wvanbergen/received_events","type":"User","site_admin":false},"repo":{"id":19999011,"name":"kafka","full_name":"wvanbergen/kafka","owner":{"login":"wvanbergen","id":15870,"avatar_url":"https://avatars.githubusercontent.com/u/15870?v=3","gravatar_id":"","url":"https://api.github.com/users/wvanbergen","html_url":"https://github.com/wvanbergen","followers_url":"https://api.github.com/users/wvanbergen/followers","following_url":"https://api.github.com/users/wvanbergen/following{/other_user}","gists_url":"https://api.github.com/users/wvanbergen/gists{/gist_id}","starred_url":"https://api.github.com/users/wvanbergen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wvanbergen/subscriptions","organizations_url":"https://api.github.com/users/wvanbergen/orgs","repos_url":"https://api.github.com/users/wvanbergen/repos","events_url":"https://api.github.com/users/wvanbergen/events{/privacy}","received_events_url":"https://api.github.com/users/wvanbergen/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/wvanbergen/kafka","description":"Load-balancing, resuming Kafka consumer for go, backed by Zookeeper.","fork":false,"url":"https://api.github.com/repos/wvanbergen/kafka","forks_url":"https://api.github.com/repos/wvanbergen/kafka/forks","keys_url":"https://api.github.com/repos/wvanbergen/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/wvanbergen/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/wvanbergen/kafka/teams","hooks_url":"https://api.github.com/repos/wvanbergen/kafka/hooks","issue_events_url":"https://api.github.com/repos/wvanbergen/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/wvanbergen/kafka/events","assignees_url":"https://api.github.com/repos/wvanbergen/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/wvanbergen/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/wvanbergen/kafka/tags","blobs_url":"https://api.github.com/repos/wvanbergen/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/wvanbergen/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/wvanbergen/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/wvanbergen/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/wvanbergen/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/wvanbergen/kafka/languages","stargazers_url":"https://api.github.com/repos/wvanbergen/kafka/stargazers","contributors_url":"https://api.github.com/repos/wvanbergen/kafka/contributors","subscribers_url":"https://api.github.com/repos/wvanbergen/kafka/subscribers","subscription_url":"https://api.github.com/repos/wvanbergen/kafka/subscription","commits_url":"https://api.github.com/repos/wvanbergen/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/wvanbergen/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/wvanbergen/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/wvanbergen/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/wvanbergen/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/wvanbergen/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/wvanbergen/kafka/merges","archive_url":"https://api.github.com/repos/wvanbergen/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/wvanbergen/kafka/downloads","issues_url":"https://api.github.com/repos/wvanbergen/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/wvanbergen/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/wvanbergen/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/wvanbergen/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/wvanbergen/kafka/labels{/name}","releases_url":"https://api.github.com/repos/wvanbergen/kafka/releases{/id}","created_at":"2014-05-20T21:55:17Z","updated_at":"2015-08-23T05:57:45Z","pushed_at":"2015-08-31T21:32:56Z","git_url":"git://github.com/wvanbergen/kafka.git","ssh_url":"git@github.com:wvanbergen/kafka.git","clone_url":"https://github.com/wvanbergen/kafka.git","svn_url":"https://github.com/wvanbergen/kafka","homepage":"","size":1052,"stargazers_count":51,"watchers_count":51,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":27,"mirror_url":null,"open_issues_count":5,"forks":27,"open_issues":5,"watchers":51,"default_branch":"master"}},"base":{"label":"wvanbergen:master","ref":"master","sha":"134f7c372997822bf92c25753451b62148db4310","user":{"login":"wvanbergen","id":15870,"avatar_url":"https://avatars.githubusercontent.com/u/15870?v=3","gravatar_id":"","url":"https://api.github.com/users/wvanbergen","html_url":"https://github.com/wvanbergen","followers_url":"https://api.github.com/users/wvanbergen/followers","following_url":"https://api.github.com/users/wvanbergen/following{/other_user}","gists_url":"https://api.github.com/users/wvanbergen/gists{/gist_id}","starred_url":"https://api.github.com/users/wvanbergen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wvanbergen/subscriptions","organizations_url":"https://api.github.com/users/wvanbergen/orgs","repos_url":"https://api.github.com/users/wvanbergen/repos","events_url":"https://api.github.com/users/wvanbergen/events{/privacy}","received_events_url":"https://api.github.com/users/wvanbergen/received_events","type":"User","site_admin":false},"repo":{"id":19999011,"name":"kafka","full_name":"wvanbergen/kafka","owner":{"login":"wvanbergen","id":15870,"avatar_url":"https://avatars.githubusercontent.com/u/15870?v=3","gravatar_id":"","url":"https://api.github.com/users/wvanbergen","html_url":"https://github.com/wvanbergen","followers_url":"https://api.github.com/users/wvanbergen/followers","following_url":"https://api.github.com/users/wvanbergen/following{/other_user}","gists_url":"https://api.github.com/users/wvanbergen/gists{/gist_id}","starred_url":"https://api.github.com/users/wvanbergen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wvanbergen/subscriptions","organizations_url":"https://api.github.com/users/wvanbergen/orgs","repos_url":"https://api.github.com/users/wvanbergen/repos","events_url":"https://api.github.com/users/wvanbergen/events{/privacy}","received_events_url":"https://api.github.com/users/wvanbergen/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/wvanbergen/kafka","description":"Load-balancing, resuming Kafka consumer for go, backed by Zookeeper.","fork":false,"url":"https://api.github.com/repos/wvanbergen/kafka","forks_url":"https://api.github.com/repos/wvanbergen/kafka/forks","keys_url":"https://api.github.com/repos/wvanbergen/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/wvanbergen/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/wvanbergen/kafka/teams","hooks_url":"https://api.github.com/repos/wvanbergen/kafka/hooks","issue_events_url":"https://api.github.com/repos/wvanbergen/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/wvanbergen/kafka/events","assignees_url":"https://api.github.com/repos/wvanbergen/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/wvanbergen/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/wvanbergen/kafka/tags","blobs_url":"https://api.github.com/repos/wvanbergen/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/wvanbergen/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/wvanbergen/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/wvanbergen/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/wvanbergen/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/wvanbergen/kafka/languages","stargazers_url":"https://api.github.com/repos/wvanbergen/kafka/stargazers","contributors_url":"https://api.github.com/repos/wvanbergen/kafka/contributors","subscribers_url":"https://api.github.com/repos/wvanbergen/kafka/subscribers","subscription_url":"https://api.github.com/repos/wvanbergen/kafka/subscription","commits_url":"https://api.github.com/repos/wvanbergen/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/wvanbergen/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/wvanbergen/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/wvanbergen/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/wvanbergen/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/wvanbergen/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/wvanbergen/kafka/merges","archive_url":"https://api.github.com/repos/wvanbergen/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/wvanbergen/kafka/downloads","issues_url":"https://api.github.com/repos/wvanbergen/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/wvanbergen/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/wvanbergen/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/wvanbergen/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/wvanbergen/kafka/labels{/name}","releases_url":"https://api.github.com/repos/wvanbergen/kafka/releases{/id}","created_at":"2014-05-20T21:55:17Z","updated_at":"2015-08-23T05:57:45Z","pushed_at":"2015-08-31T21:32:56Z","git_url":"git://github.com/wvanbergen/kafka.git","ssh_url":"git@github.com:wvanbergen/kafka.git","clone_url":"https://github.com/wvanbergen/kafka.git","svn_url":"https://github.com/wvanbergen/kafka","homepage":"","size":1052,"stargazers_count":51,"watchers_count":51,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":27,"mirror_url":null,"open_issues_count":5,"forks":27,"open_issues":5,"watchers":51,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/wvanbergen/kafka/pulls/72"},"html":{"href":"https://github.com/wvanbergen/kafka/pull/72"},"issue":{"href":"https://api.github.com/repos/wvanbergen/kafka/issues/72"},"comments":{"href":"https://api.github.com/repos/wvanbergen/kafka/issues/72/comments"},"review_comments":{"href":"https://api.github.com/repos/wvanbergen/kafka/pulls/72/comments"},"review_comment":{"href":"https://api.github.com/repos/wvanbergen/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/wvanbergen/kafka/pulls/72/commits"},"statuses":{"href":"https://api.github.com/repos/wvanbergen/kafka/statuses/9abf9ee238b111b92b8fe8689fbfea583be60952"}}}}