Fix EOF errors when using SASL_SSL PLAIN (#4125)

The consumergroup reconciler was using the legacy secret format resolver
to configure the Sarama client even in the Kafka Broker case.

We can't really add a E2E regression test as strimzi doesn't support
`SASL/PLAIN`: see `strimzi/strimzi-kafka-operator/issues/2221`

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

control-plane/pkg/reconciler/consumergroup/auth.go

@@ -27,32 +27,27 @@ import (
 )
 
 func (r *Reconciler) newAuthSecret(ctx context.Context, cg *kafkainternals.ConsumerGroup) (*corev1.Secret, error) {
-	var secret *corev1.Secret
-
 	if hasSecretSpecConfig(cg.Spec.Template.Spec.Auth) {
 		secret, err := security.Secret(ctx, &SecretSpecSecretLocator{cg}, security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient))
 		if err != nil {
 			return nil, err
 		}
+		return secret, nil
+	}
 
-		authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
-		if err != nil {
-			return nil, err
-		}
-		return authContext.VirtualSecret, nil
-
-	} else if hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
+	if hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
 		auth, err := security.ResolveAuthContextFromNetSpec(r.SecretLister, cg.GetNamespace(), *cg.Spec.Template.Spec.Auth.NetSpec)
 		if err != nil {
 			return nil, err
 		}
-		secret, err = security.Secret(ctx, &NetSpecSecretLocator{cg}, security.NetSpecSecretProviderFunc(auth))
+		secret, err := security.Secret(ctx, &NetSpecSecretLocator{cg}, security.NetSpecSecretProviderFunc(auth))
 		if err != nil {
 			return nil, fmt.Errorf("failed to get secret: %w", err)
 		}
+		return secret, nil
 	}
 
-	return secret, nil
+	return nil, nil
 }
 
 type NetSpecSecretLocator struct {

control-plane/pkg/reconciler/consumergroup/consumergroup.go

@@ -293,14 +293,14 @@ func (r *Reconciler) reconcileStatusSelector(cg *kafkainternals.ConsumerGroup) {
 }
 
 func (r *Reconciler) deleteConsumerGroupMetadata(ctx context.Context, cg *kafkainternals.ConsumerGroup) error {
-	kafakSecret, err := r.newAuthSecret(ctx, cg)
+	kafkaSecret, err := r.newAuthSecret(ctx, cg)
 	if err != nil {
 		return fmt.Errorf("failed to get secret for Kafka cluster auth: %w", err)
 	}
 
 	bootstrapServers := kafka.BootstrapServersArray(cg.Spec.Template.Spec.Configs.Configs["bootstrap.servers"])
 
-	kafkaClusterAdminClient, err := r.GetKafkaClusterAdmin(ctx, bootstrapServers, kafakSecret)
+	kafkaClusterAdminClient, err := r.GetKafkaClusterAdmin(ctx, bootstrapServers, kafkaSecret)
 	if err != nil {
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}