Merge pull request #8057 from alenkacz/av/fix-broker-tls

k8s: broker_tls is enabled only if internal listener has tls

src/go/k8s/pkg/resources/certmanager/type_helpers.go

@@ -138,6 +138,7 @@ type apiCertificates struct {
 	clientCertificates []resources.Resource
 	rootResources      []resources.Resource
 	tlsEnabled         bool
+	internalTLSEnabled bool
 	// true if api is using our own generated self-signed issuer
 	selfSignedNodeCertificate bool
 
@@ -251,6 +252,9 @@ func (cc *ClusterCertificates) prepareAPI(
 		return tlsDisabledAPICertificates(), nil
 	}
 	result := tlsEnabledAPICertificates(cc.pandaCluster.Namespace)
+	if internalTLSListener != nil {
+		result.internalTLSEnabled = true
+	}
 
 	// TODO(#3550): Do not create rootIssuer if nodeSecretRef is passed and mTLS is disabled
 	toApplyRoot, rootIssuerRef := prepareRoot(rootCertSuffix, cc.client, cc.pandaCluster, cc.scheme, cc.logger)
@@ -669,7 +673,7 @@ func (cc *ClusterCertificates) GetTLSConfig(
 
 // KafkaClientBrokerTLS returns configuration to connect to kafka api with tls
 func (cc *ClusterCertificates) KafkaClientBrokerTLS(mountPoints *resourcetypes.TLSMountPoints) *config.ServerTLS {
-	if !cc.kafkaAPI.tlsEnabled {
+	if !cc.kafkaAPI.internalTLSEnabled {
 		return nil
 	}
 	result := config.ServerTLS{

src/go/k8s/pkg/resources/statefulset.go

@@ -194,7 +194,7 @@ func (r *StatefulSetResource) Ensure(ctx context.Context) error {
 	return r.handleScaling(ctx)
 }
 
-// GetCentralizedConfigurationHashFromCluster retrieves the current centralized configuratino hash from the statefulset
+// GetCentralizedConfigurationHashFromCluster retrieves the current centralized configuration hash from the statefulset
 func (r *StatefulSetResource) GetCentralizedConfigurationHashFromCluster(
 	ctx context.Context,
 ) (string, error) {