[release-1.15] Support explicit protocol configuration in KafkaChannel secret

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

@@ -708,6 +708,15 @@
  resource.Auth = &contract.Resource_MultiAuthSecret{
  MultiAuthSecret: auth.MultiSecretReference,
  }
+ } else if auth != nil && auth.VirtualSecret != nil {
+ resource.Auth = &contract.Resource_AuthSecret{
+ AuthSecret: &contract.Reference{
+ Uuid: string(auth.VirtualSecret.UID),
+ Namespace: auth.VirtualSecret.Namespace,
+ Name: auth.VirtualSecret.Name,
+ Version: auth.VirtualSecret.ResourceVersion,
+ },
+ }
  }
 
  if channel.Status.Address != nil && channel.Status.Address.Audience != nil {

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

@@ -240,22 +240,21 @@
  return fmt.Errorf("failed to get secret: %w", err)
  }
 
- if _, ok := secret.Data[security.ProtocolKey]; !ok {
- authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
- if err != nil {
- return err
- }
-
+ authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
+ if err != nil {
+ return err
+ }
+ if authContext.MultiSecretReference != nil {
  resource.Auth = &contract.Resource_MultiAuthSecret{
  MultiAuthSecret: authContext.MultiSecretReference,
  }
- } else {
+ } else if authContext.VirtualSecret != nil {
  resource.Auth = &contract.Resource_AuthSecret{
  AuthSecret: &contract.Reference{
- Uuid: string(secret.UID),
- Namespace: secret.Namespace,
- Name: secret.Name,
- Version: secret.ResourceVersion,
+ Uuid: string(authContext.VirtualSecret.UID),
+ Namespace: authContext.VirtualSecret.Namespace,
+ Name: authContext.VirtualSecret.Name,
+ Version: authContext.VirtualSecret.ResourceVersion,
  },
  }
  }

control-plane/pkg/security/secrets_provider_legacy_channel_secret.go

@@ -29,6 +29,12 @@
  return &NetSpecAuthContext{}, nil
  }
 
+ // Check if the secret is a legacy secret format without the explicit `protocol` key
+ if v, ok := s.Data[ProtocolKey]; ok && len(v) > 0 {
+ // The secret is explicitly using `protocol` configuration, no need to guess it.
+ return &NetSpecAuthContext{VirtualSecret: s}, nil
+ }
+
  protocolStr, protocolContract := getProtocolFromLegacyChannelSecret(s)
 
  virtualSecret := s.DeepCopy()

control-plane/pkg/security/secrets_provider_net_spec.go

@@ -25,6 +25,7 @@ import (
  corev1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  corelisters "k8s.io/client-go/listers/core/v1"
+
  bindings "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
 
  "knative.dev/eventing-kafka-broker/control-plane/pkg/contract"

test/e2e-common.sh

@@ -459,7 +459,6 @@ function create_sasl_secrets() {
  --from-literal=user="my-sasl-user" \
  --from-literal=protocol="SASL_SSL" \
  --from-literal=sasl.mechanism="SCRAM-SHA-512" \
- --from-literal=saslType="SCRAM-SHA-512" \
  --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
 
  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-secret-legacy \
@@ -474,7 +473,6 @@ function create_sasl_secrets() {
  --from-literal=user="my-sasl-user" \
  --from-literal=protocol="SASL_PLAINTEXT" \
  --from-literal=sasl.mechanism="SCRAM-SHA-512" \
- --from-literal=saslType="SCRAM-SHA-512" \
  --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
 
  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-plain-secret-legacy \

test/rekt/features/kafka_source.go

@@ -344,7 +344,7 @@ func kafkaSourceFeature(name string,
  kafkasource.WithSASLEnabled(),
  kafkasource.WithSASLUser(secretName, "user"),
  kafkasource.WithSASLPassword(secretName, "password"),
- kafkasource.WithSASLType(secretName, "saslType"),
+ kafkasource.WithSASLType(secretName, "sasl.mechanism"),
  kafkasource.WithTLSEnabled(),
  kafkasource.WithTLSCACert(secretName, "ca.crt"),
  )

test/rekt/features/kafka_source_create_secrets_after.go

@@ -48,7 +48,7 @@ func CreateSecretsAfterKafkaSource() *feature.Feature {
  kafkasource.WithSASLEnabled(),
  kafkasource.WithSASLUser(saslSecretName, "user"),
  kafkasource.WithSASLPassword(saslSecretName, "password"),
- kafkasource.WithSASLType(saslSecretName, "saslType"),
+ kafkasource.WithSASLType(saslSecretName, "sasl.mechanism"),
  kafkasource.WithTLSEnabled(),
  kafkasource.WithTLSCACert(tlsSecretName, "ca.crt"),
  ))