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

Issue:
Missing / Invalid Kafka Secret Handling
When configured for `azure` KAFKA_PROVIDER usage, the current implementation handles the case where there are no Kafka Secrets (ie - those labelled with `eventing-kafka.knative.dev/kafka-secret: "true"`).   It does not do so when using standard `kafka` KAFKA_PROVIDER.  

In general we should have the controller be stable when there are no Kafka Secrets, or when the Kafka Secrets are malformed or include the dummy/default data from the yaml.

 It would also be good to document the current handling of edge cases (e.g. deleting secret when KafkaChannels exist, etc) so that we can determine requirements for them in a future issue.

Pull request:
Handle missing or invalid kafka secrets
Fixes #12 

Added minimal stability fixes so that the "Kafka" use case behaves similar to the "EventHubs" use case when the Kafka Secret is missing or invalid.

Also re-added the Makefile which was accidentally removed.  We can refactor it out into hack scripts or something else at some point, but it still has useful development utilities (partial test runs with code coverage and such...)



Patch:
diff --git a/Makefile b/Makefile
new file mode 100644
index 000000000..df2705b9f
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,71 @@
+################################################################################
+##
+##                     Eventing-Kafka Makefile
+##
+################################################################################
+
+# Project Directories
+BUILD_ROOT:=$(shell dirname $(realpath $(lastword $(MAKEFILE_LIST))))
+BUILD_DIR:=$(BUILD_ROOT)/build
+CHANNEL_BUILD_DIR=$(BUILD_DIR)/channel
+COMMON_BUILD_DIR=$(BUILD_DIR)/common
+CONTROLLER_BUILD_DIR=$(BUILD_DIR)/controller
+DISPATCHER_BUILD_DIR=$(BUILD_DIR)/dispatcher
+
+
+#
+# Mod / Fmt / Vet / Lint
+#
+
+mod:
+	@echo 'Ensuring Dependencies'
+	cd $(BUILD_ROOT); go mod vendor; go mod tidy
+
+format:
+	@echo 'Formatting Go Source'
+	go fmt ./...
+
+vet:
+	@echo 'Vetting Packages'
+	go vet -v ./cmd/...
+	go vet -v ./pkg/...
+
+lint:
+	@echo 'Linting Packages'
+	golint ./cmd/...
+	golint ./pkg/...
+
+.PHONY: mod format vet lint
+
+
+#
+# Testing
+#
+
+test-channel:
+	@echo 'Testing Channel'
+	mkdir -p $(CHANNEL_BUILD_DIR)
+	cd $(BUILD_ROOT); go test -v ./pkg/channel/... -coverprofile ${CHANNEL_BUILD_DIR}/coverage.out
+	cd $(BUILD_ROOT); go tool cover -func=${CHANNEL_BUILD_DIR}/coverage.out
+
+test-common:
+	@echo 'Testing Common'
+	mkdir -p $(COMMON_BUILD_DIR)
+	cd $(BUILD_ROOT); go test -v ./pkg/common/... -coverprofile ${COMMON_BUILD_DIR}/coverage.out
+	cd $(BUILD_ROOT); go tool cover -func=${COMMON_BUILD_DIR}/coverage.out
+
+test-controller:
+	@echo 'Testing Controller'
+	mkdir -p $(CONTROLLER_BUILD_DIR)
+	cd $(BUILD_ROOT); go test -v ./pkg/controller/... -coverprofile ${CONTROLLER_BUILD_DIR}/coverage.out
+	cd $(BUILD_ROOT); go tool cover -func=${CONTROLLER_BUILD_DIR}/coverage.out
+
+test-dispatcher:
+	@echo 'Testing Dispatcher'
+	mkdir -p $(DISPATCHER_BUILD_DIR)
+	cd $(BUILD_ROOT); go test -v ./pkg/dispatcher/... -coverprofile ${DISPATCHER_BUILD_DIR}/coverage.out
+	cd $(BUILD_ROOT); go tool cover -func=${DISPATCHER_BUILD_DIR}/coverage.out
+
+test-all: test-channel test-common test-controller test-dispatcher
+
+.PHONY: test-channel test-common test-controller test-dispatcher test-all
diff --git a/OWNERS b/OWNERS
index 7af973255..39aadea27 100644
--- a/OWNERS
+++ b/OWNERS
@@ -5,6 +5,7 @@ approvers:
 - grantr
 - matzew
 - lberk
+- slinkydeveloper
 - davyodom
 - travis-minke-sap
 - phamilton
@@ -15,6 +16,7 @@ reviewers:
 - grantr
 - matzew
 - lberk
+- slinkydeveloper
 - davyodom
 - travis-minke-sap
 - phamilton
diff --git a/pkg/common/kafka/admin/admin_kafka.go b/pkg/common/kafka/admin/admin_kafka.go
index 8d9412d4e..e27e14bc3 100644
--- a/pkg/common/kafka/admin/admin_kafka.go
+++ b/pkg/common/kafka/admin/admin_kafka.go
@@ -56,11 +56,12 @@ func NewKafkaAdminClient(ctx context.Context, namespace string) (AdminClientInte
 		return nil, err
 	}
 
-	// Should Only Be One Kafka Secret In The Namespace
+	//  Currently Only Support One Kafka Secret - Invalid AdminClient For All Other Cases!
 	if len(kafkaSecrets.Items) != 1 {
-		logger.Error(fmt.Sprintf("Expected 1 Kafka Secret But Found %d", len(kafkaSecrets.Items)))
-		err = errors.New(fmt.Sprintf("expected 1 Kafka Secret but found %d", len(kafkaSecrets.Items)))
-		return nil, err
+		logger.Warn(fmt.Sprintf("Expected 1 Kafka Secret But Found %d - Kafka AdminClient Will Not Be Functional!", len(kafkaSecrets.Items)))
+		return &KafkaAdminClient{logger: logger, namespace: namespace}, nil
+	} else {
+		logger.Info("Found 1 Kafka Secret", zap.String("Secret", kafkaSecrets.Items[0].Name))
 	}
 
 	// Extract The Relevant Data From The Kafka Secret
@@ -101,17 +102,31 @@ func NewKafkaAdminClient(ctx context.Context, namespace string) (AdminClientInte
 
 // CreateTopics - Confluent Pass-Through Function
 func (c KafkaAdminClient) CreateTopics(ctx context.Context, topicSpecifications []kafka.TopicSpecification, options ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error) {
-	return c.adminClient.CreateTopics(ctx, topicSpecifications, options...)
+	if c.adminClient == nil {
+		c.logger.Error("Unable To Create Topics Due To Invalid AdminClient - Check Kafka Authorization Secret")
+		return nil, fmt.Errorf("unable to create topics due to invalid AdminClient - check Kafka authorization secrets")
+	} else {
+		return c.adminClient.CreateTopics(ctx, topicSpecifications, options...)
+	}
 }
 
 // DeleteTopics - Confluent Pass-Through Function
 func (c KafkaAdminClient) DeleteTopics(ctx context.Context, topics []string, options ...kafka.DeleteTopicsAdminOption) ([]kafka.TopicResult, error) {
-	return c.adminClient.DeleteTopics(ctx, topics, options...)
+	if c.adminClient == nil {
+		c.logger.Error("Unable To Delete Topics Due To Invalid AdminClient - Check Kafka Authorization Secret")
+		return nil, fmt.Errorf("unable to delete topics due to invalid AdminClient - check Kafka authorization secrets")
+	} else {
+		return c.adminClient.DeleteTopics(ctx, topics, options...)
+	}
 }
 
 // Close - Confluent Pass-Through Function
 func (c KafkaAdminClient) Close() {
-	c.adminClient.Close()
+	if c.adminClient == nil {
+		c.logger.Error("Unable To Close Invalid AdminClient - Check Kafka Authorization Secret")
+	} else {
+		c.adminClient.Close()
+	}
 }
 
 // Get The K8S Secret With Kafka Credentials For The Specified Topic


Test patch:
diff --git a/pkg/common/kafka/admin/admin_kafka_test.go b/pkg/common/kafka/admin/admin_kafka_test.go
index 5ea7b8bf3..ade26afec 100644
--- a/pkg/common/kafka/admin/admin_kafka_test.go
+++ b/pkg/common/kafka/admin/admin_kafka_test.go
@@ -55,8 +55,8 @@ func TestNewKafkaAdminClientNoSecrets(t *testing.T) {
 	adminClient, err := NewKafkaAdminClient(ctx, namespace)
 
 	// Verify The Results
-	assert.NotNil(t, err)
-	assert.Nil(t, adminClient)
+	assert.Nil(t, err)
+	assert.NotNil(t, adminClient)
 }
 
 // Test The Kafka AdminClient CreateTopics() Functionality
@@ -103,6 +103,36 @@ func TestKafkaAdminClientCreateTopics(t *testing.T) {
 	mockConfluentAdminClient.AssertExpectations(t)
 }
 
+// Test The Kafka AdminClient CreateTopics() Without AdminClient Functionality
+func TestKafkaAdminClientCreateTopicsInvalidAdminClient(t *testing.T) {
+
+	// Test Data
+	ctx := context.TODO()
+	topicName := "TestTopicName"
+	topicNumPartitions := 4
+	topicRetentionMillis := int64(3 * constants.MillisPerDay)
+
+	// Create The Kafka TopicSpecification For The Topic/EventHub To Be Created
+	topicSpecifications := []kafka.TopicSpecification{{
+		Topic:         topicName,
+		NumPartitions: topicNumPartitions,
+		Config:        map[string]string{constants.TopicSpecificationConfigRetentionMs: strconv.FormatInt(topicRetentionMillis, 10)},
+	}}
+
+	// Test Logger
+	logger := logtesting.TestLogger(t).Desugar()
+
+	// Create A New Kafka AdminClient To Test
+	adminClient := &KafkaAdminClient{logger: logger}
+
+	// Perform The Test
+	kafkaTopicResults, err := adminClient.CreateTopics(ctx, topicSpecifications)
+
+	// Verify The Results
+	assert.NotNil(t, err)
+	assert.Nil(t, kafkaTopicResults)
+}
+
 // Test The Kafka AdminClient DeleteTopics() Functionality
 func TestKafkaAdminClientDeleteTopics(t *testing.T) {
 
@@ -138,6 +168,27 @@ func TestKafkaAdminClientDeleteTopics(t *testing.T) {
 	mockConfluentAdminClient.AssertExpectations(t)
 }
 
+// Test The Kafka AdminClient DeleteTopics() Without AdminClient Functionality
+func TestKafkaAdminClientDeleteTopicsInvalidAdminClient(t *testing.T) {
+
+	// Test Data
+	ctx := context.TODO()
+	topicName := "TestTopicName"
+
+	// Test Logger
+	logger := logtesting.TestLogger(t).Desugar()
+
+	// Create A New Kafka AdminClient To Test
+	adminClient := &KafkaAdminClient{logger: logger}
+
+	// Perform The Test
+	kafkaTopicResults, err := adminClient.DeleteTopics(ctx, []string{topicName})
+
+	// Verify The Results
+	assert.NotNil(t, err)
+	assert.Nil(t, kafkaTopicResults)
+}
+
 // Test The Kafka AdminClient Close() Functionality
 func TestKafkaAdminClientClose(t *testing.T) {
 
@@ -161,6 +212,21 @@ func TestKafkaAdminClientClose(t *testing.T) {
 	mockConfluentAdminClient.AssertExpectations(t)
 }
 
+// Test The Kafka AdminClient Close() Without AdminClient Functionality
+func TestKafkaAdminClientCloseInvalidAdminClient(t *testing.T) {
+
+	// Test Logger
+	logger := logtesting.TestLogger(t).Desugar()
+
+	// Create A New Kafka AdminClient To Test
+	adminClient := &KafkaAdminClient{logger: logger}
+
+	// Perform The Test
+	adminClient.Close()
+
+	// Nothing To Verify
+}
+
 // Test The Kafka AdminClient GetKafkaSecretName() Functionality
 func TestKafkaAdminClientGetKafkaSecretName(t *testing.T) {