[解決済み] SASL ハンドシェイク中に METADATA タイプの予期しない Kafka リクエストが発生する。

質問

SASL Plainを使用してKafka Java ClientをKafkaブローカーに接続しようとしています。しかし、Producerからメッセージを送信しようとすると、Kafka Serverは以下のエラーをログに記録します。

[2020-04-30 14:48:14,955] INFO [SocketServer brokerId=0] Failed authentication with /127.0.0.1 (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

見た目では、producerはSASLハンドシェイクの前に、メタデータ・リクエストを送信しようとします。メッセージを送る前にハンドシェイクを行うにはどうしたらよいでしょうか？

以下は私の kafka_server_jaas.conf ファイルで、Kafka Serverで使用されています。

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret";
};

Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret";
};

以下は、私の zookeeper_jaas.conf ファイルで、zookeeperに使用されています。

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret";
};

私のJavaプロデューサーでは、以下のプロパティを設定しました。

Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin_secret\"");
properties.put("sasl.mechanisms", "PLAIN");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer kafkaProducer = new KafkaProducer(properties);

何か間違ったことをしているのでしょうか？

どうすればいいですか？

を指定する必要があります。 security.protocol そうでない場合、Kafkaクライアントはデフォルトで SASL .

クライアントのプロパティに、追加します。

properties.put("security.protocol", "SASL_SSL");

また SASL_PLAINTEXT を使用することは推奨されませんが PLAIN 機構を SASL_PLAINTEXT というのは、事実上、パスワードは平文で交換されるからです。