snapshot and batching removal, recovery transactions

kafkamock.go

@@ -76,7 +76,7 @@ type Tester interface {
 // NewKafkaMock returns a new testprocessor mocking every external service
 func NewKafkaMock(t Tester, groupName Group) *KafkaMock {
 	kafkaMock := &KafkaMock{
-		storage:        storage.NewMock(new(codec.Bytes)),
+		storage:        storage.NewMemory(new(codec.Bytes)),
 		t:              t,
 		incomingEvents: make(chan kafka.Event),
 		consumerEvents: make(chan kafka.Event),

mock/storage.go

@@ -92,6 +92,16 @@ func (_mr *_MockStorageRecorder) Iterator() *gomock.Call {
 	return _mr.mock.ctrl.RecordCall(_mr.mock, "Iterator")
 }
 
+func (_m *MockStorage) MarkRecovered() error {
+	ret := _m.ctrl.Call(_m, "MarkRecovered")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+func (_mr *_MockStorageRecorder) MarkRecovered() *gomock.Call {
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "MarkRecovered")
+}
+
 func (_m *MockStorage) Open() error {
 	ret := _m.ctrl.Call(_m, "Open")
 	ret0, _ := ret[0].(error)

options.go

@@ -3,13 +3,13 @@ package goka
 import (
 	"fmt"
 	"path/filepath"
-	"time"
 
 	"github.com/lovoo/goka/kafka"
 	"github.com/lovoo/goka/logger"
 	"github.com/lovoo/goka/storage"
 
 	metrics "github.com/rcrowley/go-metrics"
+	"github.com/syndtr/goleveldb/leveldb"
 )
 
 // UpdateCallback is invoked upon arrival of a message for a table partition.
@@ -25,9 +25,22 @@ type StorageBuilder func(topic string, partition int32, codec Codec, reg metrics
 ///////////////////////////////////////////////////////////////////////////////
 
 const (
+	defaultBaseStoragePath = "/tmp/goka"
+
 	defaultClientID = "goka"
 )
 
+// DefaultProcessorStoragePath is the default path where processor state
+// will be stored.
+func DefaultProcessorStoragePath(group Group) string {
+	return filepath.Join(defaultBaseStoragePath, "processor", string(group))
+}
+
+// DefaultViewStoragePath returns the default path where view state will be stored.
+func DefaultViewStoragePath() string {
+	return filepath.Join(defaultBaseStoragePath, "view")
+}
+
 // DefaultUpdate is the default callback used to update the local storage with
 // from the table topic in Kafka. It is called for every message received
 // during recovery of processors and during the normal operation of views.
@@ -37,6 +50,20 @@ func DefaultUpdate(s storage.Storage, partition int32, key string, value []byte)
 	return s.SetEncoded(key, value)
 }
 
+// DefaultStorageBuilder builds a LevelDB storage with default configuration.
+// The database will be stored in the given path.
+func DefaultStorageBuilder(path string) StorageBuilder {
+	return func(topic string, partition int32, codec Codec, reg metrics.Registry) (storage.Storage, error) {
+		fp := filepath.Join(path, fmt.Sprintf("%s.%d", topic, partition))
+		db, err := leveldb.OpenFile(fp, nil)
+		if err != nil {
+			return nil, fmt.Errorf("error opening leveldb: %v", err)
+		}
+
+		return storage.New(db, codec)
+	}
+}
+
 type consumerBuilder func(brokers []string, group string, registry metrics.Registry) (kafka.Consumer, error)
 type producerBuilder func(brokers []string, registry metrics.Registry) (kafka.Producer, error)
 type topicmgrBuilder func(brokers []string) (kafka.TopicManager, error)
@@ -67,11 +94,9 @@ type poptions struct {
 	log      logger.Logger
 	clientID string
 
-	updateCallback          UpdateCallback
-	storagePath             string
-	storageSnapshotInterval time.Duration
-	registry                metrics.Registry
-	partitionChannelSize    int
+	updateCallback       UpdateCallback
+	registry             metrics.Registry
+	partitionChannelSize int
 
 	builders struct {
 		storage  StorageBuilder
@@ -141,13 +166,6 @@ func WithProducer(p kafka.Producer) ProcessorOption {
 	}
 }
 
-// WithStoragePath defines the base path for the local storage on disk
-func WithStoragePath(storagePath string) ProcessorOption {
-	return func(o *poptions) {
-		o.storagePath = storagePath
-	}
-}
-
 // WithKafkaMetrics sets a go-metrics registry to collect
 // kafka metrics.
 // The metric-points are https://godoc.org/github.com/Shopify/sarama
@@ -166,15 +184,6 @@ func WithPartitionChannelSize(size int) ProcessorOption {
 	}
 }
 
-// WithStorageSnapshotInterval sets the interval in which the storage will snapshot to disk (if it is supported by the storage at all)
-// Greater interval -> less writes to disk, more memory usage
-// Smaller interval -> more writes to disk, less memory usage
-func WithStorageSnapshotInterval(interval time.Duration) ProcessorOption {
-	return func(o *poptions) {
-		o.storageSnapshotInterval = interval
-	}
-}
-
 // WithLogger sets the logger the processor should use. By default, processors
 // use the standard library logger.
 func WithLogger(log logger.Logger) ProcessorOption {
@@ -199,9 +208,9 @@ func (opt *poptions) applyOptions(group string, opts ...ProcessorOption) error {
 		o(opt)
 	}
 
-	// config not set, use default one
+	// StorageBuilder should always be set as a default option in NewProcessor
 	if opt.builders.storage == nil {
-		opt.builders.storage = opt.defaultStorageBuilder
+		return fmt.Errorf("StorageBuilder not set")
 	}
 	if opt.builders.consumer == nil {
 		opt.builders.consumer = defaultConsumerBuilder
@@ -212,9 +221,6 @@ func (opt *poptions) applyOptions(group string, opts ...ProcessorOption) error {
 	if opt.builders.topicmgr == nil {
 		opt.builders.topicmgr = defaultTopicManagerBuilder
 	}
-	if opt.storageSnapshotInterval == 0 {
-		opt.storageSnapshotInterval = storage.DefaultStorageSnapshotInterval
-	}
 
 	// prefix registry
 	opt.gokaRegistry = metrics.NewPrefixedChildRegistry(opt.registry, fmt.Sprintf("goka.processor-%s.", group))
@@ -227,14 +233,6 @@ func (opt *poptions) applyOptions(group string, opts ...ProcessorOption) error {
 	return nil
 }
 
-func (opt *poptions) storagePathForPartition(topic string, partitionID int32) string {
-	return filepath.Join(opt.storagePath, "processor", fmt.Sprintf("%s.%d", topic, partitionID))
-}
-
-func (opt *poptions) defaultStorageBuilder(topic string, partition int32, codec Codec, reg metrics.Registry) (storage.Storage, error) {
-	return storage.New(opt.log, opt.storagePathForPartition(topic, partition), codec, reg, opt.storageSnapshotInterval)
-}
-
 ///////////////////////////////////////////////////////////////////////////////
 // view options
 ///////////////////////////////////////////////////////////////////////////////
@@ -243,13 +241,11 @@ func (opt *poptions) defaultStorageBuilder(topic string, partition int32, codec
 type ViewOption func(*voptions)
 
 type voptions struct {
-	log                     logger.Logger
-	tableCodec              Codec
-	updateCallback          UpdateCallback
-	storagePath             string
-	storageSnapshotInterval time.Duration
-	registry                metrics.Registry
-	partitionChannelSize    int
+	log                  logger.Logger
+	tableCodec           Codec
+	updateCallback       UpdateCallback
+	registry             metrics.Registry
+	partitionChannelSize int
 
 	builders struct {
 		storage  StorageBuilder
@@ -316,22 +312,6 @@ func WithViewTopicManager(tm kafka.TopicManager) ViewOption {
 	}
 }
 
-// WithViewStoragePath defines the base path for the local storage on disk
-func WithViewStoragePath(storagePath string) ViewOption {
-	return func(o *voptions) {
-		o.storagePath = storagePath
-	}
-}
-
-// WithViewStorageSnapshotInterval sets the interval in which the storage will snapshot to disk (if it is supported by the storage at all)
-// Greater interval -> less writes to disk, more memory usage
-// Smaller interval -> more writes to disk, less memory usage
-func WithViewStorageSnapshotInterval(interval time.Duration) ViewOption {
-	return func(o *voptions) {
-		o.storageSnapshotInterval = interval
-	}
-}
-
 // WithViewKafkaMetrics sets a go-metrics registry to collect
 // kafka metrics.
 // The metric-points are https://godoc.org/github.com/Shopify/sarama
@@ -355,9 +335,9 @@ func (opt *voptions) applyOptions(topic Table, opts ...ViewOption) error {
 		o(opt)
 	}
 
-	// config not set, use default one
+	// StorageBuilder should always be set as a default option in NewView
 	if opt.builders.storage == nil {
-		opt.builders.storage = opt.defaultStorageBuilder
+		return fmt.Errorf("StorageBuilder not set")
 	}
 	if opt.builders.consumer == nil {
 		opt.builders.consumer = defaultConsumerBuilder
@@ -374,21 +354,9 @@ func (opt *voptions) applyOptions(topic Table, opts ...ViewOption) error {
 	// prefix registry
 	opt.gokaRegistry = metrics.NewPrefixedChildRegistry(opt.registry, fmt.Sprintf("goka.view-%s.", topic))
 
-	if opt.storageSnapshotInterval == 0 {
-		opt.storageSnapshotInterval = storage.DefaultStorageSnapshotInterval
-	}
-
 	return nil
 }
 
-func (opt *voptions) storagePathForPartition(topic string, partitionID int32) string {
-	return filepath.Join(opt.storagePath, "view", fmt.Sprintf("%s.%d", topic, partitionID))
-}
-
-func (opt *voptions) defaultStorageBuilder(topic string, partition int32, codec Codec, reg metrics.Registry) (storage.Storage, error) {
-	return storage.New(opt.log, opt.storagePathForPartition(topic, partition), codec, reg, opt.storageSnapshotInterval)
-}
-
 ///////////////////////////////////////////////////////////////////////////////
 // emitter options
 ///////////////////////////////////////////////////////////////////////////////

options_test.go

@@ -2,6 +2,7 @@ package goka
 
 import (
 	"fmt"
+	"regexp"
 	"testing"
 
 	"github.com/facebookgo/ensure"
@@ -10,17 +11,12 @@ import (
 func newMockOptions(t *testing.T) *poptions {
 	opts := new(poptions)
 	err := opts.applyOptions("")
+	ensure.Err(t, err, regexp.MustCompile("StorageBuilder not set$"))
+
+	opts.builders.storage = nullStorageBuilder()
+	err = opts.applyOptions("")
 	ensure.Nil(t, err)
-	opts.storagePath = "/tmp/goka-test"
 
 	fmt.Printf("%+v\n", opts)
 	return opts
 }
-
-func TestOptions_storagePathForPartition(t *testing.T) {
-	topic := "test"
-	var id int32
-	opts := newMockOptions(t)
-	path := opts.storagePathForPartition(topic, id)
-	ensure.DeepEqual(t, path, fmt.Sprintf("/tmp/goka-test/processor/%s.%d", topic, id))
-}

partition.go

@@ -52,9 +52,10 @@ type partition struct {
 	readyFlag  int32
 	initialHwm int64
 
-	st      *storageProxy
-	proxy   kafkaProxy
-	process processCallback
+	st            *storageProxy
+	recoveredOnce sync.Once
+	proxy         kafkaProxy
+	process       processCallback
 
 	// metrics
 	registry          metrics.Registry
@@ -89,9 +90,10 @@ func newPartition(log logger.Logger, topic string, cb processCallback, st *stora
 		dying: make(chan bool),
 		done:  make(chan bool),
 
-		st:      st,
-		proxy:   proxy,
-		process: cb,
+		st:            st,
+		recoveredOnce: sync.Once{},
+		proxy:         proxy,
+		process:       cb,
 
 		// metrics
 		registry:               reg,
@@ -289,6 +291,13 @@ func (p *partition) load(catchup bool) error {
 					p.log.Printf("readyFlag was false when EOF arrived")
 					p.mxStatus.Update(partitionRecovered)
 					atomic.StoreInt32(&p.readyFlag, 1)
+					var err error
+					p.recoveredOnce.Do(func() {
+						err = p.st.MarkRecovered()
+					})
+					if err != nil {
+						return fmt.Errorf("error setting recovered: %v", err)
+					}
 				}
 				if catchup {
 					continue
@@ -318,6 +327,13 @@ func (p *partition) load(catchup bool) error {
 					p.mxStatus.Update(partitionRecovering)
 				} else {
 					p.mxStatus.Update(partitionRecovered)
+					var err error
+					p.recoveredOnce.Do(func() {
+						err = p.st.MarkRecovered()
+					})
+					if err != nil {
+						return fmt.Errorf("error setting recovered: %v", err)
+					}
 				}
 			case *kafka.NOP:
 				// don't do anything

partition_test.go

@@ -31,7 +31,7 @@ func newStorageProxy(st storage.Storage, id int32, update UpdateCallback) *stora
 
 func newNullStorageProxy(id int32) *storageProxy {
 	return &storageProxy{
-		Storage:   storage.NewMock(nil),
+		Storage:   storage.NewMemory(nil),
 		partition: id,
 		stateless: true,
 	}
@@ -344,6 +344,7 @@ func TestPartition_runStatefulWithError(t *testing.T) {
 		st.EXPECT().Open().Return(nil),
 		st.EXPECT().GetOffset(int64(-2)).Return(int64(offset), nil),
 		proxy.EXPECT().Add(topic, offset),
+		st.EXPECT().MarkRecovered(),
 		proxy.EXPECT().Remove(topic),
 		proxy.EXPECT().AddGroup(),
 		st.EXPECT().Sync(),
@@ -427,6 +428,7 @@ func TestPartition_loadStateful(t *testing.T) {
 		st.EXPECT().SetEncoded(key, value),
 		st.EXPECT().SetOffset(int64(offset)).Return(nil),
 		st.EXPECT().Sync(),
+		st.EXPECT().MarkRecovered(),
 		proxy.EXPECT().Remove(topic),
 		proxy.EXPECT().AddGroup(),
 		st.EXPECT().Sync(),
@@ -629,6 +631,7 @@ func TestPartition_catchupStateful(t *testing.T) {
 		st.EXPECT().SetEncoded(key, value),
 		st.EXPECT().SetOffset(offset+1).Return(nil),
 		st.EXPECT().Sync(),
+		st.EXPECT().MarkRecovered(),
 		st.EXPECT().SetEncoded(key, value),
 		st.EXPECT().SetOffset(offset+2).Return(nil),
 		st.EXPECT().Sync(),
@@ -749,6 +752,7 @@ func TestPartition_catchupStatefulWithError(t *testing.T) {
 		st.EXPECT().SetEncoded(key, value),
 		st.EXPECT().SetOffset(offset+1).Return(nil),
 		st.EXPECT().Sync(),
+		st.EXPECT().MarkRecovered(),
 		proxy.EXPECT().Remove(topic),
 		st.EXPECT().Close().Return(nil),
 		proxy.EXPECT().Stop(),

processor.go

@@ -63,8 +63,8 @@ func NewProcessor(brokers []string, gg *GroupGraph, options ...ProcessorOption)
 			WithRegistry(metrics.NewRegistry()),
 			WithLogger(logger.Default()),
 			WithUpdateCallback(DefaultUpdate),
-			WithStoragePath("/tmp/goka"),
 			WithPartitionChannelSize(defaultPartitionChannelSize),
+			WithStorageBuilder(DefaultStorageBuilder(DefaultProcessorStoragePath(gg.Group()))),
 		},
 
 		// user-defined options (may overwrite default ones)
@@ -470,7 +470,7 @@ func (g *Processor) newJoinStorage(topic string, id int32, codec Codec, update U
 func (g *Processor) newStorage(topic string, id int32, codec Codec, update UpdateCallback, reg metrics.Registry) (*storageProxy, error) {
 	if g.isStateless() {
 		return &storageProxy{
-			Storage:   storage.NewMock(codec),
+			Storage:   storage.NewMemory(codec),
 			partition: id,
 			stateless: true,
 		}, nil