Connect config file is not generated accordingly

[shahriar52]

cfk operator version: 2.50
OpenShift version: 4.9

I was trying to create a connect cluster in OpenShift that needs to connect to an on-prem Kafka Brokers. My CRD instance definition is as follows,

apiVersion: platform.confluent.io/v1beta1
kind: Connect
metadata:
  name: connect-ocp
  namespace: kafka-poc
spec:
  license:
    globalLicense: true
  replicas: 1
  image:
    application: confluentinc/cp-server-connect:7.2.2
    init: confluentinc/confluent-init-container:2.5.0
  configOverrides:
    server:
      - group.id=ocp-connect-cluster-01      
  tls:
    autoGeneratedCerts: true
  authorization:
    type: rbac
  dependencies:
    kafka:
      bootstrapEndpoint: server1.domain:9093, server2.domain:9093
      authentication:
        type: plain
        jaasConfig:
          secretRef: credential
      tls:
        enabled: true
        ignoreTrustStoreConfig: true        
    mds:
      endpoint: https://server1.domain:8090, https://server2.domain:8090
      tokenKeyPair:
        secretRef: mds-token
      authentication:
        type: bearer
        bearer:
          secretRef: connect-mds-client
      tls:
        enabled: true
        ignoreTrustStoreConfig: true

It creates the connect-ocp-shared-config configmap as follows,

admin.bootstrap.servers=server1.domain:9073, server2.domain:9073
admin.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required metadataServerUrls="https://server1.domain:8090, https://server2.domain:8090" username="${file:/mnt/secrets/connect-mds-client/bearer.txt:username}" password="${file:/mnt/secrets/connect-mds-client/bearer.txt:password}";
admin.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
admin.sasl.mechanism=OAUTHBEARER
admin.security.protocol=SASL_SSL
bootstrap.servers=server1.domain:9073, server2.domain:9073
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
config.storage.replication.factor=3
config.storage.topic=kafka-poc.connect-ocp-configs
confluent.license=${file:/mnt/secrets/internal-confluent-operator-licensing/license.txt:license}
confluent.metadata.basic.auth.user.info=${file:/mnt/secrets/connect-mds-client/bearer.txt:username}:${file:/mnt/secrets/connect-mds-client/bearer.txt:password}
confluent.metadata.bootstrap.server.urls=https://server1.domain:8090, https://server2.domain:8090
confluent.metadata.http.auth.credentials.provider=BASIC
confluent.topic.replication.factor=3
connector.client.config.override.policy=All
consumer.bootstrap.servers=server1.domain:9073, server2.domain:9073
consumer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required metadataServerUrls="https://server1.domain:8090, https://server2.domain:8090" username="${file:/mnt/secrets/connect-mds-client/bearer.txt:username}" password="${file:/mnt/secrets/connect-mds-client/bearer.txt:password}";
consumer.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
consumer.sasl.mechanism=OAUTHBEARER
consumer.security.protocol=SASL_SSL
group.id=ocp-connect-cluster-01
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
listeners=https://0.0.0.0:8083
listeners.https.ssl.enabled.protocols=TLSv1.2
listeners.https.ssl.key.password=${file:/mnt/sslcerts/jksPassword.txt:jksPassword}
listeners.https.ssl.keystore.location=/mnt/sslcerts/keystore.jks
listeners.https.ssl.keystore.password=${file:/mnt/sslcerts/jksPassword.txt:jksPassword}
listeners.https.ssl.truststore.location=/mnt/sslcerts/truststore.jks
listeners.https.ssl.truststore.password=${file:/mnt/sslcerts/jksPassword.txt:jksPassword}
offset.flush.interval.ms=10000
offset.storage.replication.factor=3
offset.storage.topic=kafka-poc.connect-ocp-offsets
plugin.path=/usr/share/java,/usr/share/confluent-hub-components
producer.bootstrap.servers=server1.domain:9073, server2.domain:9073
producer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required metadataServerUrls="https://server1.domain:8090, https://server2.domain:8090" username="${file:/mnt/secrets/connect-mds-client/bearer.txt:username}" password="${file:/mnt/secrets/connect-mds-client/bearer.txt:password}";
producer.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
producer.sasl.mechanism=OAUTHBEARER
producer.security.protocol=SASL_SSL
public.key.path=/mnt/secrets/mds-token/mdsPublicKey.pem
request.timeout.ms=20000
rest.advertised.listener=https
rest.extension.classes=io.confluent.connect.security.ConnectSecurityExtension
rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler
retry.backoff.ms=500
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required metadataServerUrls="https://server1.domain:8090, https://server2.domain:8090" username="${file:/mnt/secrets/connect-mds-client/bearer.txt:username}" password="${file:/mnt/secrets/connect-mds-client/bearer.txt:password}";
sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
sasl.mechanism=OAUTHBEARER
security.protocol=SASL_SSL
status.storage.replication.factor=3
status.storage.topic=kafka-poc.connect-ocp-status
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

Note that the bootstrap server port is 9073 instead of 9093. Also, the SASL mechanism and JAAS configs are related to OAUTHBEARER and not related to PLAIN.

What am I doing wrong?

[MosheBlumbergX]

There are few issues with the config, it looks like you want to do a SASL_SSL

First:
bootstrapEndpoint | string bootstrapEndpoint specifies the Kafka bootstrap endpoint.
https://docs.confluent.io/operator/current/co-api.html#tag/Connect

You should pass a single string and not a list of brokers/

Second: here is an example for RBAC
https://github.com/confluentinc/confluent-kubernetes-examples/blob/master/security/production-secure-deploy/confluent-platform-production.yaml#L102-L139

If you're looking to do SASL_SSL, you should remove the

  authorization:
    type: rbac