Repository: knative-sandbox/eventing-kafka
Pull request #1045 (2022-01-06 22:28:55)

[release-1.1] Consumer manager use factory merged errors (#1044)
* ConsumerGroup Manager-use factory merged errors channel (fix for #1043)

This is a cherry-pick of #1044

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/1045#" title="Author self-approved">eric-sap</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-1.1/OWNERS)~~ [eric-sap]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | fbbd7fc4208c4a1a3e5fbde839650d0ef5069d2f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1045/pull-knative-sandbox-eventing-kafka-unit-tests/1479218453838565376) | true | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/1045). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1045?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-1.1@6e49fd4`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1045/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff               @@
##             release-1.1    #1045   +/-   ##
==============================================
  Coverage               ?   74.74%           
==============================================
  Files                  ?      121           
  Lines                  ?     5698           
  Branches               ?        0           
==============================================
  Hits                   ?     4259           
  Misses                 ?     1216           
  Partials               ?      223           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1045?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1045?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [6e49fd4...fbbd7fc](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1045?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

travis-minke-sap: /lgtm

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..38ddd4e6d 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -77,7 +77,7 @@ func TestReconfigure(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, group, _, server := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
+			manager, group, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
 			if group != nil {
 				group.On("Close").Return(testCase.closeErr)
 			}
@@ -91,6 +91,10 @@ func TestReconfigure(t *testing.T) {
 				assert.Equal(t, testCase.groupId, err.GroupIds[0])
 			}
 			server.AssertExpectations(t)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -199,7 +203,7 @@ func TestCloseConsumerGroup(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, group, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			manager, group, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, false)
 			if group != nil {
 				group.On("Close").Return(testCase.closeErr)
 				group.On("Errors").Return(make(chan error))
@@ -207,6 +211,10 @@ func TestCloseConsumerGroup(t *testing.T) {
 			err := manager.CloseConsumerGroup(testCase.groupId)
 			assert.Equal(t, testCase.expectErr, err != nil)
 			server.AssertExpectations(t)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -293,7 +301,7 @@ func TestErrors(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, _, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			manager, _, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, false)
 			valid := manager.IsManaged(testCase.groupId)
 			stopped := manager.IsStopped(testCase.groupId)
 			mgrErrors := manager.Errors(testCase.groupId)
@@ -301,6 +309,10 @@ func TestErrors(t *testing.T) {
 			assert.Equal(t, !testCase.expectErr, valid)
 			assert.False(t, stopped) // Not actually using a stopped group in this test
 			server.AssertExpectations(t)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -501,6 +513,10 @@ func TestNotifications(t *testing.T) {
 				group.AssertExpectations(t)
 			}
 			serverHandler.AssertExpectations(t)
+			if managedGrp != nil {
+				close(managedGrp.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -550,7 +566,7 @@ func TestManagerEvents(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, _, _, _ := getManagerWithMockGroup(t, "", false)
+			manager, _, mgdGroup, _ := getManagerWithMockGroup(t, "", false)
 			impl := manager.(*kafkaConsumerGroupManagerImpl)
 
 			var notifyChannels []<-chan ManagerEvent
@@ -592,6 +608,10 @@ func TestManagerEvents(t *testing.T) {
 			manager.ClearNotifications()
 			waitGroup.Wait()
 			assert.Equal(t, testCase.expectClose, count)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -633,10 +653,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..fd3ccb626 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
 		})
 	}
 }
@@ -340,6 +348,8 @@ func TestManagedGroupConsume(t *testing.T) {
 				assert.Nil(t, mgdGroup.saramaGroup.Close()) // Stops the MockConsumerGroup's Consume() call
 			}
 			waitGroup.Wait() // Allows the goroutine with the consume call to finish
+			close(mgdGroup.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -402,7 +412,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 +450,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 +484,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
 		})
 	}
 }