Shopify/sarama — pull request #36 by burke (2013-10-01 19:36:55+00:00)
Title: MultiConsumer

The goal of this is to be much less wasteful with network traffic. Albert currently has north of 100 connections open to kafka (and uses 40Mb/s because I can't get MaxWaitTime to play nice). With `MultiConsumer` I can drop that to one connection per-broker with sensible timeouts and no message loss on low-volume channels like I'm experiencing with the current version.

There's a hell of a lot of code duplication between this and `Consumer`, and some of the boundaries are weird (`ConsumerConfig`, the `OffsetMethod` semantics in `AddTopicPartition`) -- I wrote most of this while trying to sort out a cross-country move and fight off a cold. It still needs tests, and lots of refactoring, but it does function (well) with this sample code:

```go
package main

import (
"fmt"
"github.com/Shopify/sarama"
)

func main() {

client, err := sarama.NewClient("test", []string{"localhost:9092"}, &sarama.ClientConfig{})
if err != nil {
panic(err)
}

consumer, err := sarama.NewMultiConsumer(client, "mcg", &sarama.ConsumerConfig{
MinFetchSize: 100 * 1024,
MaxWaitTime: 2000,
})
if err != nil {
panic(err)
}
defer consumer.Close()

topics, err := client.Topics()
if err != nil {
panic(err)
}

for _, topic := range topics {
partitions, err := client.Partitions(topic)
if err != nil {
panic(err)
}
for _, partition := range partitions {
consumer.AddTopicPartition(topic, partition, -1, sarama.OffsetMethodOldest)
}
}

for ev := range consumer.Events() {
fmt.Printf("%s:%d: %s\n", ev.Topic, ev.Partition, ev.Value)
}

}
```

/cc @Sirupsen @fw42, no review necessary yet.