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

Issue:
[distributed kafka-channel] AdminClient Creation Error Handling
**Describe the bug**

Currently the `distributed` KafkaChannel implementation creates a new AdminClient for each reconciliation iteration.  This is required due to Sarama client limitations.  If the AdminClient fails to create correctly, an error is logged but not handled, and reconciliation continues which will result in a **panic** crash/restart of the Controller due to the `Reconciler.adminClient` being **nil**.

 **Expected behavior**

While the underlying reason for the AdminClient creation failure might be unrecoverable (Kafka cluster/config removed), the fact that we could not create an AdminClient should be handled more gracefully and not crash the Controller.  It is ok to fail the reconciliation with an error and re-queue the resource in an endless loop ; )

 **To Reproduce**
1.  Create a KafkaChannel and ensure reconciliation is successful.
2.  Shutdown or delete the backing Kafka cluster so that a new admin client cannot be created.
3.  Attempt to delete the KafkaChannel triggering a reconciliation (finalization) and tail the logs, watch the Controller panic and restart.

 **Knative release version**

v0.24.1 and main/ as of creation of this Issue July 19, 2021.

 **Additional context**
- I am creating this issue based on user feedback from #760 and a brief code-read.
- The fix for this should be easy enough...
    - Return an error from `SetKafkaAdminClient()`
    - Handle this error by returning an **err** to force re-reconciliation.


Pull request description:
[distributed KafkaChannel] Kafka AdminClient Error Handling
Fixes #774

Handle Kafka AdminClient failures by requeuing the KafkaChannel instead of panic/crashing.  Note... KafkaChannel will remain in perpetual reconcile/finalize loop until underlying Kafka connection issue is resolved.  If the user is intending to delete a KafkaChannel instance whose backing Kafka cluster has been torn down, they will need to manually remove the Finalizer from the KafkaChannel to escape this loop.  This is as intended ; )

### Proposed Changes
-🐛  Throw and handle Kafka AdminClient creation error instead of just logging error and continuing until NPE occurs.



Patch:
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler.go b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
index f5f742d8d..06552b715 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
@@ -28,6 +28,10 @@ import (
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/reconciler"
+
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/types"
@@ -40,9 +44,6 @@ import (
 	kafkalisters "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
 	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
-	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/logging"
-	"knative.dev/pkg/reconciler"
 )
 
 // Reconciler Implements controller.Reconciler for KafkaChannel Resources
@@ -79,8 +80,8 @@ var (
 // lightweight REST clients so recreating them isn't a big deal and it simplifies the code significantly to
 // not have to support both use cases.
 //
-func (r *Reconciler) SetKafkaAdminClient(ctx context.Context) {
-	r.ClearKafkaAdminClient(ctx)
+func (r *Reconciler) SetKafkaAdminClient(ctx context.Context) error {
+	_ = r.ClearKafkaAdminClient(ctx) // Attempt to close any lingering connections, ignore errors and continue
 	var err error
 	brokers := strings.Split(r.config.Kafka.Brokers, ",")
 	r.adminClient, err = admin.CreateAdminClient(ctx, brokers, r.config.Sarama.Config, r.adminClientType)
@@ -88,18 +89,21 @@ func (r *Reconciler) SetKafkaAdminClient(ctx context.Context) {
 		logger := logging.FromContext(ctx)
 		logger.Error("Failed To Create Kafka AdminClient", zap.Error(err))
 	}
+	return err
 }
 
 // ClearKafkaAdminClient Clears (Closes) The Reconciler's Kafka AdminClient
-func (r *Reconciler) ClearKafkaAdminClient(ctx context.Context) {
+func (r *Reconciler) ClearKafkaAdminClient(ctx context.Context) error {
+	var err error
 	if r.adminClient != nil {
-		err := r.adminClient.Close()
+		err = r.adminClient.Close()
 		if err != nil {
 			logger := logging.FromContext(ctx)
 			logger.Error("Failed To Close Kafka AdminClient", zap.Error(err))
 		}
 		r.adminClient = nil
 	}
+	return err
 }
 
 // ReconcileKind Implements The Reconciler Interface & Is Responsible For Performing The Reconciliation (Creation)
@@ -126,15 +130,18 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, channel *kafkav1beta1.Ka
 	defer r.adminMutex.Unlock()
 
 	// Create A New Kafka AdminClient For Each Reconciliation Attempt
-	r.SetKafkaAdminClient(ctx)
-	defer r.ClearKafkaAdminClient(ctx)
+	err := r.SetKafkaAdminClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = r.ClearKafkaAdminClient(ctx) }() // Ignore errors as nothing else can be done
 
 	// Reset The Channel's Status Conditions To Unknown (Addressable, Topic, Service, Deployment, etc...)
 	channel.Status.InitializeConditions()
 
 	// Perform The KafkaChannel Reconciliation & Handle Error Response
 	logger.Info("Channel Owned By Controller - Reconciling", zap.Any("Channel.Spec", channel.Spec))
-	err := r.reconcile(ctx, channel)
+	err = r.reconcile(ctx, channel)
 	if err != nil {
 		logger.Error("Failed To Reconcile KafkaChannel", zap.Any("Channel", channel), zap.Error(err))
 		return err
@@ -163,11 +170,14 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, channel *kafkav1beta1.Kaf
 	defer r.adminMutex.Unlock()
 
 	// Create A New Kafka AdminClient For Each Reconciliation Attempt
-	r.SetKafkaAdminClient(ctx)
-	defer r.ClearKafkaAdminClient(ctx)
+	err := r.SetKafkaAdminClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = r.ClearKafkaAdminClient(ctx) }() // Ignore errors as nothing else can be done
 
 	// Finalize The Dispatcher (Manual Finalization Due To Cross-Namespace Ownership)
-	err := r.finalizeDispatcher(ctx, channel)
+	err = r.finalizeDispatcher(ctx, channel)
 	if err != nil {
 		logger.Info("Failed To Finalize KafkaChannel", zap.Error(err))
 		return fmt.Errorf(constants.FinalizationFailedError)


Test patch:
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index 2e2b8b085..b05bad7d3 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -24,6 +24,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -45,6 +46,7 @@ import (
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkaadmintesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/testing"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/types"
+	kafkaadminwrapper "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/wrapper"
 	kafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	controllertesting "knative.dev/eventing-kafka/pkg/channel/distributed/controller/testing"
@@ -77,9 +79,10 @@ func TestSetKafkaAdminClient(t *testing.T) {
 	reconciler := &Reconciler{config: &commonconfig.EventingKafkaConfig{}, adminClient: mockAdminClient1}
 
 	// Perform The Test
-	reconciler.SetKafkaAdminClient(context.TODO())
+	resultErr := reconciler.SetKafkaAdminClient(context.TODO())
 
 	// Verify Results
+	assert.Nil(t, resultErr)
 	assert.True(t, mockAdminClient1.CloseCalled())
 	assert.NotNil(t, reconciler.adminClient)
 	assert.Equal(t, mockAdminClient2, reconciler.adminClient)
@@ -102,9 +105,10 @@ func TestSetKafkaAdminClientError(t *testing.T) {
 	reconciler := &Reconciler{config: &commonconfig.EventingKafkaConfig{}, adminClient: mockAdminClient1}
 
 	// Perform The Test
-	reconciler.SetKafkaAdminClient(context.TODO())
+	resultErr := reconciler.SetKafkaAdminClient(context.TODO())
 
 	// Verify Results
+	assert.NotNil(t, resultErr)
 	assert.Nil(t, reconciler.adminClient)
 }
 
@@ -118,9 +122,10 @@ func TestClearKafkaAdminClient(t *testing.T) {
 	reconciler := &Reconciler{config: &commonconfig.EventingKafkaConfig{}, adminClient: mockAdminClient}
 
 	// Perform The Test
-	reconciler.ClearKafkaAdminClient(context.TODO())
+	resultErr := reconciler.ClearKafkaAdminClient(context.TODO())
 
 	// Verify Results
+	assert.Nil(t, resultErr)
 	assert.True(t, mockAdminClient.CloseCalled())
 	assert.Nil(t, reconciler.adminClient)
 }
@@ -139,9 +144,10 @@ func TestClearKafkaAdminClientError(t *testing.T) {
 	reconciler := &Reconciler{config: &commonconfig.EventingKafkaConfig{}, adminClient: mockAdminClient}
 
 	// Perform The Test
-	reconciler.ClearKafkaAdminClient(context.TODO())
+	resultErr := reconciler.ClearKafkaAdminClient(context.TODO())
 
 	// Verify Results
+	assert.NotNil(t, resultErr)
 	assert.True(t, mockAdminClient.CloseCalled())
 	assert.Nil(t, reconciler.adminClient)
 }
@@ -312,6 +318,30 @@ func TestReconcile(t *testing.T) {
 				event.ReceiverDeploymentReconciliationFailed.String(),
 				"Failed To Reconcile Receiver Deployment: encountered Receiver Deployment with DeletionTimestamp eventing-test-ns/kafkasecret-name-b9176d5f-receiver - potential race condition")),
 
+		//
+		// AdminClient Errors
+		//
+
+		newStableSystemTest("Reconciliation AdminClient Error",
+			withNewAdminClientFn(
+				func(ctx context.Context, brokers []string, config *sarama.Config, adminClientType types.AdminClientType) (types.AdminClientInterface, error) {
+					return nil, fmt.Errorf("test-adminclient-error")
+				},
+			),
+			withSingleErrorEvent("InternalError", "test-adminclient-error"),
+		),
+
+		newDeletedKafkaChannelTest("Finalization AdminClient Error",
+			withoutUpdates,
+			withoutDeletes,
+			withNewAdminClientFn(
+				func(ctx context.Context, brokers []string, config *sarama.Config, adminClientType types.AdminClientType) (types.AdminClientInterface, error) {
+					return nil, fmt.Errorf("test-adminclient-error")
+				},
+			),
+			withSingleErrorEvent("InternalError", "test-adminclient-error"),
+		),
+
 		//
 		// Miscellaneous
 		//
@@ -388,22 +418,27 @@ func TestReconcile(t *testing.T) {
 		newServicePatchFailureTest("Missing Labels", controllertesting.WithoutServiceLabels),
 	}
 
-	// Create A Mock AdminClient
-	mockAdminClient := &controllertesting.MockAdminClient{}
-
-	// Stub The Creation Of AdminClient
-	kafkaadmintesting.StubNewAdminClientFn(kafkaadmintesting.NonValidatingNewAdminClientFn(mockAdminClient))
+	// Restore The NewAdminClient Wrapper On Completion
 	defer kafkaadmintesting.RestoreNewAdminClientFn()
 
 	// Run The TableTest Using The KafkaChannel Reconciler Provided By The Factory
 	logger := logtesting.TestLogger(t)
 	tableTest.Test(t, controllertesting.MakeFactory(func(ctx context.Context, listers *controllertesting.Listers, cmw configmap.Watcher, options map[string]interface{}) controller.Reconciler {
 
-		configOptionsInt, ok := options["configOptions"]
-		if !ok || configOptionsInt == nil {
-			configOptionsInt = []controllertesting.KafkaConfigOption{}
+		// Use Table Row AdminClient If Specified, Otherwise Use Valid Success Mock AdminClient
+		newAdminClientFnInterface, ok := options["newAdminClientFn"]
+		if !ok || newAdminClientFnInterface == nil {
+			newAdminClientFnInterface = kafkaadmintesting.NonValidatingNewAdminClientFn(&controllertesting.MockAdminClient{})
+		}
+		newAdminClientFn := newAdminClientFnInterface.(kafkaadminwrapper.NewAdminClientFnType)
+		kafkaadmintesting.StubNewAdminClientFn(newAdminClientFn)
+
+		// Use Table Row ConfigOptions If Specified, Otherwise Empty
+		configOptionsInterface, ok := options["configOptions"]
+		if !ok || configOptionsInterface == nil {
+			configOptionsInterface = []controllertesting.KafkaConfigOption{}
 		}
-		configOptions := configOptionsInt.([]controllertesting.KafkaConfigOption)
+		configOptions := configOptionsInterface.([]controllertesting.KafkaConfigOption)
 
 		r := &Reconciler{
 			kubeClientset:        kubeclient.Get(ctx),
@@ -665,6 +700,16 @@ func withErrorEvent(finalEvent string, errorEvent string, errorDescription strin
 	}
 }
 
+// withSingleErrorEvent returns a function that replaces the WantEvents field with the specified error event.
+func withSingleErrorEvent(errorEvent string, errorDescription string) testOption {
+	return func(test *TableRow) {
+		test.WantEvents = []string{
+			Eventf(corev1.EventTypeWarning, errorEvent, errorDescription),
+		}
+		test.WantErr = true
+	}
+}
+
 // withFirstEvent returns a function that inserts the given event into the WantEvents slice
 func withFirstEvent(event string) testOption {
 	return func(test *TableRow) {
@@ -711,6 +756,16 @@ func withKafkaChannel(kafkachannel *kafkav1beta1.KafkaChannel) testOption {
 	}
 }
 
+// withNewAdminClientFn returns a function that sets the "adminClient" in the TableRow.OtherTestData
+func withNewAdminClientFn(newAdminClientFn kafkaadminwrapper.NewAdminClientFnType) testOption {
+	return func(test *TableRow) {
+		if test.OtherTestData == nil {
+			test.OtherTestData = make(map[string]interface{})
+		}
+		test.OtherTestData["newAdminClientFn"] = newAdminClientFn
+	}
+}
+
 // getReceiverOnlyKafkaChannel returns a KafkaChannel with all of its non-dispatcher-related ancillary fields
 // ready (Receiver service and deployment, KafkaChannel service, Topic, etc.)
 func getReceiverOnlyKafkaChannel(options ...controllertesting.KafkaChannelOption) *kafkav1beta1.KafkaChannel {
@@ -927,7 +982,7 @@ func newDeletedKafkaChannelTest(name string, options ...testOption) TableRow {
 	service := controllertesting.NewKafkaChannelDispatcherService(controllertesting.WithoutFinalizersService)
 	deployment := controllertesting.NewKafkaChannelDispatcherDeployment(controllertesting.WithoutFinalizersDeployment)
 
-	test := newStableSystemTest("Finalize Deleted KafkaChannel, "+name,
+	test := newStableSystemTest("Finalize Deleted KafkaChannel "+name,
 		withKafkaChannel(controllertesting.NewKafkaChannel(
 			controllertesting.WithInitializedConditions,
 			controllertesting.WithLabels,