Address Kafka review comments

Author: jiangpengcheng

controllers/spec/utils.go

@@ -250,8 +250,7 @@ func makeKafkaInputSpecs(function *v1alpha1.Function) map[string]interface{} {
 		key := kafkaInputSpecKey(function, topic)
 		spec, ok := inputSpecs[key].(map[string]interface{})
 		if !ok {
-			spec = map[string]interface{}{}
-			inputSpecs[key] = spec
+			continue
 		}
 		spec["kafka_schema"] = makeKafkaSchema(function.Spec.Input.SourceSpecs[topic].SchemaType, &schemaConfig)
 	}

controllers/spec/utils_test.go

@@ -181,6 +181,9 @@ func TestConvertFunctionDetailsWithKafkaConfig(t *testing.T) {
 			"orders": {
 				Type: stringPtr("json"),
 			},
+			"unknown-topic": {
+				Type: stringPtr("json"),
+			},
 		},
 		OutputSchemaConfig: &v1alpha1.KafkaSchemaConfig{
 			Type: stringPtr("json"),
@@ -197,7 +200,9 @@ func TestConvertFunctionDetailsWithKafkaConfig(t *testing.T) {
 	assert.Equal(t, "earliest", kafkaConfig["consumer_config"].(map[string]interface{})["auto.offset.reset"])
 	assert.Equal(t, "kafka:9092", kafkaConfig["producer_config"].(map[string]interface{})["bootstrap.servers"])
 	assert.Equal(t, float64(5), kafkaConfig["producer_config"].(map[string]interface{})["linger.ms"])
-	assert.Equal(t, "json", kafkaConfig["input_specs"].(map[string]interface{})["orders"].(map[string]interface{})["kafka_schema"].(map[string]interface{})["type"])
+	inputSpecs := kafkaConfig["input_specs"].(map[string]interface{})
+	assert.Equal(t, "json", inputSpecs["orders"].(map[string]interface{})["kafka_schema"].(map[string]interface{})["type"])
+	assert.NotContains(t, inputSpecs, "unknown-topic")
 	assert.Equal(t, "json", kafkaConfig["output_specs"].(map[string]interface{})["enriched-orders"].(map[string]interface{})["kafka_schema"].(map[string]interface{})["type"])
 }
 

pkg/webhook/validate.go

@@ -461,6 +461,12 @@ func validateFunctionMessaging(spec *v1alpha1.FunctionSpec) *field.Error {
 			return field.Invalid(field.NewPath("spec").Child("kafka", "bootstrapServers"),
 				spec.Kafka.BootstrapServers, "kafka.bootstrapServers needs to be set")
 		}
+		if spec.Kafka.AuthConfig != nil && spec.Kafka.AuthConfig.PlainAuthConfig != nil &&
+			spec.Kafka.AuthConfig.PlainAuthConfig.SecretName == "" {
+			return field.Invalid(field.NewPath("spec").Child("kafka", "authConfig", "plainAuthConfig", "secretName"),
+				spec.Kafka.AuthConfig.PlainAuthConfig.SecretName,
+				"kafka.authConfig.plainAuthConfig.secretName needs to be set")
+		}
 		return nil
 	}
 	return validateMessaging(&spec.Messaging)

pkg/webhook/validate_test.go

@@ -46,6 +46,22 @@ func TestValidateFunctionMessagingRejectsMissingKafkaBootstrapServers(t *testing
 	}
 }
 
+func TestValidateFunctionMessagingRejectsMissingKafkaPlainAuthSecretName(t *testing.T) {
+	err := validateFunctionMessaging(&v1alpha1.FunctionSpec{
+		Messaging: v1alpha1.Messaging{
+			Kafka: &v1alpha1.KafkaMessaging{
+				BootstrapServers: "kafka:9092",
+				AuthConfig: &v1alpha1.KafkaAuthConfig{
+					PlainAuthConfig: &v1alpha1.KafkaPlainAuthConfig{},
+				},
+			},
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "kafka.authConfig.plainAuthConfig.secretName needs to be set") {
+		t.Fatalf("expected missing kafka plain auth secretName error, got %v", err)
+	}
+}
+
 func TestValidateKafkaMessagingUnsupportedRejectsSourceAndSink(t *testing.T) {
 	messaging := &v1alpha1.Messaging{
 		Pulsar: &v1alpha1.PulsarMessaging{PulsarConfig: "pulsar-config"},