KafkaConnector example for scram-sha-512-auth

[andreyolv]

Suggestion / Problem
Can you please add a KafkaConnector example for scram-sha-512-auth in repository? Thank you very much

Documentation Link
examples/security/scram-sha-512-auth/

I am with the following problem, and an example would help a lot
Output of command: k get kafkaconnectors inventory-connector -o yaml
connectorStatus:
connector:
state: RUNNING
worker_id: 10.244.1.9:8083
name: inventory-connector
tasks:
- id: 0
state: FAILED
trace: |
org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata
worker_id: 10.244.1.9:8083
type: source
observedGeneration: 3
tasksMax: 1
topics: []

[scholzj]

The Connector does not configure any authentication. That is done in the KafkaConnect resource which already has an example. So I'm not sure what is there to add.

[andreyolv]

I am using KafkaConnector with class io.debezium.connector.mysql.MySqlConnector.
I don't need to pass these settings below in KafkaConnector?

database.history.security.protocol: "SASL_SSL"
database.history.sasl.mechanism: "SCRAM-SHA-512"
database.history.sasl.username: "xxx"
database.history.sasl.password: "xxx"
database.history.ssl.ca.location: "ca.crt"
database.history.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username='xxx' password='xxx';

[andreyolv]

Do you know why i am with this error in KafkaConnector: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata ?

My yamls:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: gac-cdc
  namespace: kafka-strimzi
spec:
  kafka:
    version: 3.3.1
    replicas: 3
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
    authorization:
      type: simple
    readinessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    livenessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.3"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 1Gi
        deleteClaim: false
      - id: 1
        type: persistent-claim
        size: 1Gi
        deleteClaim: false
    resources:
      requests:
        memory: 512Mi
        cpu: "500m"
      limits:
        memory: 1Gi
        cpu: "1000m"
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 1Gi
      deleteClaim: false
    resources:
      requests:
        memory: 512Mi
        cpu: "250m"
      limits:
        memory: 1Gi
        cpu: "500m"
  entityOperator:
    topicOperator: {}
    userOperator: {}

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: gac
  namespace: kafka-strimzi
  labels:
    strimzi.io/cluster: gac-cdc
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
    # Kafka Connects internal topics used to store configuration, offsets or status
    - resource:
        type: group
        name: connect-cluster
      operations:
        - Create
        - Describe
        - Read
        - Write
    - resource:
        type: topic
        name: connect-cluster-configs
      operations:
        - Create
        - Describe
        - Read
        - Write
    - resource:
        type: topic
        name: connect-cluster-status
      operations:
        - Create
        - Describe
        - Read
        - Write
    - resource:
        type: topic
        name: connect-cluster-offsets
      operations:
        - Create
        - Describe
        - Read
        - Write
    # Additional topics and groups used by connectors
    # Change to match the topics used by your connectors
    - resource:
        type: group
        name: connect-cluster
      operations:
        - Create
        - Describe
        - Read
        - Write
    - resource:
        type: topic
        name: dbserver1.inventory.products
      operations:
        - Create
        - Describe
        - Read
        - Write
       

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: connect-cluster
  namespace: kafka-strimzi
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.3.1
  image: mateushenrique/owshq-kafka-connect-strimzi:3.2.3
  replicas: 1
  bootstrapServers: gac-cdc-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: gac-cdc-cluster-ca-cert
        certificate: ca.crt
  authentication:
    type: scram-sha-512
    username: gac
    passwordSecret:
      secretName: gac
      password: password
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
  resources:
    requests:
        memory: 1096Mi
        cpu: "800m"
    limits:
        memory: 2096Mi
        cpu: "800m"
  externalConfiguration:
    volumes:
      - name: connector-config
        secret:
          secretName: mysql-credentials

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: "inventory-connector"
  namespace: kafka-strimzi
  labels:
    strimzi.io/cluster: connect-cluster
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    #key.converter: io.confluent.connect.avro.AvroConverter
    #key.converter.schema.registry.url: http://schema-registry-cp-schema-registry:8081
    #value.converter: io.confluent.connect.avro.AvroConverter
    #value.converter.schema.registry.url: http://schema-registry-cp-schema-registry:8081
    database.hostname: mysql.mysql
    database.port: "3306"
    database.user: "xxx"
    database.password: "xxx"
    database.server.id: "184054"
    database.server.name: "dbserver1"
    database.whitelist: "inventory" 
    database.history.kafka.bootstrap.servers: "gac-cdc-kafka-bootstrap:9093"
    database.history.kafka.topic: "sddl-dbhistory-cdc-tables"
    #database.history.security.protocol: "SASL_SSL"
    #database.history.sasl.mechanism: "SCRAM-SHA-512"
    #database.history.sasl.username: "xxx"
    #database.history.sasl.password: "xxx"
    #database.history.ssl.ca.location: "infrastructure/kafka-connect/ca.crt"
    #database.history.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username='xxx' password='xxx';"
    include.schema.changes: "true"
    database.allowPublicKeyRetrieval: "true"

[scholzj]

If you can format it as a code, it would show namespaces and be readable.

Debezium is special with its history connector. The best way to work around it is to use the config providers and either load the configs from a secret or from the Connect Kafka properties file. But I'm not aware of any Connector other than Debezium doing this (I'm sure they have good reasons for it, never the less it is not usual).

[andreyolv]

I edited the yamls for better visualization

[andreyolv]

I am also in doubt about KafkaUser, there is a user file in the examples/security/scram-sha-512-auth folder, but in the connect.yaml file there is also another user. The first called my-user, the second called my-connect. Why? Do I need the 2 KafkaUser? Thanks

[scholzj]

The user.yaml is an example of an user you would use for example with yout consumers or producers. The user in connect.yaml is the user used the for the Connect. You will anyway need to customize it to match the ACLs your connector needs.