0

ローカルでwurstmeister kafka とZookeeper dockerイメージを使用して、 kafkaで SASL と ACL をテストしています。

私の docker-compose.ymlは -

version: '3'

services:
  zookeeper:
    image: wurstmeister/zookeeper
    hostname: zookeeper
    container_name: zookeeper
    volumes:
    - ./zookeeper/zookeeper.sasl.jaas.config:/etc/kafka/zookeeper_server_jaas.conf
    - ./zk/data:/var/lib/zookeeper/data
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_SET_ACL: 'true'
      KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/zookeeper_server_jaas.conf
          -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
          -Dzookeeper.allowSaslFailedClients=false
          -Dzookeeper.requireClientAuthScheme=sasl

  broker:
    image: wurstmeister/kafka:2.13-2.6.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    volumes:
      - ./kafka/kafka.jaas.conf:/etc/kafka/kafka_server_jaas.conf
      - ./kfk/data:/kafka
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:SASL_PLAINTEXT
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_AUTO_CREATE_TOPIC: 'true'
      KAFKA_LISTENERS: EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: EXTERNAL://localhost:9092
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_LISTENER_NAME_EXTERNAL_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_LISTENER_NAME_EXTERNAL_PLAIN_SASL_JAAS_CONFIG: |
            org.apache.kafka.common.security.plain.PlainLoginModule required \
            username="broker" \
            password="broker" \
            user_broker="broker" \
            user_client="client-secret" \
            user_alice="alice-secret";
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
      KAFKA_INTER_BROKER_LISTENER_NAME: EXTERNAL

以下は、zookeeper と kafka の jaas ファイルです。

Zookeeper.sasl.jaas.config -

Server {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   user_kafka="kafka";
};

kafka.jaas.config -

Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="kafka"
   password="kafka";
};

Zookeeper と kafka コンテナーを作成し、kafka コンテナー内でコマンドを実行しました -

/opt/kafka_2.13-2.6.0/bin # ./kafka-acls.sh --authorizer-properties zookeeper.connect=zookeeper:2181 --add --allow-principal User:alice --producer --topic testtopic
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=testtopic, patternType=LITERAL)`: 
    (principal=User:alice, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:alice, host=*, operation=WRITE, permissionType=ALLOW)
    (principal=User:alice, host=*, operation=CREATE, permissionType=ALLOW) 

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=testtopic, patternType=LITERAL)`: 
    (principal=User:alice, host=*, operation=DESCRIBE, permissionType=ALLOW)
    (principal=User:alice, host=*, operation=WRITE, permissionType=ALLOW)
    (principal=User:alice, host=*, operation=CREATE, permissionType=ALLOW)

しかし、goコードから(saramaを使用して)イベントを生成しようとすると、エラーが発生します

kafka server: In the middle of a leadership election, there is currently no leader for this partition and hence it is unavailable for writes.

私の行くコードは -

package main

import "github.com/Shopify/sarama"

var brokers = []string{"127.0.0.1:9092"}

func newProducer() (sarama.SyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Return.Successes = true
    config.Net.SASL.User = "alice"
    config.Net.SASL.Password = "alice-secret"
    config.Net.SASL.Handshake = true
    config.Net.SASL.Enable = true
    producer, err := sarama.NewSyncProducer(brokers, config)

    return producer, err
}

func prepareMessage(topic, message string) *sarama.ProducerMessage {
    msg := &sarama.ProducerMessage{
        Topic:     topic,
        Partition: -1,
        Value:     sarama.StringEncoder(message),
    }

    return msg
}

func panicOnError(err error) {
    if err != nil {
        panic(err)
    }
}

func main() {
    producer, err := newProducer()
    panicOnError(err)
    msg := prepareMessage("testtopic", `{"key":"value"}`)
    _, _, err = producer.SendMessage(msg)
    panicOnError(err)
}

--bootstrap-server (command - ./kafka-acls.sh --bootstrap-server localhost:9092 --add --allow-principal User:alice --producer --topic testtopic) 引数を指定して kafka-acls.sh も試しましたが、スクリプトが動かなくなり、kafka docker ログで認証エラーを確認できます -

[2021-05-29 16:27:46,288] INFO [SocketServer brokerId=1002] Failed authentication with /127.0.0.1 (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

PS: SASL のみ (ACL なし) を使用すると、すべてが正常に機能します。

今、私は acl 部分で立ち往生しています。誰かが私が見逃しているアイデアを持っていますか(おそらくzookeeperまたはkafka configで)?

どんな助けでも大歓迎です。前もって感謝します。

4

1 に答える 1

0

最初の問題については、次の提案を試してみます https://github.com/Shopify/sarama/issues/272

2 番目の問題については、コマンド ラインに --command-config /path/cmd.cfg を追加する必要があります。

mechainsem SASL など、ブローカーに接続するための管理クライアント プロパティを示します...

KAFKA_OPTS jaas ファイルの設定

また、jaas ファイルには、KafkaClient とユーザー、パスワードを指定して、PLAIN 認証方式でブローカーに接続する必要があります。

于 2021-05-29T21:59:21.560 に答える