Strimzi-kafka-operator: [Question] authentication, type: scram-sha-512 for NodePort listener is it possible with simple authorization for user?

Created on 29 Jan 2020  路  8Comments  路  Source: strimzi/strimzi-kafka-operator

Greeting team! :)
In short: I'm trying to configure external listener (NodePort) to use authentication scram-sha-512 , and Kafka user with the same authentication method. Bellow will be my configs for KafkaUser and Kafka:

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Read
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
      - resource:
          type: group
          name: my-group
          patternType: prefix
        operation: Read

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.4.0
    replicas: 1
    authentication:
      type: scram-sha-512
    listeners:
      external:
        type: nodeport
        authentication:
          type: scram-sha-512
        overrides:
          bootstrap:
            nodePort: 32100
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      log.message.format.version: "2.4"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 10Gi
        deleteClaim: true
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 10Gi
      deleteClaim: true
  entityOperator:
    topicOperator: {}
    userOperator: {}



md5-80e1188337ef5b99f31de61ea2bfd8d5



$ kubectl logs my-cluster-kafka-0 -n kafka -c kafka
...
2020-01-29 09:19:50,631 INFO Removing PRODUCE quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:19:50,637 INFO Removing FETCH quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:19:50,638 INFO Removing REQUEST quota for user my-user (kafka.server.ClientRequestQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:19:50,640 INFO Processing notification(s) to /kafka-acl-changes (kafka.common.ZkNodeChangeNotificationListener) [/kafka-acl-changes-event-process-thread]
2020-01-29 09:19:50,805 INFO Processing notification(s) to /config/changes (kafka.common.ZkNodeChangeNotificationListener) [/config/changes-event-process-thread]
2020-01-29 09:19:50,808 INFO Processing override for entityPath: users/my-user with config: Map(SCRAM-SHA-512 -> [hidden]) (kafka.server.DynamicConfigManager) [/config/changes-event-process-thread]
2020-01-29 09:19:50,808 INFO Removing PRODUCE quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:19:50,808 INFO Removing FETCH quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:19:50,808 INFO Removing REQUEST quota for user my-user (kafka.server.ClientRequestQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:20:32,500 INFO Processing notification(s) to /config/changes (kafka.common.ZkNodeChangeNotificationListener) [/config/changes-event-process-thread]
2020-01-29 09:20:32,504 INFO Processing override for entityPath: users/my-user with config: Map(SCRAM-SHA-512 -> [hidden]) (kafka.server.DynamicConfigManager) [/config/changes-event-process-thread]
2020-01-29 09:20:32,504 INFO Removing PRODUCE quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:20:32,504 INFO Removing FETCH quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:20:32,504 INFO Removing REQUEST quota for user my-user (kafka.server.ClientRequestQuotaManager) [/config/changes-event-process-thread]
2020-01-29 09:20:56,295 INFO [SocketServer brokerId=0] Failed authentication with gateway/10.204.29.1 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_SSL-3]
2020-01-29 09:20:56,464 INFO [SocketServer brokerId=0] Failed authentication with gateway/10.204.29.1 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_SSL-4]
2020-01-29 09:20:56,755 INFO [SocketServer brokerId=0] Failed authentication with gateway/10.204.29.1 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_SSL-5]
2020-01-29 09:20:57,174 INFO [SocketServer brokerId=0] Failed authentication with gateway/10.204.29.1 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_SSL-3]
2020-01-29 09:20:58,315 INFO [SocketServer brokerId=0] Failed authentication with gateway/10.204.29.1 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_SSL-4]
2020-01-29 09:20:59,998 INFO [SocketServer brokerId=0] Failed authentication with gateway/10.204.29.1 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_SSL-5]



md5-59c05605c79679509819c566c9a700e7



kafkacat -C -b server-ip:32100 -t my-topic -L -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN -X sasl.mechanisms=SCRAM-SHA-512 -X sasl.username="my-user" -X sasl.password="L6ODHIhCV2Gk" -v -X debug=generic,broker,security

%7|1580289656.069|SASL|rdkafka#producer-1| [thrd:app]: Selected provider SCRAM (builtin) for SASL mechanism SCRAM-SHA-512
%7|1580289656.069|OPENSSL|rdkafka#producer-1| [thrd:app]: Using OpenSSL version OpenSSL 1.1.1d  10 Sep 2019 (0x1010104f, librdkafka built with 0x1010104f)
%7|1580289656.080|BROKER|rdkafka#producer-1| [thrd:app]: sasl_ssl://198.18.64.29:32100/bootstrap: Added new broker with NodeId -1
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:app]: sasl_ssl://198.18.64.29:32100/bootstrap: Selected for cluster connection: bootstrap servers added (broker has 0 connection attempt(s))
%7|1580289656.080|BRKMAIN|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: Enter main broker thread
%7|1580289656.080|INIT|rdkafka#producer-1| [thrd:app]: librdkafka v1.2.1 (0x10201ff) rdkafka#producer-1 initialized (builtin.features gzip,snappy,ssl,sasl,regex,lz4,sasl_gssapi,sasl_plain,sasl_scram,plugins,zstd,sasl_oauthbearer, CC CXX PKGCONFIG OSXLD LIBDL PLUGINS ZLIB SSL SASL_CYRUS ZSTD HDRHISTOGRAM LZ4_EXT SNAPPY SOCKEM SASL_SCRAM SASL_OAUTHBEARER CRC32C_HW, debug 0x203)
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: Received CONNECT op
%7|1580289656.080|STATE|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: Broker changed state INIT -> TRY_CONNECT
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:app]: Not selecting any broker for cluster connection: still suppressed for 49ms: leader query
%7|1580289656.080|BROADCAST|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: Broadcasting state change
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: broker in state TRY_CONNECT connecting
%7|1580289656.080|STATE|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: Broker changed state TRY_CONNECT -> CONNECT
%7|1580289656.080|BROADCAST|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: Broadcasting state change
%7|1580289656.080|BRKMAIN|rdkafka#producer-1| [thrd::0/internal]: :0/internal: Enter main broker thread
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:app]: Not selecting any broker for cluster connection: still suppressed for 49ms: leader query
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:app]: Not selecting any broker for cluster connection: still suppressed for 49ms: application metadata request
%7|1580289656.080|CONNECT|rdkafka#producer-1| [thrd:app]: Not selecting any broker for cluster connection: still suppressed for 49ms: application metadata request
%7|1580289656.081|CONNECT|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: Connecting to ipv4#198.18.64.29:32100 (sasl_ssl) with socket 7
%7|1580289656.161|CONNECT|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: Connected to ipv4#198.18.64.29:32100
%7|1580289656.258|BROKERFAIL|rdkafka#producer-1| [thrd:sasl_ssl://198.18.64.29:32100/bootstrap]: sasl_ssl://198.18.64.29:32100/bootstrap: failed: err: Local: SSL error: (errno: Undefined error: 0)

I could provide any additional info, topic was created with command:
kubectl -n kafka run kafka-producer -ti --image=strimzi/kafka:0.16.1-kafka-2.4.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9092 --topic my-topic
Any suggestions, is it possible to configure such authentication setup?

question
Source
picture lubouski

Most helpful comment

@scholzj thank you for your support! The problem was in my limited knowledge of kafkacat utility :)

Yeap, the problem was that topic not created with command:
kubectl -n kafka run kafka-producer -ti --image=strimzi/kafka:0.16.1-kafka-2.4.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9092 --topic my-topic

but I create it with: 

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824

After that everything start working, when I added permissions to user :)

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Read
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Write
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Create
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
      - resource:
          type: group
          name: my-group
          patternType: prefix
        operation: Read

./kafka-console-producer --broker-list 198.18.64.29:32100 --producer.config consumer.properties  --topic my-topic
[2020-01-29 15:46:09,933] WARN The configuration 'producer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
[2020-01-29 15:46:09,933] WARN The configuration 'producer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
[2020-01-29 15:46:09,933] WARN The configuration 'producer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
>kafka
>kafka
>kafka
>^C%

And consumer could read from topic:

./kafka-console-consumer --bootstrap-server 198.18.64.29:32100 --consumer.config consumer.properties  --group my-group --topic my-topic
[2020-01-29 15:46:29,472] WARN The configuration 'producer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.consumer.ConsumerConfig)
[2020-01-29 15:46:29,472] WARN The configuration 'producer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.consumer.ConsumerConfig)
[2020-01-29 15:46:29,472] WARN The configuration 'producer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.consumer.ConsumerConfig)
kafka
kafka
kafka
^CProcessed a total of 3 messages

And here a config file for consumer.properties if someone would search for it:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=my-user password=QHXaVlTSmovv;
producer.security.protocol=SASL_PLAINTEXT
producer.sasl.mechanism=SCRAM-SHA-512
producer.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=my-user password=QHXaVlTSmovv;
picture lubouski on 29 Jan 2020
👍2

All 8 comments

I'm not kafkacat user. But I see some issues with your command:

  • You have two SASL mechanisms configured ... -X sasl.mechanisms=PLAIN -X sasl.mechanisms=SCRAM-SHA-512 ... the SCRAM-SHA is the right one. I'm not sure how kafkacat deal with this and whether it even supports SCRAM-SHA-512.
  • You seem to enable SSL. But you do not pass the cluster CA certificate. So I would expect the client to fail server authentication. Also with node ports, you might need to disable TLS hostname verification (again, I have no idea how in kafkacat).

In general, this is regular setup which should work, so I owuld expect this to be mostly a confguration issue. Maybe you can take is step by step - try to disable TLS (tls: false in Kafka.spec.kafkalisteners.external) and see if the SASL it self works for you and than move to figure out the TLS.

scholzj picture scholzj on 29 Jan 2020

Jakub many thanks to your fast answer!

Here reconfigured configs without TLS, and fixed kafkacat command
kafkacat -C -b my-server:32100 -t my-topic -L -X security.protocol=SASL_SSL -X sasl.mechanisms=SCRAM-SHA-512 -X sasl.username="my-user" -X sasl.password="xOaNOYm72yPd" -v -X debug=generic,broker,security

Configs:

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.4.0
    replicas: 1
    listeners:
      external:
        type: nodeport
        authentication:
          type: scram-sha-512
        overrides:
          bootstrap:
            nodePort: 32100
        tls: false
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      log.message.format.version: "2.4"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 10Gi
        deleteClaim: true
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 10Gi
      deleteClaim: true
  entityOperator:
    topicOperator: {}
    userOperator: {}

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Read
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
      - resource:
          type: group
          name: my-group
          patternType: prefix
        operation: Read



md5-3d3423d508b647cd8a34ec2ef1657056



2020-01-29 12:08:35,689 INFO Removing REQUEST quota for user my-user (kafka.server.ClientRequestQuotaManager) [/config/changes-event-process-thread]
2020-01-29 12:08:45,224 WARN [SocketServer brokerId=0] Unexpected error from /10.204.29.1; closing connection (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_PLAINTEXT-3]
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295617 larger than 524288)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
    at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:268)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at kafka.network.Processor.poll(SocketServer.scala:890)
    at kafka.network.Processor.run(SocketServer.scala:789)
    at java.lang.Thread.run(Thread.java:748)
2020-01-29 12:08:45,380 WARN [SocketServer brokerId=0] Unexpected error from /10.204.29.1; closing connection (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_PLAINTEXT-4]
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295617 larger than 524288)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
    at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:268)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at kafka.network.Processor.poll(SocketServer.scala:890)
    at kafka.network.Processor.run(SocketServer.scala:789)
    at java.lang.Thread.run(Thread.java:748)
2020-01-29 12:08:45,601 WARN [SocketServer brokerId=0] Unexpected error from /10.204.29.1; closing connection (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_PLAINTEXT-5]
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295617 larger than 524288)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
    at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:268)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at kafka.network.Processor.poll(SocketServer.scala:890)
    at kafka.network.Processor.run(SocketServer.scala:789)
    at java.lang.Thread.run(Thread.java:748)
2020-01-29 12:08:45,912 WARN [SocketServer brokerId=0] Unexpected error from /10.204.29.1; closing connection (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_PLAINTEXT-3]
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295617 larger than 524288)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
    at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:268)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at kafka.network.Processor.poll(SocketServer.scala:890)
    at kafka.network.Processor.run(SocketServer.scala:789)
    at java.lang.Thread.run(Thread.java:748)
2020-01-29 12:08:46,612 WARN [SocketServer brokerId=0] Unexpected error from /10.204.29.1; closing connection (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_PLAINTEXT-4]
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295617 larger than 524288)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
    at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:268)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at kafka.network.Processor.poll(SocketServer.scala:890)
    at kafka.network.Processor.run(SocketServer.scala:789)
    at java.lang.Thread.run(Thread.java:748)
2020-01-29 12:08:48,170 WARN [SocketServer brokerId=0] Unexpected error from /10.204.29.1; closing connection (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SASL_PLAINTEXT-5]
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295617 larger than 524288)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:104)
    at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:268)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
    at kafka.network.Processor.poll(SocketServer.scala:890)
    at kafka.network.Processor.run(SocketServer.scala:789)
    at java.lang.Thread.run(Thread.java:748)
2020-01-29 12:09:18,835 INFO [Controller id=0] Processing automatic preferred replica leader election (kafka.controller.KafkaController) [controller-event-thread]
2020-01-29 12:09:18,835 TRACE [Controller id=0] Checking need to trigger auto leader balancing (kafka.controller.KafkaController) [controller-event-thread]
2020-01-29 12:09:18,835 DEBUG [Controller id=0] Preferred replicas by broker Map() (kafka.controller.KafkaController) [controller-event-thread]
lubouski picture lubouski on 29 Jan 2020

No idea what that means I'm afraid. Can you try this instead:

  • Download and unpack the Kafka ZIP
  • Create file consumer.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=my-user password=xOaNOYm72yPd;
  • Run followig command:
bin/kafka-console-consumer.sh --bootstrap-server my-server:32100 --consumer.config consumer.properties 聽--group my-group --topic my-topic

The console consumer is something I use and I'm more familiar with it than kafkacat.

scholzj picture scholzj on 29 Jan 2020
👍1

Here the results of ./kafka-console-consumer

$  cat consumer.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=my-user password=xOaNOYm72yPd;

command:

./kafka-console-consumer --bootstrap-server my-server:32100 --consumer.config consumer.properties  --group my-group --topic my-topic
[2020-01-29 15:28:16,314] WARN [Consumer clientId=consumer-my-group-1, groupId=my-group] Error while fetching metadata with correlation id 2 : {my-topic=TOPIC_AUTHORIZATION_FAILED} (org.apache.kafka.clients.NetworkClient)
[2020-01-29 15:28:16,318] ERROR [Consumer clientId=consumer-my-group-1, groupId=my-group] Topic authorization failed for topics [my-topic] (org.apache.kafka.clients.Metadata)
[2020-01-29 15:28:16,319] ERROR Error processing message, terminating consumer process:  (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [my-topic]

And logs from kafka

2020-01-29 12:27:03,243 TRACE [Controller id=0 epoch=1] Received response {error_code=0,_tagged_fields={}} for request UPDATE_METADATA with correlation id 2 sent to broker my-cluster-kafka-0.my-cluster-kafka-brokers.kafka.svc:9091 (id: 0 rack: null) (state.change.logger) [Controller-0-to-broker-0-send-thread]
2020-01-29 12:28:16,290 INFO Principal = User:my-user is Denied Operation = Create from host = 10.204.29.1 on resource = Cluster:LITERAL:kafka-cluster (kafka.authorizer.logger) [data-plane-kafka-request-handler-3]
2020-01-29 12:28:16,291 INFO Principal = User:my-user is Denied Operation = Create from host = 10.204.29.1 on resource = Topic:LITERAL:my-topic (kafka.authorizer.logger) [data-plane-kafka-request-handler-3]
2020-01-29 12:28:35,669 INFO Processing notification(s) to /config/changes (kafka.common.ZkNodeChangeNotificationListener) [/config/changes-event-process-thread]
2020-01-29 12:28:35,671 INFO Processing override for entityPath: users/my-user with config: Map(SCRAM-SHA-512 -> [hidden]) (kafka.server.DynamicConfigManager) [/config/changes-event-process-thread]
2020-01-29 12:28:35,671 INFO Removing PRODUCE quota for user my-user (kafka.server.ClientQuotaManager) [/config/changes-event-process-thread]
lubouski picture lubouski on 29 Jan 2020

Ok, it looks like the topic does not exist and the user has no permission to create it. So you either need to give the user the create rights as well or create the topics manually.

scholzj picture scholzj on 29 Jan 2020

@scholzj thank you for your support! The problem was in my limited knowledge of kafkacat utility :)

Yeap, the problem was that topic not created with command:
kubectl -n kafka run kafka-producer -ti --image=strimzi/kafka:0.16.1-kafka-2.4.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9092 --topic my-topic

but I create it with: 

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824

After that everything start working, when I added permissions to user :)

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Read
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Write
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Create
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
      - resource:
          type: group
          name: my-group
          patternType: prefix
        operation: Read

./kafka-console-producer --broker-list 198.18.64.29:32100 --producer.config consumer.properties  --topic my-topic
[2020-01-29 15:46:09,933] WARN The configuration 'producer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
[2020-01-29 15:46:09,933] WARN The configuration 'producer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
[2020-01-29 15:46:09,933] WARN The configuration 'producer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
>kafka
>kafka
>kafka
>^C%

And consumer could read from topic:

./kafka-console-consumer --bootstrap-server 198.18.64.29:32100 --consumer.config consumer.properties  --group my-group --topic my-topic
[2020-01-29 15:46:29,472] WARN The configuration 'producer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.consumer.ConsumerConfig)
[2020-01-29 15:46:29,472] WARN The configuration 'producer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.consumer.ConsumerConfig)
[2020-01-29 15:46:29,472] WARN The configuration 'producer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.consumer.ConsumerConfig)
kafka
kafka
kafka
^CProcessed a total of 3 messages

And here a config file for consumer.properties if someone would search for it:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=my-user password=QHXaVlTSmovv;
producer.security.protocol=SASL_PLAINTEXT
producer.sasl.mechanism=SCRAM-SHA-512
producer.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=my-user password=QHXaVlTSmovv;
lubouski picture lubouski on 29 Jan 2020
👍2

@lubouski enbale SASL mechanisms,for the question Invalid receive (size = 369295617 larger than 524288) ,
How did you solve it

jianke221 picture jianke221 on 18 Nov 2020

Invalid receive (size = 369295617 larger than 524288)

Could it be that you connect to TLS interface without TLS or the other way around? That sometimes produces this kind of messages.

scholzj picture scholzj on 18 Nov 2020
Was this page helpful?
0 / 5 - 0 ratings

Related issues

MacPower picture MacPower  路  4Comments
eazhilan-nagarajan picture eazhilan-nagarajan  路  8Comments
ddlees picture ddlees  路  5Comments
octopyth picture octopyth  路  4Comments
skbly7 picture skbly7  路  6Comments