Merge pull request #100 from lovoo/bugfix/start-streams-from-newest

consume streams from newest

kafka/config.go

@@ -12,8 +12,10 @@ func NewConfig() *cluster.Config {
 
 	// consumer configuration
 	config.Consumer.Return.Errors = true
-	config.Consumer.Offsets.Initial = sarama.OffsetOldest
 	config.Consumer.MaxProcessingTime = defaultMaxProcessingTime
+	// this configures the initial offset for streams. Tables are always
+	// consumed from OffsetOldest.
+	config.Consumer.Offsets.Initial = sarama.OffsetNewest
 
 	// producer configuration
 	config.Producer.RequiredAcks = sarama.WaitForLocal

kafka/consumer.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"time"
 
+	"github.com/Shopify/sarama"
 	cluster "github.com/bsm/sarama-cluster"
 	"github.com/lovoo/goka/multierr"
 )
@@ -62,7 +63,11 @@ func NewSaramaConsumer(brokers []string, group string, config *cluster.Config) (
 	if err != nil {
 		return nil, err
 	}
-	c, err := newSimpleConsumer(brokers, events, &config.Config)
+
+	// since simple consumer only handle tables, be sure to start from oldest
+	simpleConfig := config.Config // copy config
+	simpleConfig.Consumer.Offsets.Initial = sarama.OffsetOldest
+	c, err := newSimpleConsumer(brokers, events, &simpleConfig)
 	if err != nil {
 		return nil, err
 	}