SSL Security to Debezium KafkaConnector - Errors

[andreyolv]

Hi, I'm trying run security with debezium, but without success, it's giving an error:

trace: "org.apache.kafka.connect.errors.ConnectException: Creation of database
        history topic failed, please create the topic manually\n\tat io.debezium.relational.history.KafkaDatabaseHistory.initializeStorage(KafkaDatabaseHistory.java:509)\n\tat
        io.debezium.relational.HistorizedRelationalDatabaseSchema.initializeStorage(HistorizedRelationalDatabaseSchema.java:81)\n\tat
        io.debezium.connector.mysql.MySqlDatabaseSchema.initializeStorage(MySqlDatabaseSchema.java:384)\n\tat
        io.debezium.connector.mysql.MySqlConnectorTask.validateAndLoadDatabaseHistory(MySqlConnectorTask.java:347)\n\tat
        io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:108)\n\tat
        io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:130)\n\tat
        org.apache.kafka.connect.runtime.WorkerSourceTask.initializeAndStart(WorkerSourceTask.java:226)\n\tat
        org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186)\n\tat
        org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)\n\tat
        java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat
        java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat
        java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat
        java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.util.concurrent.ExecutionException:
        org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization
        failed.\n\tat java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)\n\tat
        java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2022)\n\tat
        org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180)\n\tat
        io.debezium.relational.history.KafkaDatabaseHistory.getKafkaBrokerConfig(KafkaDatabaseHistory.java:547)\n\tat
        io.debezium.relational.history.KafkaDatabaseHistory.getDefaultTopicReplicationFactor(KafkaDatabaseHistory.java:515)\n\tat
        io.debezium.relational.history.KafkaDatabaseHistory.initializeStorage(KafkaDatabaseHistory.java:498)\n\t...
        13 more\nCaused by: org.apache.kafka.common.errors.ClusterAuthorizationException:
        Cluster authorization failed.\n"

What is wrong with my yamls configs for me to have the above error? Can someone help me with a clear and complete example?
Just KafkaConnect show errors, Broker and KafkaConnect run without errors. I tried to create a topic called 'sddl-dbhistory-cdc-tables' by KafkaTopic kind, but doesn't work too.

Thanks

My yamls:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  namespace: kafka-strimzi
  labels:
    strimzi.io/cluster: gac-cdc
spec:
  quotas:
   producerByteRate: 1048576 # 1 MB/s
   consumerByteRate: 1048576 # 1 MB/s
  authentication:
    type: tls #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:
        - Read
        - 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: topic
        name: sddl-dbhistory-cdc-tables
      operations:
        - Create
        - Describe
        - Read
        - Write
    - resource:
        type: topic
        name: dbserver1.inventory.*
      operations:
        - Create
        - Describe
        - Read
        - Write

---

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: tls 
    authorization:
      type: simple
...

---

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: tls
    certificateAndKey:
      secretName: my-user
      certificate: user.crt
      key: user.key
  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
    config.providers: directory
    config.providers.directory.class: org.apache.kafka.common.config.provider.DirectoryConfigProvider
  externalConfiguration:
    volumes:
      - name: connector-config
        secret:
          secretName: gac-secrets
      - name: cluster-ca 
        secret:
          secretName: gac-cdc-cluster-ca-cert
      - name: my-user
        secret:
          secretName: my-user

---

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
    key.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true

    database.hostname: mysql.mysql
    database.port: "3306"
    database.user: "${directory:/opt/kafka/external-configuration/connector-config:gac_user}"
    database.password: "${directory:/opt/kafka/external-configuration/connector-config:gac_password}"
    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"

    # PEM general definition
    security.protocol: SSL
    ssl.truststore.type: PEM
    ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/cluster-ca:ca.crt}"
    ssl.keystore.type: PEM
    ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/my-user:user.crt}"
    ssl.keystore.key: "${directory:/opt/kafka/external-configuration/my-user:user.key}"
    # PEM config for producer and consumer
    # producer
    database.history.producer.security.protocol: SSL
    database.history.producer.ssl.truststore.type: PEM
    database.history.producer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/cluster-ca:ca.crt}"
    database.history.producer.ssl.keystore.type: PEM
    database.history.producer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/my-user:user.crt}"
    database.history.producer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/my-user:user.key}"
    # consumer
    database.history.consumer.security.protocol: SSL
    database.history.consumer.ssl.truststore.type: PEM
    database.history.consumer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/cluster-ca:ca.crt}"
    database.history.consumer.ssl.keystore.type: PEM
    database.history.consumer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/my-user:user.crt}"
    database.history.consumer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/my-user:user.key}"

---

[scholzj]

Do you have any authorization errors in the Kafka Broker logs? I guess Cluster authorization failed suggests you are missing some ACLs maybe? It looks like some topic cannot be created.

[andreyolv]

I added this for my user yaml, so my user would be allowed to create any topic, but doesn't work.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  namespace: kafka-strimzi
  labels:
    strimzi.io/cluster: gac-cdc
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
    - resource:
        type: cluster
        name: gac-cdc
        patternType: literal
      operations:
        - Create
        - Describe
        - Alter
      host: "*"
    - resource:
        type: group
        name: connect-cluster
        patternType: literal
      operations: 
        - Read
        - Describe
        - Delete
      host: "*"
    - resource:
        type: topic
        name: "*"
        patternType: literal
      operations: 
        - Read
        - Describe
        - Write
        - Create
      host: "*"

The Kafka Broker logs is repeatedly this:
2023-02-09 20:47:38,888 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.244.1.4 (channelId=10.244.2.8:9093-10.244.1.4:51242-369) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(TLS-9093)-SSL-5]

[scholzj]

And is the 10.244.1.4 IP the IP address of the Connect pod? This would suggest an issue with authentication or TLS configuration and not authorization. So that would not correspond with the first error. But the configs seem to be ok.

In the ACLs ... I think that this name: dbserver1.inventory.* is wrong. It does not accept any regexes like this. You can use the patternType: prefix and specify the prefix of the topic only. But not a wildcards ro reges like that.

[andreyolv]

The IP 10.244.1.4 it was from the schema registry pod that I hadn't configured anything security yet, so I just deleted the schema registry and the broker log disappeared. But the error in KafkaConnect remains.
I also adjusted KafkaUser's ACLs as you said, I edited the above yaml with my new yaml giving permissions for resources type: cluster, group and topic.

No more error in the Broker logs.

I saw that the zookeeper pods have the following error:

2023-02-10 01:31:34,939 ERROR Exception while listening (org.apache.zookeeper.server.quorum.QuorumCnxManager) [ListenerHandler-gac-cdc-zookeeper-0.gac-cdc-zookeeper-nodes.kafka-strimzi.svc:3888]
java.net.SocketException: Unresolved address
        at java.base/java.net.ServerSocket.bind(ServerSocket.java:388)
        at java.base/java.net.ServerSocket.bind(ServerSocket.java:349)
        at org.apache.zookeeper.server.quorum.QuorumCnxManager$Listener$ListenerHandler.createNewServerSocket(QuorumCnxManager.java:1135)
        at org.apache.zookeeper.server.quorum.QuorumCnxManager$Listener$ListenerHandler.acceptConnections(QuorumCnxManager.java:1064)
        at org.apache.zookeeper.server.quorum.QuorumCnxManager$Listener$ListenerHandler.run(QuorumCnxManager.java:1033)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)

The Connector pod has this extra error log:

2023-02-10 01:35:56,453 ERROR [inventory-connector|task-0] WorkerSourceTask{id=inventory-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-inventory-connector-0]

The EntityOperator pod has this warning:

2023-02-10 01:35:11,39816 WARN  [vert.x-eventloop-thread-5] ZkTopicsWatcher:152 - Reconciliation #421(/brokers/topics 4:+connect-cluster-status) KafkaTopic(kafka-strimzi/connect-cluster-status): Error responding to creation of topic connect-cluster-status
org.apache.kafka.common.errors.UnknownTopicOrPartitionException: 

but topic connect-cluster-status was created

Any more ideas of what it could be?

[scholzj]

It's quite hard to keep track of the different errors. You should maybe share the full logs. The ZooKeeper error suggests some networking issue, but if ZooKeepers work it was just some temp glitch or from some startup.

[BBQigniter]

Hi @andreyolv ,

at your request in #6536 (comment)

our strimzi/kafka-cluster config got quite huge and it would take some time for me to obfuscate stuff etc.

Also we are using the latest Debezium MySQL-connector (v2.1.2) and other additional libraries where we cannot use the built-in image-building method with the Kaniko thingy (which would need a separate HowTo for understanding our whole setup with Strimzi :)).

Our user.yaml looks like this

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: app-kafka-test-dev-tls
  namespace: kafka-test
  labels:
    strimzi.io/cluster: kafka-test
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
    # topics
    - resource:
        type: topic
        name: "ms-"
        patternType: prefix
      operation: All
    # groups
    - resource:
        type: group
        name: "ms-"
        patternType: prefix
      operation: All
    # for kafka-connector kafka-test-connect
    - resource:
        type: topic
        name: "kafka-test-connect"
        patternType: prefix
      operation: All
    - resource:
        type: group
        name: "kafka-test-connect"
        patternType: literal
      operation: All

Are your operation-settings maybe too restrictive?

A connector.yaml - I had to remove some stuff, hopefully it's still clear and I didn't mess something else up.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: ms-connector
  namespace: kafka-test
  labels:
    # The strimzi.io/cluster label identifies the KafkaConnect instance
    # in which to create this connector. That KafkaConnect instance
    # must have the strimzi.io/use-connector-resources annotation
    # set to true.
    strimzi.io/cluster: kafka-test-connect
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    database.hostname: "ms.test.database"
    database.port: "3306"
    ...
    # to create the credentials as secret you have to do the following
    # cat <<EOF > kafka-test-ms-test-connector-connector-config-mysql-credentials.properties
    # mysql_username: app_debezium
    # mysql_password: the_password
    # EOF
    #
    # kubectl -n kafka-test create secret generic kafka-test-connector-app-debezium-credentials --from-file=kafka-test-connector-app-debezium-credentials.properties
    database.user: "${file:/opt/kafka/external-configuration/kafka-test-connector-app-debezium-credentials-config/kafka-test-connector-app-debezium-credentials.properties:mysql_username}"
    database.password: "${file:/opt/kafka/external-configuration/kafka-test-connector-app-debezium-credentials-config/kafka-test-connector-app-debezium-credentials.properties:mysql_password}"
    database.server.id: "25501002"
    database.server.name: "ms-test"
    database.whitelist: "ms"
    database.connectionTimeZone: "Europe/Vienna"
    ...
    topic.prefix: "ms-"
    ...
    errors.log.enable: "true"
    errors.log.include.messages: "true"
    # PEM general definition
    security.protocol: SSL
    ssl.truststore.type: PEM
    ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/kafka-test-cluster-ca-cert-config:ca.crt}"
    ssl.keystore.type: PEM
    ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.crt}"
    ssl.keystore.key: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.key}"
    # debezium 1.9.x
    #database.history.kafka.bootstrap.servers: "kafka-test-kafka-bootstrap:9093"
    #database.history.kafka.topic: "ms-test.test.schema"
    ## PEM config for producer and consumer
    ## producer
    #database.history.producer.security.protocol: SSL
    #database.history.producer.ssl.truststore.type: PEM
    #database.history.producer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/kafka-test-cluster-ca-cert-config:ca.crt}"
    #database.history.producer.ssl.keystore.type: PEM
    #database.history.producer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.crt}"
    #database.history.producer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.key}"
    ## consumer
    #database.history.consumer.security.protocol: SSL
    #database.history.consumer.ssl.truststore.type: PEM
    #database.history.consumer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/kafka-test-cluster-ca-cert-config:ca.crt}"
    #database.history.consumer.ssl.keystore.type: PEM
    #database.history.consumer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.crt}"
    #database.history.consumer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.key}"
    # debezium 2.0.x and 2.1.x
    schema.history.internal.kafka.bootstrap.servers: "kafka-test-kafka-bootstrap:9093"
    schema.history.internal.kafka.topic: "ms-test.test.schema"
    schema.history.internal.skip.unparseable.ddl: "true"
    # PEM config for producer and consumer
    # producer
    schema.history.internal.producer.security.protocol: SSL
    schema.history.internal.producer.ssl.truststore.type: PEM
    schema.history.internal.producer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/kafka-test-cluster-ca-cert-config:ca.crt}"
    schema.history.internal.producer.ssl.keystore.type: PEM
    schema.history.internal.producer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.crt}"
    schema.history.internal.producer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.key}"
    # consumer
    schema.history.internal.consumer.security.protocol: SSL
    schema.history.internal.consumer.ssl.truststore.type: PEM
    schema.history.internal.consumer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/kafka-test-cluster-ca-cert-config:ca.crt}"
    schema.history.internal.consumer.ssl.keystore.type: PEM
    schema.history.internal.consumer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.crt}"
    schema.history.internal.consumer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/app-kafka-test-dev-tls-config:user.key}"

Watch out for the config-changes between Debezium connector v1.9 and v2.x!

We create the Kafka history-topic normally before we start/apply the connector with a yaml-file similar to this.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  labels:
    strimzi.io/cluster: kafka-test
  name: ms-test.test.schema
  namespace: kafka-test
spec:
  config:
    cleanup.policy: delete
    min.insync.replicas: "2"
    retention.bytes: "-1"
    retention.ms: "9223372036854775807"
    segment.bytes: "1038336"
  partitions: 1
  replicas: 3
  topicName: ms-test.test.schema

The main Kafka configuration is mostly done by our main Kafka dude - he normally sends me just the configs and I make sure it runs with Kubernetes/Strimzi. So for detailed Kafka-help I may not be much of a help :)

[andreyolv]

Thank you! It worked out!

[BBQigniter]

nice, and how did you fix it? :)

[andreyolv]

I changed my kafka user with the ACLs like your yaml :)