[WIP] Reimplementation

.gitignore

@@ -1,3 +1,5 @@
 /examples/consumergroup/consumergroup
+/examples/kafkaconsumer/kafkaconsumer
 /consumergroup/consumergroup.test
+/kafkaconsumer/kafkaconsumer.test
 /tools/transferoffsets/transferoffsets

.travis.yml

@@ -24,4 +24,4 @@ before_install:
 script:
 - go build ./...
 - go vet ./...
-- GORACE="halt_on_error=1" go test -v -race ./...
+- GORACE="halt_on_error=1" DEBUG=1 go test -v -race ./... -timeout 1m

examples/kafkaconsumer/main.go

@@ -0,0 +1,73 @@
+package main
+
+import (
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"strings"
+	"time"
+
+	"github.com/Shopify/sarama"
+	"github.com/wvanbergen/kafka/kafkaconsumer"
+)
+
+const (
+	DefaultKafkaTopics   = "test_topic"
+	DefaultConsumerGroup = "consumer_example.go"
+)
+
+var (
+	consumerGroup  = flag.String("group", DefaultConsumerGroup, "The name of the consumer group, used for coordination and load balancing")
+	kafkaTopicsCSV = flag.String("topics", DefaultKafkaTopics, "The comma-separated list of topics to consume")
+	zookeeper      = flag.String("zookeeper", "", "A comma-separated Zookeeper connection string (e.g. `zookeeper1.local:2181,zookeeper2.local:2181,zookeeper3.local:2181`)")
+)
+
+func init() {
+	sarama.Logger = log.New(os.Stdout, "[sarama]        ", log.LstdFlags|log.Lshortfile)
+	kafkaconsumer.Logger = log.New(os.Stdout, "[kafkaconsumer] ", log.LstdFlags|log.Lshortfile)
+}
+
+func main() {
+	flag.Parse()
+
+	if *zookeeper == "" {
+		flag.PrintDefaults()
+		os.Exit(1)
+	}
+
+	config := kafkaconsumer.NewConfig()
+	config.Offsets.Initial = sarama.OffsetNewest
+
+	subscription := kafkaconsumer.TopicSubscription(strings.Split(*kafkaTopicsCSV, ",")...)
+	consumer, err := kafkaconsumer.Join(*consumerGroup, subscription, *zookeeper, config)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt)
+	go func() {
+		<-c
+		consumer.Interrupt()
+	}()
+
+	go func() {
+		for err := range consumer.Errors() {
+			log.Println(err)
+		}
+	}()
+
+	var count int64
+	for message := range consumer.Messages() {
+		// Simulate processing that takes some time
+		time.Sleep(10 * time.Millisecond)
+
+		// Acknowledge that we have processed the message
+		consumer.Ack(message)
+
+		count++
+	}
+
+	log.Printf("Processed %d events.", count)
+}

kafkaconsumer/config.go

@@ -0,0 +1,60 @@
+package kafkaconsumer
+
+import (
+	"time"
+
+	"github.com/Shopify/sarama"
+	"github.com/wvanbergen/kazoo-go"
+)
+
+type Config struct {
+	*sarama.Config
+
+	Zookeeper *kazoo.Config
+
+	MaxProcessingTime time.Duration // Time to wait for all the offsets for a partition to be processed after stopping to consume from it. Defaults to 1 minute.
+
+	Offsets struct {
+		Initial int64 // The initial offset method to use if the consumer has no previously stored offset. Must be either sarama.OffsetOldest (default) or sarama.OffsetNewest.
+	}
+}
+
+func NewConfig() *Config {
+	config := &Config{}
+	config.Config = sarama.NewConfig()
+	config.Zookeeper = kazoo.NewConfig()
+	config.Offsets.Initial = sarama.OffsetOldest
+	config.MaxProcessingTime = 60 * time.Second
+
+	return config
+}
+
+func (cgc *Config) Validate() error {
+	if cgc.Zookeeper.Timeout <= 0 {
+		return sarama.ConfigurationError("Zookeeper.Timeout should have a duration > 0")
+	}
+
+	if cgc.MaxProcessingTime <= 0 {
+		return sarama.ConfigurationError("MaxProcessingTime should have a duration > 0")
+	}
+
+	if cgc.Offsets.Initial != sarama.OffsetOldest && cgc.Offsets.Initial != sarama.OffsetNewest {
+		return sarama.ConfigurationError("Offsets.Initial should be sarama.OffsetOldest or sarama.OffsetNewest.")
+	}
+
+	if cgc.Config != nil {
+		if err := cgc.Config.Validate(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+var Logger sarama.StdLogger
+
+func init() {
+	if Logger == nil {
+		Logger = sarama.Logger
+	}
+}