Unable to connect confluent schema to strimzi kafka cluster

[Sai-Charan-Madhvaraj]

Hi Team,

we have strimzi kafka cluster running and trying to connect confluent schema registry (helm charts) to strimzi kafka cluster using PLAINTEXT mechanism but it fails error

1. " ERROR Error starting the schema registry (io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication)
2. org.apache.kafka.common.errors.TopicAuthorizationException: Authorization failed.

here is the complete log

(org.apache.kafka.clients.admin.AdminClientConfig)
[2022-05-13 11:00:18,738] WARN The configuration 'sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
[2022-05-13 11:00:18,738] WARN The configuration 'group.id' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
[2022-05-13 11:00:18,738] INFO Kafka version: 6.1.0-ce (org.apache.kafka.common.utils.AppInfoParser)
[2022-05-13 11:00:18,738] INFO Kafka commitId: 958ad0f3c7030f1c (org.apache.kafka.common.utils.AppInfoParser)
[2022-05-13 11:00:18,738] INFO Kafka startTimeMs: 1652439618738 (org.apache.kafka.common.utils.AppInfoParser)
[2022-05-13 11:00:18,758] INFO Creating schemas topic _schemas (io.confluent.kafka.schemaregistry.storage.KafkaStore)
[2022-05-13 11:00:18,838] INFO App info kafka.admin.client for adminclient-2 unregistered (org.apache.kafka.common.utils.AppInfoParser)
[2022-05-13 11:00:18,841] INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics)
[2022-05-13 11:00:18,841] INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics)
[2022-05-13 11:00:18,841] INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics)
[2022-05-13 11:00:18,841] ERROR Error starting the schema registry (io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication)
io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryInitializationException: Error initializing kafka store while initializing schema registry
	at io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.init(KafkaSchemaRegistry.java:297)
	at io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication.initSchemaRegistry(SchemaRegistryRestApplication.java:73)
	at io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication.configureBaseApplication(SchemaRegistryRestApplication.java:88)
	at io.confluent.rest.Application.configureHandler(Application.java:255)
	at io.confluent.rest.ApplicationServer.doStart(ApplicationServer.java:227)
	at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:72)
	at io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain.main(SchemaRegistryMain.java:43)
Caused by: io.confluent.kafka.schemaregistry.storage.exceptions.StoreInitializationException: Failed trying to create or validate schema topic configuration
	at io.confluent.kafka.schemaregistry.storage.KafkaStore.createOrVerifySchemaTopic(KafkaStore.java:181)
	at io.confluent.kafka.schemaregistry.storage.KafkaStore.init(KafkaStore.java:118)
	at io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.init(KafkaSchemaRegistry.java:295)
	... 6 more
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Authorization failed.
	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:104)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:272)
	at io.confluent.kafka.schemaregistry.storage.KafkaStore.createSchemaTopic(KafkaStore.java:220)
	at io.confluent.kafka.schemaregistry.storage.KafkaStore.createOrVerifySchemaTopic(KafkaStore.java:173)
	... 8 more
Caused by: org.apache.kafka.common.errors.TopicAuthorizationException: Authorization failed.

[Sai-Charan-Madhvaraj]

here are the parameters

bootstrap.servers = [PLAINTEXT://kafka-bootstrap:9092]
	client.dns.lookup = use_all_dns_ips
	client.id = 
	connections.max.idle.ms = 300000
	default.api.timeout.ms = 60000
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = [hidden]
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = PLAIN
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 127000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS

[scholzj]

It says authorization failed ... so it is an ACL error. Did you checked the broker logs as I suggested?

[scholzj]

Also, your configuration suggests it is using SASL PLAIN. We do not support SASL PLAIN in the brokers and you creted the user as SASL SCRAM-SHA-512. So maybe you actually didn't configured any authentication on the broker at all?

[Sai-Charan-Madhvaraj]

@scholzj
Below is the kafka config file we have running.. the broker configuration and authentication is present only for listener 9094.. do we need to add the broker configuration for all listeners (9092, 9093 ) ?

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: xxx
  namespace: xxx
spec:
  kafka:
    version: 3.0.0
    replicas: 5
    resources:
      requests:
        memory: 8Gi
        cpu: 2
      limits:
        memory: 16Gi
        cpu: 4
    listeners:
      - name: external
        port: 9094
        type: ingress
        tls: true
        authentication:
          type: scram-sha-512
        configuration:
          bootstrap:
            host: xxx
          brokers:
          - broker: 0
            host: xxx
          - broker: 1
            host: xxx
          - broker: 2
            host: xxx
          - broker: 3
            host: xxx
          - broker: 4
            host: xxx
          
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: plain
        port: 9092
        type: internal
        tls: false
    authorization:
      type: simple
      superUsers:
        - admin

[scholzj]

Well, you need to decide what do you want to use. But without the authentication, the registry connects as ANONYMOUS (which you would see in the vbroker logs which I suggested twice!). Using authorization without authentication makes in general very little sense.

[Sai-Charan-Madhvaraj]

@scholzj , thanks for the information.. from broker logs there is nothing i see with respect to connection of admin or anonymous user... however have tried using SASL connection with SCRAM-SHA-512 and able to connect succesfully.