Repository: knative-sandbox/eventing-kafka (go)

Issue:
Repeated broker failures cause distributed dispatcher to hang
**Describe the bug**
When using the distributed channel type, the 11th time a consumer group fails (typically due to inability to connect to the broker), the dispatcher pod will hang and not re-call the ConsumerGroup's consume() function.  The only way to recover is to restart the dispatcher pod.

 **Expected behavior**
broker connection issues should not break the dispatcher.  When the broker is available again, the ConsumerGroups should resume their consume() functions.

 **To Reproduce**

- Create a distributed channel and subscription and verify basic message functionality
- Monitor the dispatcher log for the channel
- Stop the broker (for example, with a Strimzi kafka backend, removing the Strimzi pods is sufficient)
- Keep the broker down until the 11th time Sarama fails to reconnect to the broker (e.g. keep killing the Strimzi pods).  The dispatcher log (if debug is enabled in the config-logging configmap) will show a "Managed Consume Finished Without Stop" debug message each time this happens (and if Sarama logging is enabled in the config-kafka configmap there will be voluminous output from Sarama).
- After the 11th failure, the dispatcher pod will hang

 **Knative release version**
1.0.0

 **Additional context**
The hanging has been identified as happening in the startExistingConsumerGroup() function when "errorCh <- err" is executed.  The errorCh is never read so when the channel buffer fills, it blocks.


Pull request description:
Consumer manager use factory merged errors
Fixes #1043

- 🐛 ConsumerGroup manager was only reading from the ConsumerGroup error channel and not the merged channel from the KafkaConsumerGroupFactory.  This has been rectified.

Note:  As this is potentially a large problem and the fix is relatively small, this fix should be back-ported


Patch:
diff --git a/pkg/common/consumer/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
index 4ef170cdf..9744b675f 100644
--- a/pkg/common/consumer/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -102,19 +102,20 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 	errorCh := make(chan error, 10)
 	releasedCh := make(chan bool)
 	ctx, cancel := context.WithCancel(context.Background())
+	groupLogger := logger.With(zap.Any("topics", topics), zap.String("groupId", groupID), zap.String("channel", channelRef.String()))
 
 	go func() {
 		// this is a blocking func
 		// do not proceed until the check is done
 		err := c.offsetsChecker.WaitForOffsetsInitialization(ctx, groupID, topics, logger, c.addrs, c.config)
 		if err != nil {
-			logger.Errorw("error while checking if offsets are initialized", zap.Any("topics", topics), zap.String("groupId", groupID), zap.String("channel", channelRef.String()), zap.Error(err))
+			groupLogger.Errorw("error while checking if offsets are initialized", zap.Error(err))
 			errorCh <- err
 			c.enqueue(channelRef)
 			return
 		}
 
-		logger.Debugw("all offsets are initialized", zap.Any("topics", topics), zap.Any("groupID", groupID))
+		groupLogger.Debugw("all offsets are initialized")
 
 		defer func() {
 			close(errorCh)
@@ -123,8 +124,9 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 		for {
 			consumerHandler := NewConsumerHandler(logger, handler, errorCh, options...)
 
-			err := consume(ctx, topics, &consumerHandler)
+			err = consume(ctx, topics, &consumerHandler)
 			if err == sarama.ErrClosedConsumerGroup {
+				groupLogger.Infow("consumer group was closed", zap.Error(err))
 				return
 			}
 			if err != nil {
@@ -133,6 +135,7 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 
 			select {
 			case <-ctx.Done():
+				groupLogger.Info("consumer group terminated gracefully")
 				return
 			default:
 			}
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 882e4143d..293756280 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -285,9 +285,10 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(ctx context.Context,
 		return m.consume(ctx, groupId, topics, handler)
 	}
 
-	// The only thing we really want from the factory is the cancel function for the customConsumerGroup
+	// The only things we really want from the factory are the cancel function for the customConsumerGroup
+	// and the error channel (if the error channel isn't read it will fill up and block between consume calls)
 	customGroup := m.factory.startExistingConsumerGroup(groupId, group, consume, topics, logger, handler, ref, options...)
-	managedGrp := createManagedGroup(ctx, m.logger, group, cancel, customGroup.cancel)
+	managedGrp := createManagedGroup(ctx, m.logger, group, customGroup.Errors(), cancel, customGroup.cancel)
 
 	// Add the Sarama ConsumerGroup we obtained from the factory to the managed group map,
 	// so that it can be stopped and started via control-protocol messages.
diff --git a/pkg/common/consumer/managed_group.go b/pkg/common/consumer/managed_group.go
index b71dc1a79..5cefb7045 100644
--- a/pkg/common/consumer/managed_group.go
+++ b/pkg/common/consumer/managed_group.go
@@ -63,7 +63,7 @@ type managedGroupImpl struct {
 // createManagedGroup associates a Sarama ConsumerGroup and cancel function (usually from the factory)
 // inside a new managedGroup struct.  If a timeout is given (nonzero), the lockId will be reset to an
 // empty string (i.e. "unlocked") after that time has passed.
-func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.ConsumerGroup, cancelErrors func(), cancelConsume func()) managedGroup {
+func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.ConsumerGroup, errors <-chan error, cancelErrors func(), cancelConsume func()) managedGroup {
 
 	managedGrp := &managedGroupImpl{
 		logger:            logger,
@@ -79,8 +79,8 @@ func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.Co
 	managedGrp.lockedBy.Store("")   // Empty token string indicates "unlocked"
 	managedGrp.stopped.Store(false) // A managed group defaults to "started" when created
 
-	// Begin listening on the group's Errors() channel and write them to the managedGroup's errors channel
-	managedGrp.transferErrors(ctx)
+	// Begin listening on the provided errors channel and write them to the managedGroup's errors channel
+	managedGrp.transferErrors(ctx, errors)
 
 	return managedGrp
 }
@@ -142,7 +142,7 @@ func (m *managedGroupImpl) consume(ctx context.Context, topics []string, handler
 		// Call the internal sarama ConsumerGroup's Consume function directly
 		err := m.getSaramaGroup().Consume(ctx, topics, handler)
 		if !m.isStopped() {
-			m.logger.Debug("Managed Consume Finished Without Stop", zap.Error(err))
+			m.logger.Warn("Managed Consume Finished Without Stop", zap.Error(err))
 			// This ConsumerGroup wasn't stopped by the manager, so pass the error along to the caller
 			return err
 		}
@@ -313,28 +313,30 @@ func (m *managedGroupImpl) waitForStart(ctx context.Context) bool {
 	}
 }
 
-// transferErrors starts a goroutine that reads errors from the managedGroup's internal group.Errors() channel
-// and sends them to the m.errors channel.  This is done so that when the group.Errors() channel is closed during
-// a stop ("pause") of the group, the m.errors channel can remain open (so that users of the manager do not
-// receive a closed error channel during stop/start events).
-func (m *managedGroupImpl) transferErrors(ctx context.Context) {
+// transferErrors starts a goroutine that reads errors from the provided errors channel and sends them to the m.errors
+// channel.  Typically, this provided channel comes from the merged errors channel created in the KafkaConsumerGroupFactory
+// This is done so that when the group.Errors() channel is closed during a stop ("pause") of the group, the m.errors
+// channel can remain open (so that users of the manager do not receive a closed error channel during stop/start events).
+func (m *managedGroupImpl) transferErrors(ctx context.Context, errors <-chan error) {
 	go func() {
 		for {
-			m.logger.Debug("Starting managed group error transfer")
-			for groupErr := range m.getSaramaGroup().Errors() {
+			m.logger.Info("Starting managed group error transfer")
+			for groupErr := range errors {
 				m.transferredErrors <- groupErr
 			}
 			if !m.isStopped() {
 				// If the error channel was closed without the consumergroup being marked as stopped,
 				// or if we were unable to wait for the group to be restarted, that is outside
-				// of the manager's responsibility, so we are finished transferring errors.
+				// the manager's responsibility, so we are finished transferring errors.
+				m.logger.Warn("Consumer group's error channel was closed unexpectedly")
 				close(m.transferredErrors)
 				return
 			}
 			// Wait for the manager to restart the Consumer Group before calling m.group.Errors() again
-			m.logger.Debug("Error transfer is waiting for managed group restart")
+			m.logger.Info("Error transfer is waiting for managed group restart")
 			if !m.waitForStart(ctx) {
 				// Abort if the context was canceled
+				m.logger.Info("Wait for managed group restart was canceled")
 				return
 			}
 		}


Test patch:
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index 51b767368..298163aff 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -633,10 +633,7 @@ func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (Kaf
 
 func createMockAndManagedGroups(t *testing.T) (*kafkatesting.MockConsumerGroup, *managedGroupImpl) {
 	mockGroup := kafkatesting.NewMockConsumerGroup()
-	mockGroup.On("Errors").Return(make(chan error))
-	managedGrp := createManagedGroup(context.Background(), logtesting.TestLogger(t).Desugar(), mockGroup, func() {}, func() {})
-	// let the transferErrors function start (otherwise AssertExpectations will randomly fail because Errors() isn't called)
-	time.Sleep(shortTimeout)
+	managedGrp := createManagedGroup(context.Background(), logtesting.TestLogger(t).Desugar(), mockGroup, make(chan error), func() {}, func() {})
 	return mockGroup, managedGrp.(*managedGroupImpl)
 }
 
diff --git a/pkg/common/consumer/managed_group_test.go b/pkg/common/consumer/managed_group_test.go
index 6d684065c..4ee53e6e2 100644
--- a/pkg/common/consumer/managed_group_test.go
+++ b/pkg/common/consumer/managed_group_test.go
@@ -62,8 +62,9 @@ func TestManagedGroup(t *testing.T) {
 			defer cancel()
 
 			mockGroup := kafkatesting.NewMockConsumerGroup()
-			mockGroup.On("Errors").Return(make(chan error))
-			group := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGroup, cancel, func() {}).(*managedGroupImpl)
+			errorChannel := make(chan error)
+			mockGroup.On("Errors").Return(errorChannel)
+			group := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGroup, errorChannel, cancel, func() {}).(*managedGroupImpl)
 			waitGroup := sync.WaitGroup{}
 			assert.False(t, group.isStopped())
 
@@ -85,6 +86,8 @@ func TestManagedGroup(t *testing.T) {
 				cancel()
 			}
 			waitGroup.Wait() // Let the waitForStart function finish
+			close(errorChannel)
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 
@@ -189,6 +192,9 @@ func TestProcessLock(t *testing.T) {
 			assert.Equal(t, testCase.expectUnlock, managedGrp.lockedBy.Load())
 			assert.Equal(t, testCase.expectErrBefore, errBefore)
 			assert.Equal(t, testCase.expectErrAfter, errAfter)
+
+			close(managedGrp.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -280,6 +286,8 @@ func TestResetLockTimer(t *testing.T) {
 				time.Sleep(2 * testCase.timeout)
 				assert.Equal(t, "", managedGrp.lockedBy.Load())
 			}
+			close(managedGrp.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -402,7 +410,8 @@ func TestStopStart(t *testing.T) {
 			assert.Equal(t, testCase.errStopping, err != nil)
 
 			mockGroup.AssertExpectations(t)
-
+			close(mgdGroup.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -439,6 +448,8 @@ func TestClose(t *testing.T) {
 			assert.Nil(t, err)
 			assert.Equal(t, testCase.cancel, cancelConsumeCalled)
 			assert.Equal(t, testCase.cancel, cancelErrorsCalled)
+			close(mgdGroup.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -471,51 +482,53 @@ func TestTransferErrors(t *testing.T) {
 		t.Run(testCase.name, func(t *testing.T) {
 			time.Sleep(time.Millisecond)
 
+			// errorChan simulates the merged error channel usually provided via customConsumerGroup.handlerErrorChannel
+			errorChan := make(chan error)
+
+			// mockGrp represents the internal Sarama ConsumerGroup
 			mockGrp := kafkatesting.NewMockConsumerGroup()
-			managedGrp := managedGroupImpl{
-				logger:            logtesting.TestLogger(t).Desugar(),
-				saramaGroup:       mockGrp,
-				transferredErrors: make(chan error),
-				groupMutex:        sync.RWMutex{},
-			}
-			managedGrp.lockedBy.Store("")
-			managedGrp.stopped.Store(false)
 
-			mockGrp.On("Errors").Return(mockGrp.ErrorChan)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
-			managedGrp.transferErrors(ctx)
 
-			mockGrp.ErrorChan <- fmt.Errorf("test-error")
+			// Call the transferErrors function (via createManagedGroup) with the simulated customConsumerGroup error channel
+			managedGrp := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGrp, errorChan, func() {}, func() {})
+			managedGrpImpl := managedGrp.(*managedGroupImpl)
+
+			// Send an error to the simulated customConsumerGroup error channel
+			errorChan <- fmt.Errorf("test-error")
+
+			// Verify that the error appears in the managed group's error channel
 			err := <-managedGrp.errors()
 			assert.NotNil(t, err)
 			assert.Equal(t, "test-error", err.Error())
+
 			if testCase.stopGroup {
-				managedGrp.createRestartChannel()
+				managedGrpImpl.createRestartChannel()
 			}
 			if testCase.cancel {
 				cancel()
 			}
-			close(mockGrp.ErrorChan)
 			mockGrp.AssertExpectations(t)
 
-			time.Sleep(shortTimeout) // Let the error handling loop move forward
 			if testCase.startGroup {
+				time.Sleep(shortTimeout) // Let the error handling loop move forward
 				// Simulate the effects of startConsumerGroup (new ConsumerGroup, same managedConsumerGroup)
 				mockGrp = kafkatesting.NewMockConsumerGroup()
-				mockGrp.On("Errors").Return(mockGrp.ErrorChan)
-				managedGrp.saramaGroup = mockGrp
-				managedGrp.closeRestartChannel()
+				managedGrpImpl.saramaGroup = mockGrp
+				managedGrpImpl.closeRestartChannel()
 
 				time.Sleep(shortTimeout) // Let the waitForStart function finish
-				// Verify that errors work again after restart
-				mockGrp.ErrorChan <- fmt.Errorf("test-error-2")
+
+				// Verify that error transfer continues to work after the restart
+				errorChan <- fmt.Errorf("test-error-2")
 				err = <-managedGrp.errors()
 				assert.NotNil(t, err)
 				assert.Equal(t, "test-error-2", err.Error())
-				close(mockGrp.ErrorChan)
 				mockGrp.AssertExpectations(t)
 			}
+			close(managedGrp.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }