simplify SaramaConsumerGroup.Start method

I believe there was confusion about how this is called or should run. These
changes provide equivalent functionality, while being easier to read and
understand.

There are instructions from the library authors here:
https://github.com/IBM/sarama/blob/v1.38.1/consumer_group.go#L19

I think there was confusion around the steps detailed in the link above. So
let me clarify my understanding:

- Canceling the context passed to Consume is how the implementor of the
  sarama.ConsumerGroupHandler tells Consume to return. In other words, when
  we're shutting down, we cancel the context to get Consume to return. I base
  this off of #4 in the link.

- No goroutine needs to be launched on our side, the sarama library will
  launch the goroutines and those goroutines will call our implementation of
  ConsumeClaim(). This comes from #3.

- The warnings about thread-safety apply to our implementation of
  ConsumeClaim(). In other words, we need to be careful that s.consumer isn't
  nil, but since nothing in our code modifies s.consumer after instantiation,
  we're fine. This comes from #3.

- Since our own events.EventConsumer has to be able to respond to Stop()
  calls, we need to be able to break the for-loop in Start(), which is done by
  canceling the context that we pass to Consume(). The context.CancelFunc is
  guarded with a mutex for for extra thread-safety, so that if Stop() is called
  from multiple goroutines, we don't have any race conditions to worry about.

We can reason that the goroutine being previously launched necessary, as there
was only a single goroutine being launched, and same function that launched
it then blocked, waiting for it to return before continuing.

Without the goroutine, the WaitGroup and stop channel and stopOnce are no
longer required.

Author: ewollesen

events/consumer_group.go

@@ -2,7 +2,6 @@ package events
 
 import (
 	"context"
-	"fmt"
 	"log"
 	"sync"
 
@@ -17,10 +16,10 @@ type SaramaConsumerGroup struct {
 	config        *CloudEventsConfig
 	consumerGroup sarama.ConsumerGroup
 	consumer      MessageConsumer
-	stop          chan struct{}
-	stopOnce      *sync.Once
 	topic         string
-	wg            *sync.WaitGroup
+
+	cancelFuncMu sync.Mutex
+	cancelFunc   context.CancelFunc
 }
 
 func NewSaramaConsumerGroup(config *CloudEventsConfig, consumer MessageConsumer) (EventConsumer, error) {
@@ -32,9 +31,6 @@ func NewSaramaConsumerGroup(config *CloudEventsConfig, consumer MessageConsumer)
 		config:   config,
 		consumer: consumer,
 		topic:    config.GetPrefixedTopic(),
-		wg:       &sync.WaitGroup{},
-		stop:     make(chan struct{}),
-		stopOnce: &sync.Once{},
 	}, nil
 }
 
@@ -70,66 +66,43 @@ func (s *SaramaConsumerGroup) Start() error {
 		return err
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	go func() {
-		<-s.stop
-		cancel()
-	}()
-
-	errChan := make(chan error)
-	s.wg.Add(1)
-	go func() {
-		defer s.wg.Done()
-		for {
-			// `Consume` should be called inside an infinite loop, when a
-			// server-side rebalance happens, the consumer session will need to be
-			// recreated to get the new claims
-			if err := s.consumerGroup.Consume(ctx, []string{s.topic}, s); err != nil {
-				log.Printf("Error from consumer: %v", err)
-				// It's not clear whether this condition can be true
-				if err == context.Canceled {
-					err = ErrConsumerStopped
-				}
-				errChan <- err
-				return
-			}
-			// check if context was cancelled, signaling that the consumer should stop
-			if ctx.Err() != nil {
-				errChan <- ErrConsumerStopped
-				return
+	ctx, cancel := s.newContext()
+	defer cancel()
+
+	for {
+		// `Consume` should be called inside an infinite loop, when a
+		// server-side rebalance happens, the consumer session will need to be
+		// recreated to get the new claims
+		if err := s.consumerGroup.Consume(ctx, []string{s.topic}, s); err != nil {
+			log.Printf("Error from consumer: %v", err)
+			if err == context.Canceled {
+				return ErrConsumerStopped
 			}
+			return err
 		}
-	}()
-
-	err := <-errChan
-	if err == ErrConsumerStopped {
-		return err
-	}
-
-	// The consumer group was terminated with an unexpected error.
-	// We need to call stop, so we cancel the context and stop the
-	// go routine so it doesn't leak on restart.
-	if e := s.Stop(); e != nil {
-		err = fmt.Errorf("%w: %s", err, e.Error())
 	}
-
-	return err
 }
 
 func (s *SaramaConsumerGroup) Stop() error {
-	// Initialization failed
-	if s.consumerGroup == nil {
-		return nil
-	}
+	s.cancelFuncMu.Lock()
+	defer s.cancelFuncMu.Unlock()
 
-	// Signal that the consumer group should be terminated
-	s.stopOnce.Do(func() {
-		s.stop <- struct{}{}
-	})
+	if s.cancelFunc != nil {
+		s.cancelFunc()
+		s.cancelFunc = nil
+	}
+	return nil
+}
 
-	// Wait for the consumer group to exit
-	s.wg.Wait()
-	return s.consumerGroup.Close()
+func (s *SaramaConsumerGroup) newContext() (context.Context, context.CancelFunc) {
+	s.cancelFuncMu.Lock()
+	defer s.cancelFuncMu.Unlock()
+	if s.cancelFunc != nil {
+		s.cancelFunc()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancelFunc = cancel
+	return ctx, cancel
 }
 
 func (s *SaramaConsumerGroup) initialize() error {