diff --git a/asyncevents/cascade.go b/asyncevents/cascade.go index 0fb8196..cdd5e9e 100644 --- a/asyncevents/cascade.go +++ b/asyncevents/cascade.go @@ -5,175 +5,198 @@ import ( "context" "errors" "fmt" - "github.com/IBM/sarama" "log/slog" "os" "strconv" "sync" "time" + + "github.com/IBM/sarama" ) -// SaramaRunner interfaces between [events.Runner] and go-common's -// [SaramaEventsConsumer]. +// CascadingSaramaMessageConsumer cascades messages that failed to be consumed to another +// topic. It is an implementation of [SaramaMessageConsumer]. // -// This means providing Initialize(), Run(), and Terminate() to satisfy events.Runner, while -// under the hood calling SaramaEventConsumer's Run(), and canceling its Context as -// appropriate. -type SaramaRunner struct { - eventsRunner SaramaEventsRunner - cancelCtx context.CancelFunc - cancelMu sync.Mutex +// It also sets an adjustable delay via the "not-before" and "failures" headers so that as +// the message moves from topic to topic, the time between processing is increased according +// to [FailuresToDelay]. +type CascadingSaramaMessageConsumer struct { + Consumer SaramaMessageConsumer + NextTopic string + Producer LimitedAsyncProducer + Logger Logger } -func NewSaramaRunner(eventsRunner SaramaEventsRunner) *SaramaRunner { - return &SaramaRunner{ - eventsRunner: eventsRunner, - } -} +// Consume implements [SaramaMessageConsumer]. +func (c *CascadingSaramaMessageConsumer) Consume(ctx context.Context, + session sarama.ConsumerGroupSession, msg *sarama.ConsumerMessage) (err error) { -// SaramaEventsRunner is implemented by go-common's [SaramaEventsRunner]. -type SaramaEventsRunner interface { - Run(ctx context.Context) error + if err := c.Consumer.Consume(ctx, session, msg); err != nil { + txnErr := c.withTxn(ctx, func() error { + select { + case <-ctx.Done(): + if ctxErr := ctx.Err(); !errors.Is(ctxErr, context.Canceled) { + return ctxErr + } + return nil + case c.Producer.Input() <- c.cascadeMessage(ctx, msg): + c.Logger.Log(ctx, slog.LevelInfo, "cascaded", "from", msg.Topic, "to", c.NextTopic) + return nil + } + }) + if txnErr != nil { + c.Logger.Log(ctx, slog.LevelInfo, "Unable to complete cascading transaction", "error", err) + return err + } + } + return nil } -// SaramaRunnerConfig collects values needed to initialize a SaramaRunner. -// -// This provides isolation for the SaramaRunner from ConfigReporter, -// envconfig, or any of the other options in platform for reading config -// values. -type SaramaRunnerConfig struct { - Brokers []string - GroupID string - Topics []string - MessageConsumer SaramaMessageConsumer - - Sarama *sarama.Config +// withTxn wraps a function with a transaction that is aborted if an error is returned. +func (c *CascadingSaramaMessageConsumer) withTxn(ctx context.Context, f func() error) (err error) { + if err := c.Producer.BeginTxn(); err != nil { + return fmt.Errorf("unable to begin transaction: %w", err) + } + defer func(err *error) { + if err != nil && *err != nil { + if abortErr := c.Producer.AbortTxn(); abortErr != nil { + c.Logger.Log(ctx, slog.LevelInfo, "Unable to abort transaction", "error", abortErr) + } + return + } + if commitErr := c.Producer.CommitTxn(); commitErr != nil { + c.Logger.Log(ctx, slog.LevelInfo, "Unable to commit transaction", "error", commitErr) + } + }(&err) + return f() } -func (r *SaramaRunner) Initialize() error { return nil } - -// Run adapts platform's event.Runner to work with go-common's -// SaramaEventsConsumer. -func (r *SaramaRunner) Run() error { - if r.eventsRunner == nil { - return errors.New("unable to run SaramaRunner, eventsRunner is nil") - } +// cascadeMessage to the next topic. +func (c *CascadingSaramaMessageConsumer) cascadeMessage(ctx context.Context, + msg *sarama.ConsumerMessage) *sarama.ProducerMessage { - r.cancelMu.Lock() - ctx, err := func() (context.Context, error) { - defer r.cancelMu.Unlock() - if r.cancelCtx != nil { - return nil, errors.New("unable to Run SaramaRunner, it's already initialized") - } - var ctx context.Context - ctx, r.cancelCtx = context.WithCancel(context.Background()) - return ctx, nil - }() - if err != nil { - return err + pHeaders := make([]sarama.RecordHeader, len(msg.Headers)) + for idx, header := range msg.Headers { + pHeaders[idx] = *header } - if err := r.eventsRunner.Run(ctx); err != nil { - return fmt.Errorf("unable to Run SaramaRunner: %w", err) + return &sarama.ProducerMessage{ + Key: sarama.ByteEncoder(msg.Key), + Value: sarama.ByteEncoder(msg.Value), + Topic: c.NextTopic, + Headers: c.updateCascadeHeaders(ctx, pHeaders), } - return nil } -// Terminate adapts platform's event.Runner to work with go-common's -// SaramaEventsConsumer. -func (r *SaramaRunner) Terminate() error { - r.cancelMu.Lock() - defer r.cancelMu.Unlock() - if r.cancelCtx == nil { - return errors.New("unable to Terminate SaramaRunner, it's not running") - } - r.cancelCtx() - return nil -} +// updateCascadeHeaders calculates not before and failures header values. +// +// Existing not before and failures headers will be dropped in place of the new ones. +func (c *CascadingSaramaMessageConsumer) updateCascadeHeaders(ctx context.Context, + headers []sarama.RecordHeader) []sarama.RecordHeader { -// CappedExponentialBinaryDelay builds delay functions that use exponential -// binary backoff with a maximum duration. -func CappedExponentialBinaryDelay(cap time.Duration) func(int) time.Duration { - return func(tries int) time.Duration { - b := DelayExponentialBinary(tries) - if b > cap { - return cap + failures := 0 + notBefore := time.Now() + + keep := make([]sarama.RecordHeader, 0, len(headers)) + for _, header := range headers { + switch { + case bytes.Equal(header.Key, HeaderNotBefore): + continue // Drop this header, we'll add a new version below. + case bytes.Equal(header.Key, HeaderFailures): + parsed, err := strconv.ParseInt(string(header.Value), 10, 32) + if err != nil { + c.Logger.Log(ctx, slog.LevelInfo, "Unable to parse consumption failures count", "error", err) + } else { + failures = int(parsed) + notBefore = notBefore.Add(FailuresToDelay[failures]) + } + continue // Drop this header, we'll add a new version below. } - return b + keep = append(keep, header) } + + keep = append(keep, sarama.RecordHeader{ + Key: HeaderNotBefore, + Value: []byte(notBefore.Format(NotBeforeTimeFormat)), + }) + keep = append(keep, sarama.RecordHeader{ + Key: HeaderFailures, + Value: []byte(strconv.Itoa(failures + 1)), + }) + + return keep } -// CascadingSaramaEventsRunner manages multiple sarama consumer groups to execute a -// topic-cascading retry process. -// -// The topic names are generated from Config.Topics combined with Delays. If given a single -// topic "updates", and delays: 0s, 1s, and 5s, then the following topics will be consumed: -// updates, updates-retry-1s, updates-retry-5s. The consumer of the updates-retry-5s topic -// will write failed messages to updates-dead. -// -// The inspiration for this system was drawn from -// https://www.uber.com/blog/reliable-reprocessing/ -type CascadingSaramaEventsRunner struct { - Config SaramaRunnerConfig +// CascadingSaramaEventsManagerConfig for a [CascadingSaramaEventsManager]. +type CascadingSaramaEventsManagerConfig struct { + Consumer SaramaMessageConsumer + + Brokers []string + GroupID string + Topics []string ConsumptionTimeout time.Duration Delays []time.Duration Logger Logger SaramaBuilders SaramaBuilders + Sarama *sarama.Config } -func NewCascadingSaramaEventsRunner(config SaramaRunnerConfig, logger Logger, - delays []time.Duration, consumptionTimeout time.Duration) *CascadingSaramaEventsRunner { - - return &CascadingSaramaEventsRunner{ - Config: config, - Delays: delays, - Logger: logger, - SaramaBuilders: DefaultSaramaBuilders{}, - ConsumptionTimeout: consumptionTimeout, - } +// CascadingSaramaEventsManager manages multiple Sarama consumer groups to execute a +// topic-cascading retry process. It coordinates multiple [SaramaConsumerGroupManager] +// instances to achieve this. +// +// The topics' names are generated from a combination of the configured topics and +// configured delays. For example, if configured with a topic "updates", and delays: 0s, 1s, +// and 5s, then the following topics will be consumed: updates, updates-retry-1s, +// updates-retry-5s. The consumer of the updates-retry-5s topic will write failed messages +// to updates-dead. +// +// The inspiration for this system was drawn from +// https://www.uber.com/blog/reliable-reprocessing/ +type CascadingSaramaEventsManager struct { + CascadingSaramaEventsManagerConfig } -// LimitedAsyncProducer restricts the [sarama.AsyncProducer] interface to ensure that its -// recipient isn't able to call Close(), thereby opening the potential for a panic when -// writing to a closed channel. -type LimitedAsyncProducer interface { - AbortTxn() error - BeginTxn() error - CommitTxn() error - Input() chan<- *sarama.ProducerMessage +func NewCascadingSaramaEventsManager(config CascadingSaramaEventsManagerConfig) *CascadingSaramaEventsManager { + if config.SaramaBuilders == nil { + config.SaramaBuilders = &DefaultSaramaBuilders{} + } + return &CascadingSaramaEventsManager{ + CascadingSaramaEventsManagerConfig: config, + } } -func (r *CascadingSaramaEventsRunner) Run(ctx context.Context) error { - if len(r.Config.Topics) == 0 { +func (c *CascadingSaramaEventsManager) Run(ctx context.Context) error { + if len(c.Topics) == 0 { return errors.New("no topics") } - if len(r.Delays) == 0 { + if len(c.Delays) == 0 { return errors.New("no delays") } producersCtx, cancel := context.WithCancel(ctx) defer cancel() var wg sync.WaitGroup - errs := make(chan error, len(r.Config.Topics)*len(r.Delays)) + errs := make(chan error, len(c.Topics)*len(c.Delays)) defer func() { - r.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsRunner: waiting for consumers") + c.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsManager: waiting for managers") wg.Wait() - r.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsRunner: all consumers returned") + c.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsManager: all managers returned") close(errs) }() - for _, topic := range r.Config.Topics { - for idx, delay := range r.Delays { - producerCfg := r.producerConfig(idx, delay) - // The producer is built here rather than in buildConsumer() to control when - // producer is closed. Were the producer to be closed before consumer.Run() - // returns, it would be possible for consumer to write to the producer's + for _, topic := range c.Topics { + for idx, delay := range c.Delays { + producerCfg := c.producerConfig(idx, delay) + // The producer is built here rather than in buildManager() to control when + // producer is closed. Were the producer to be closed before manager.Run() + // returns, it would be possible for manager to write to the producer's // Inputs() channel, which if closed, would cause a panic. - producer, err := r.SaramaBuilders.NewAsyncProducer(r.Config.Brokers, producerCfg) + producer, err := c.SaramaBuilders.NewAsyncProducer(c.Brokers, producerCfg) if err != nil { - return fmt.Errorf("unable to build async producer %s: %w", r.Config.GroupID, err) + return fmt.Errorf("unable to build async producer %s: %w", c.GroupID, err) } - consumer, err := r.buildConsumer(producersCtx, idx, producer, delay, topic) + manager, err := c.buildManager(producersCtx, idx, producer, delay, topic) if err != nil { return err } @@ -183,35 +206,35 @@ func (r *CascadingSaramaEventsRunner) Run(ctx context.Context) error { defer func() { closeErr := producer.Close() if closeErr != nil { - r.Logger.Log(producersCtx, slog.LevelInfo, "CascadingSaramaEventsRunner: unable to close producer", "error", closeErr) + c.Logger.Log(producersCtx, slog.LevelInfo, "CascadingSaramaEventsManager: unable to close producer", "error", closeErr) } wg.Done() }() - if err := consumer.Run(producersCtx); err != nil { + if err := manager.Run(producersCtx); err != nil { errs <- fmt.Errorf("topics[%q]: %s", topic, err) } - r.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsRunner: consumer go proc returning", "topic", topic) + c.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsManager: manager go proc returning", "topic", topic) }(topic) } } select { case <-ctx.Done(): - r.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsRunner: context is done") + c.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsManager: context is done") return nil case err := <-errs: - r.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsRunner: Run(): error from consumer", "error", err) + c.Logger.Log(ctx, slog.LevelDebug, "CascadingSaramaEventsManager: Run(): error from manager", "error", err) return err } } -func (r *CascadingSaramaEventsRunner) producerConfig(idx int, delay time.Duration) *sarama.Config { - uniqueConfig := *r.Config.Sarama +func (c *CascadingSaramaEventsManager) producerConfig(idx int, delay time.Duration) *sarama.Config { + uniqueConfig := *c.Sarama hostID := os.Getenv("HOSTNAME") // set by default in kubernetes pods if hostID == "" { hostID = fmt.Sprintf("%d-%d", time.Now().UnixNano()/int64(time.Second), os.Getpid()) } - txnID := fmt.Sprintf("%s-%s-%d-%s", r.Config.GroupID, delay.String(), idx, hostID) + txnID := fmt.Sprintf("%s-%s-%d-%s", c.GroupID, delay.String(), idx, hostID) uniqueConfig.Producer.Transaction.ID = txnID uniqueConfig.Producer.Idempotent = true uniqueConfig.Producer.RequiredAcks = sarama.WaitForAll @@ -241,48 +264,50 @@ func (DefaultSaramaBuilders) NewConsumerGroup(brokers []string, groupID string, return sarama.NewConsumerGroup(brokers, groupID, config) } -func (r *CascadingSaramaEventsRunner) buildConsumer(ctx context.Context, idx int, +// buildManager returns a [SaramaConsumerGroupManager] that manages multiple, composed, +// [SaramaMessageConsumer]s according to its topics and delays configuration. +func (c *CascadingSaramaEventsManager) buildManager(ctx context.Context, idx int, producer LimitedAsyncProducer, delay time.Duration, baseTopic string) ( - *SaramaEventsConsumer, error) { + *SaramaConsumerGroupManager, error) { - groupID := r.Config.GroupID + groupID := c.GroupID if delay > 0 { groupID += "-retry-" + delay.String() } - group, err := r.SaramaBuilders.NewConsumerGroup(r.Config.Brokers, groupID, - r.Config.Sarama) + group, err := c.SaramaBuilders.NewConsumerGroup(c.Brokers, groupID, + c.Sarama) if err != nil { return nil, fmt.Errorf("unable to build sarama consumer group %s: %w", groupID, err) } - var consumer = r.Config.MessageConsumer - if len(r.Delays) > 0 { + var consumer SaramaMessageConsumer = c.Consumer + if len(c.Delays) > 0 { nextTopic := baseTopic + "-dead" - if idx+1 < len(r.Delays) { - nextTopic = baseTopic + "-retry-" + r.Delays[idx+1].String() + if idx+1 < len(c.Delays) { + nextTopic = baseTopic + "-retry-" + c.Delays[idx+1].String() } - consumer = &CascadingConsumer{ + consumer = &CascadingSaramaMessageConsumer{ Consumer: consumer, NextTopic: nextTopic, Producer: producer, - Logger: r.Logger, + Logger: c.Logger, } } if delay > 0 { consumer = &NotBeforeConsumer{ Consumer: consumer, - Logger: r.Logger, + Logger: c.Logger, } } - handler := NewSaramaConsumerGroupHandler(r.Logger, consumer, r.ConsumptionTimeout) + handler := NewSaramaConsumerGroupHandler(c.Logger, consumer, c.ConsumptionTimeout) topic := baseTopic if delay > 0 { topic += "-retry-" + delay.String() } - r.Logger.Log(ctx, slog.LevelDebug, "creating consumer", "topic", topic) + c.Logger.Log(ctx, slog.LevelDebug, "creating consumer", "topic", topic) - return NewSaramaEventsConsumer(group, handler, topic), nil + return NewSaramaConsumerGroupManager(group, handler, topic), nil } // NotBeforeConsumer delays consumption until a specified time. @@ -351,108 +376,24 @@ func (c *NotBeforeConsumer) notBeforeFromMsgHeaders(msg *sarama.ConsumerMessage) return time.Time{}, fmt.Errorf("header not found: x-tidepool-not-before") } -// CascadingConsumer cascades messages that failed to be consumed to another topic. -// -// It also sets an adjustable delay via the "not-before" and "failures" headers so that as -// the message moves from topic to topic, the time between processing is increased according -// to [FailuresToDelay]. -type CascadingConsumer struct { - Consumer SaramaMessageConsumer - NextTopic string - Producer LimitedAsyncProducer - Logger Logger -} - -func (c *CascadingConsumer) Consume(ctx context.Context, session sarama.ConsumerGroupSession, - msg *sarama.ConsumerMessage) (err error) { - - if err := c.Consumer.Consume(ctx, session, msg); err != nil { - txnErr := c.withTxn(func() error { - select { - case <-ctx.Done(): - if ctxErr := ctx.Err(); !errors.Is(ctxErr, context.Canceled) { - return ctxErr - } - return nil - case c.Producer.Input() <- c.cascadeMessage(msg): - c.Logger.Log(ctx, slog.LevelInfo, "cascaded", "from", msg.Topic, "to", c.NextTopic) - return nil - } - }) - if txnErr != nil { - c.Logger.Log(ctx, slog.LevelInfo, "Unable to complete cascading transaction", "error", err) - return err - } - } - return nil -} - -// withTxn wraps a function with a transaction that is aborted if an error is returned. -func (c *CascadingConsumer) withTxn(f func() error) (err error) { - if err := c.Producer.BeginTxn(); err != nil { - return fmt.Errorf("unable to begin transaction: %w", err) - } - defer func(err *error) { - if err != nil && *err != nil { - if abortErr := c.Producer.AbortTxn(); abortErr != nil { - c.Logger.Log(nil, slog.LevelInfo, "Unable to abort transaction", "error", abortErr) - } - return - } - if commitErr := c.Producer.CommitTxn(); commitErr != nil { - c.Logger.Log(nil, slog.LevelInfo, "Unable to commit transaction", "error", commitErr) +// CappedExponentialBinaryDelay builds delay functions that use exponential +// binary backoff with a maximum duration. +func CappedExponentialBinaryDelay(cap time.Duration) func(int) time.Duration { + return func(tries int) time.Duration { + b := DelayExponentialBinary(tries) + if b > cap { + return cap } - }(&err) - return f() -} - -// cascadeMessage to the next topic. -func (c *CascadingConsumer) cascadeMessage(msg *sarama.ConsumerMessage) *sarama.ProducerMessage { - pHeaders := make([]sarama.RecordHeader, len(msg.Headers)) - for idx, header := range msg.Headers { - pHeaders[idx] = *header - } - return &sarama.ProducerMessage{ - Key: sarama.ByteEncoder(msg.Key), - Value: sarama.ByteEncoder(msg.Value), - Topic: c.NextTopic, - Headers: c.updateCascadeHeaders(pHeaders), + return b } } -// updateCascadeHeaders calculates not before and failures header values. -// -// Existing not before and failures headers will be dropped in place of the new ones. -func (c *CascadingConsumer) updateCascadeHeaders(headers []sarama.RecordHeader) []sarama.RecordHeader { - failures := 0 - notBefore := time.Now() - - keep := make([]sarama.RecordHeader, 0, len(headers)) - for _, header := range headers { - switch { - case bytes.Equal(header.Key, HeaderNotBefore): - continue // Drop this header, we'll add a new version below. - case bytes.Equal(header.Key, HeaderFailures): - parsed, err := strconv.ParseInt(string(header.Value), 10, 32) - if err != nil { - c.Logger.Log(nil, slog.LevelInfo, "Unable to parse consumption failures count", "error", err) - } else { - failures = int(parsed) - notBefore = notBefore.Add(FailuresToDelay[failures]) - } - continue // Drop this header, we'll add a new version below. - } - keep = append(keep, header) - } - - keep = append(keep, sarama.RecordHeader{ - Key: HeaderNotBefore, - Value: []byte(notBefore.Format(NotBeforeTimeFormat)), - }) - keep = append(keep, sarama.RecordHeader{ - Key: HeaderFailures, - Value: []byte(strconv.Itoa(failures + 1)), - }) - - return keep +// LimitedAsyncProducer restricts the [sarama.AsyncProducer] interface to ensure that its +// recipient isn't able to call Close(), thereby opening the potential for a panic when +// writing to a closed channel. +type LimitedAsyncProducer interface { + AbortTxn() error + BeginTxn() error + CommitTxn() error + Input() chan<- *sarama.ProducerMessage } diff --git a/asyncevents/sarama.go b/asyncevents/sarama.go index 83ad151..0a9a1e3 100644 --- a/asyncevents/sarama.go +++ b/asyncevents/sarama.go @@ -11,29 +11,29 @@ import ( "github.com/IBM/sarama" ) -// SaramaEventsConsumer consumes Kafka messages for asynchronous event +// SaramaConsumerGroupManager manages a consumer group for asynchronous Kafka event // handling. -type SaramaEventsConsumer struct { +type SaramaConsumerGroupManager struct { Handler sarama.ConsumerGroupHandler ConsumerGroup sarama.ConsumerGroup Topics []string } -func NewSaramaEventsConsumer(consumerGroup sarama.ConsumerGroup, - handler sarama.ConsumerGroupHandler, topics ...string) *SaramaEventsConsumer { +func NewSaramaConsumerGroupManager(consumerGroup sarama.ConsumerGroup, + handler sarama.ConsumerGroupHandler, topics ...string) *SaramaConsumerGroupManager { - return &SaramaEventsConsumer{ + return &SaramaConsumerGroupManager{ ConsumerGroup: consumerGroup, Handler: handler, Topics: topics, } } -// Run the consumer, to begin consuming Kafka messages. +// Run the manager, to begin consuming Kafka messages. // // Run is stopped by its context being canceled. When its context is canceled, // it returns nil. -func (p *SaramaEventsConsumer) Run(ctx context.Context) (err error) { +func (p *SaramaConsumerGroupManager) Run(ctx context.Context) (err error) { for { err := p.ConsumerGroup.Consume(ctx, p.Topics, p.Handler) if err != nil { @@ -114,9 +114,9 @@ func (h *SaramaConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe // Close implements sarama.ConsumerGroupHandler. func (h *SaramaConsumerGroupHandler) Close() error { return nil } -// SaramaMessageConsumer processes Kafka messages. +// SaramaMessageConsumer is responsible for the processing of Kafka messages. type SaramaMessageConsumer interface { - // Consume should process a message. + // Consume processes a message. // // Consume is responsible for marking the message consumed, unless the // context is canceled, in which case the caller should retry, or mark the @@ -126,14 +126,11 @@ type SaramaMessageConsumer interface { var ErrRetriesLimitExceeded = errors.New("retry limit exceeded") -// NTimesRetryingConsumer enhances a SaramaMessageConsumer with a finite -// number of immediate retries. +// NTimesRetryingConsumer is a SaramaMessageConsumer with a finite number of retries. // // The delay between each retry can be controlled via the Delay property. If // no Delay property is specified, a delay based on the Fibonacci sequence is // used. -// -// Logger is intentionally minimal. The slog.Log function is used by default. type NTimesRetryingConsumer struct { Times int Consumer SaramaMessageConsumer @@ -148,6 +145,7 @@ type Logger interface { Log(ctx context.Context, level slog.Level, msg string, args ...any) } +// Consume implements SaramaMessageConsumer. func (c *NTimesRetryingConsumer) Consume(ctx context.Context, session sarama.ConsumerGroupSession, message *sarama.ConsumerMessage) (err error) { diff --git a/asyncevents/sarama_test.go b/asyncevents/sarama_test.go index 707aafd..e3bb543 100644 --- a/asyncevents/sarama_test.go +++ b/asyncevents/sarama_test.go @@ -21,7 +21,7 @@ func TestSaramaAsyncEventsConsumerLifecycle(s *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() handler := &nullSaramaConsumerGroupHandler{} - eventsConsumer := NewSaramaEventsConsumer(consumerGroup, handler, topics...) + eventsConsumer := NewSaramaConsumerGroupManager(consumerGroup, handler, topics...) err := launchStart(ctx, t, eventsConsumer) if !errors.Is(err, nil) { t.Errorf("expected nil error, got %v", err) @@ -31,7 +31,7 @@ func TestSaramaAsyncEventsConsumerLifecycle(s *testing.T) { s.Run("reports errors (that aren't context.Canceled)", func(t *testing.T) { consumerGroup := &erroringSaramaConsumerGroup{err: errTest} handler := &nullSaramaConsumerGroupHandler{} - eventsConsumer := NewSaramaEventsConsumer(consumerGroup, handler, topics...) + eventsConsumer := NewSaramaConsumerGroupManager(consumerGroup, handler, topics...) err := launchStart(context.Background(), t, eventsConsumer) if !errors.Is(err, errTest) { t.Errorf("expected %s, got %v", errTest, err) @@ -197,7 +197,7 @@ func ExampleFib() { // It uses a channel to know that the goroutine has seen some amount of CPU // time, which isn't guaranteed to alleviate the race of calling Start, but in // practice seems to be sufficient. Running with -count 10000 had 0 failures. -func launchStart(ctx context.Context, t testing.TB, ec *SaramaEventsConsumer) (err error) { +func launchStart(ctx context.Context, t testing.TB, ec *SaramaConsumerGroupManager) (err error) { t.Helper() runReturned := make(chan error) go func() { diff --git a/asyncevents/startstopadapter.go b/asyncevents/startstopadapter.go new file mode 100644 index 0000000..b6e3745 --- /dev/null +++ b/asyncevents/startstopadapter.go @@ -0,0 +1,124 @@ +package asyncevents + +import ( + "context" + "fmt" + "sync" +) + +// Runner abstracts [SaramaConsumerGroupManager], which is intended to be extended with +// additional capabilities and behaviors. +type Runner interface { + Run(context.Context) error +} + +// BlockingStartStopAdapter for [SaramaConsumerGroupManager] to use Start and Stop methods. +// +// This adapter provides a base for more specific adaptation to adjust behavior to their +// needs. For example [NonBlockingStartStopAdapter] fits well with Uber's fx.Lifecycle, +// while this adapter is more adaptable to platform's event.Runner interface. +type BlockingStartStopAdapter struct { + Runner Runner + + cancelMu sync.Mutex + cancelFunc context.CancelFunc +} + +func NewBlockingStartStopAdapter(runner Runner) *BlockingStartStopAdapter { + return &BlockingStartStopAdapter{ + Runner: runner, + } +} + +func (a *BlockingStartStopAdapter) Start(ctx context.Context) error { + cancelCtx, err := a.init(ctx) + if err != nil { + return err + } + return a.Runner.Run(cancelCtx) +} + +func (a *BlockingStartStopAdapter) init(ctx context.Context) (context.Context, error) { + a.cancelMu.Lock() + defer a.cancelMu.Unlock() + + if a.cancelFunc != nil { + return nil, fmt.Errorf("can't start consumer, it's already running") + } + cancelCtx, cancelFunc := context.WithCancel(ctx) + a.cancelFunc = cancelFunc + return cancelCtx, nil +} + +func (a *BlockingStartStopAdapter) Stop(_ context.Context) error { + a.cancelMu.Lock() + defer a.cancelMu.Unlock() + + if a.cancelFunc == nil { + return fmt.Errorf("can't stop consumer, it's not running") + } + + a.cancelFunc() + a.cancelFunc = nil + + return nil +} + +// NonBlockingStartStopAdapter for [SaramaConsumerGroupManager] for non-blocking Start and +// Stop methods. +// +// To facilitate error reporting during non-blocking operation, a callback can be provided, +// which if defined, will be called with errors that cause a [SaramaConsumerGroupManager]'s +// Run method to return. In addition, when the callback is defined, panics from within Run +// are recovered, converted to errors, and passed to the callback before being discarded. +type NonBlockingStartStopAdapter struct { + *BlockingStartStopAdapter + + onError func(error) +} + +func NewNonBlockingStartStopAdapter(consumer Runner, onError func(error)) *NonBlockingStartStopAdapter { + blocking := NewBlockingStartStopAdapter(consumer) + return &NonBlockingStartStopAdapter{ + BlockingStartStopAdapter: blocking, + onError: onError, + } +} + +func (a *NonBlockingStartStopAdapter) Start(ctx context.Context) error { + go a.start(ctx) + return nil +} + +func (a *NonBlockingStartStopAdapter) start(ctx context.Context) { + defer a.maybeRecover() + + if err := a.BlockingStartStopAdapter.Start(ctx); err != nil { + if a.onError != nil { + a.onError(err) + } + } +} + +// maybeRecover uses a callback, if defined, to process recovered panics. +// +// If the callback isn't defined, the panic will be re-raised. +func (a *NonBlockingStartStopAdapter) maybeRecover() { + if r := recover(); r != nil { + if a.onError != nil { + a.onError(a.wrapPanic(r)) + } else { + panic(r) + } + } +} + +// wrapPanic converts a non-error value to an error for passing to an on-error callback. +// +// Existing error values are wrapped, to allow later unwrapping. +func (a *NonBlockingStartStopAdapter) wrapPanic(r any) error { + if err, ok := r.(error); ok { + return fmt.Errorf("consumer panicked: %w", err) + } + return fmt.Errorf("consumer panicked: %s", r) +} diff --git a/asyncevents/startstopadapter_test.go b/asyncevents/startstopadapter_test.go new file mode 100644 index 0000000..2c54bf3 --- /dev/null +++ b/asyncevents/startstopadapter_test.go @@ -0,0 +1,291 @@ +package asyncevents + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" +) + +func TestBlockingStartStopAdapter_StartWhenAlreadyStarted(t *testing.T) { + ctx := context.Background() + a := NewBlockingStartStopAdapter(newTestConsumer(t, nil)) + if err := a.Start(ctx); err != nil { + t.Fatalf("expected nil error, got %s", err) + } + + got := a.Start(ctx) + if got == nil { + t.Errorf("expected error, got nil") + } else if !strings.Contains(got.Error(), "it's already running") { + t.Errorf("expected error to contain \"it's already running\", got %s", got) + } +} + +func TestBlockingStartStopAdapter_StopWhenNotStarted(t *testing.T) { + ctx := context.Background() + a := NewBlockingStartStopAdapter(newTestConsumer(t, nil)) + + got := a.Stop(ctx) + if got == nil { + t.Errorf("expected error, got nil") + } else if !strings.Contains(got.Error(), "it's not running") { + t.Errorf("expected error to contain \"it's not running\", got %s", got) + } +} + +func TestBlockingStartStopAdapter_StopsWhenCanceled(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + run, started := runUntilCanceled(t) + a := NewBlockingStartStopAdapter(newTestConsumer(t, run)) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + if err := a.Start(ctx); err != nil { + t.Errorf("expected nil error from Start, got %s", err) + } + }() + <-started + + got := a.Stop(ctx) + if got != nil { + t.Errorf("expected nil error from Stop, got %s", got) + } + wg.Wait() +} + +func TestBlockingStartStopAdapter_ReturnsErrorWhenDeadlineExceeded(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond) + t.Cleanup(cancel) + run, started := runUntilCanceled(t) + a := NewBlockingStartStopAdapter(newTestConsumer(t, run)) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + err := a.Start(ctx) + if err == nil || !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("expected deadline exceeded error, got %v", err) + } + }() + <-started + wg.Wait() +} + +func TestNonBlockingStartStopAdapter_StartDoesntBlock(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + run, _ := runUntilCanceled(t) + a := NewNonBlockingStartStopAdapter(newTestConsumer(t, run), nil) + + start := time.Now() + if err := a.Start(ctx); err != nil { + t.Errorf("expected nil error, got %s", err) + } + if time.Since(start) > 10*time.Millisecond { + t.Errorf("expected start to return immediately, but it took longer than expected") + } +} + +func TestNonBlockingStartStopAdapter_StopsWhenCalled(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + run, started := runUntilCanceled(t) + a := NewNonBlockingStartStopAdapter(newTestConsumer(t, run), nil) + + if err := a.Start(ctx); err != nil { + t.Errorf("expected nil error, got %s", err) + } + <-started + + if err := a.Stop(ctx); err != nil { + t.Errorf("expected nil error, got %s", err) + } +} + +func TestNonBlockingStartStopAdapter_UsesCallback(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + run, done := runReturningErrorAfter(t, time.Millisecond) + cbErrs := []error{} + cb := func(err error) { + cbErrs = append(cbErrs, err) + } + a := NewNonBlockingStartStopAdapter(newTestConsumer(t, run), cb) + + if err := a.Start(ctx); err != nil { + t.Errorf("expected nil error, got %s", err) + } + <-done + if len(cbErrs) < 1 { + t.Errorf("expected the callback to be called, but it wasn't") + err := cbErrs[0] + if !strings.Contains(err.Error(), "blowing up") { + t.Errorf("expected callback error to contain \"blowing up\", got: %s", err) + } + } +} + +func TestNonBlockingStartStopAdapter_RecoversPanicsWithCallback(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + run, done := runPanickingAfter(t, time.Millisecond) + cbErrs := []error{} + cb := func(err error) { + cbErrs = append(cbErrs, err) + } + a := NewNonBlockingStartStopAdapter(newTestConsumer(t, run), cb) + + if err := a.Start(ctx); err != nil { + t.Errorf("expected nil error, got %s", err) + } + <-done + // It can take a little while for the panic to propagate even after done is closed. + startedWaiting := time.Now() + for len(cbErrs) == 0 && time.Since(startedWaiting) < time.Second { + time.Sleep(time.Microsecond) + } + if len(cbErrs) < 1 { + t.Errorf("expected the callback to be called, but it wasn't") + err := cbErrs[0] + if !strings.Contains(err.Error(), "panic in the disco") { + t.Errorf("expected callback error to contain \"panic in the disco\", got: %s", err) + } + } +} + +func TestNonBlockingStartStopAdapter_RecoversPanicsWithCallbackIncludingError(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + run, done := runPanickingAfterWithError(t, time.Millisecond) + cbErrs := []error{} + cb := func(err error) { + cbErrs = append(cbErrs, err) + } + a := NewNonBlockingStartStopAdapter(newTestConsumer(t, run), cb) + + if err := a.Start(ctx); err != nil { + t.Errorf("expected nil error, got %s", err) + } + <-done + // It can take a little while for the panic to propagate even after done is closed. + startedWaiting := time.Now() + for len(cbErrs) == 0 && time.Since(startedWaiting) < time.Second { + time.Sleep(time.Microsecond) + } + if len(cbErrs) < 1 { + t.Errorf("expected the callback to be called, but it wasn't") + err := cbErrs[0] + if !strings.Contains(err.Error(), "dignified panic") { + t.Errorf("expected callback error to contain \"dignified panic\", got: %s", err) + } + } +} + +func TestNonBlockingStartStopAdapter_DoesntRecoversPanicsWithoutCallback(t *testing.T) { + // This can't really be tested. Why? Because the panic occurs in a separate goroutine, + // and there's no way for the test to recover that panic. It will just bubble up until + // it crashes the entire test. +} + +type testConsumer struct { + run func(context.Context) error + t testing.TB +} + +func newTestConsumer(t testing.TB, run func(context.Context) error) *testConsumer { + if run == nil { + run = runReturningImmediately + } + return &testConsumer{ + run: run, + t: t, + } +} + +func (c *testConsumer) Run(ctx context.Context) error { + c.t.Helper() + if c.run != nil { + return c.run(ctx) + } + return nil +} + +func runUntilCanceled(t testing.TB) (func(ctx context.Context) error, <-chan struct{}) { + t.Helper() + started := make(chan struct{}) + return func(ctx context.Context) error { + t.Helper() + close(started) + <-ctx.Done() + if err := ctx.Err(); !errors.Is(err, context.Canceled) { + return err + } + return nil + }, started +} + +func runReturningImmediately(ctx context.Context) error { + return nil +} + +func runReturningErrorAfter(t testing.TB, d time.Duration) (func(ctx context.Context) error, <-chan struct{}) { + t.Helper() + started := make(chan struct{}) + return func(ctx context.Context) error { + t.Helper() + defer close(started) + select { + case <-ctx.Done(): + if err := ctx.Err(); !errors.Is(err, context.Canceled) { + return err + } + case <-time.After(d): + return fmt.Errorf("blowing up") + } + return nil + }, started +} + +func runPanickingAfter(t testing.TB, d time.Duration) (func(ctx context.Context) error, <-chan struct{}) { + t.Helper() + started := make(chan struct{}) + return func(ctx context.Context) error { + t.Helper() + defer close(started) + select { + case <-ctx.Done(): + if err := ctx.Err(); !errors.Is(err, context.Canceled) { + return err + } + case <-time.After(d): + panic("panic in the disco!") + } + return nil + }, started +} + +func runPanickingAfterWithError(t testing.TB, d time.Duration) (func(ctx context.Context) error, <-chan struct{}) { + t.Helper() + started := make(chan struct{}) + return func(ctx context.Context) error { + t.Helper() + defer close(started) + select { + case <-ctx.Done(): + if err := ctx.Err(); !errors.Is(err, context.Canceled) { + return err + } + case <-time.After(d): + panic(fmt.Errorf("a dignified panic")) + } + return nil + }, started +}