fix: SASL/PLAIN

.circleci/config.yml

@@ -226,39 +226,49 @@ jobs:
         name: zookeeper
         environment:
           - ALLOW_ANONYMOUS_LOGIN=yes
-      - image: docker.io/bitnami/kafka:3
+      - image: docker.io/bitnami/kafka:3.9.0
         name: kafka-0
         environment:
           - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
           - KAFKA_CFG_BROKER_ID=0
           - ALLOW_PLAINTEXT_LISTENER=yes
-          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-          - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
-          - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-0:9092,EXTERNAL://kafka-0:9093
+          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT,SECURE:SASL_PLAINTEXT
+          - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093,SECURE://:9094
+          - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-0:9092,EXTERNAL://kafka-0:9093,SECURE://kafka-0:9094
           - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
           - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
-      - image: docker.io/bitnami/kafka:3
+          - KAFKA_CLIENT_USERS=admin
+          - KAFKA_CLIENT_PASSWORDS=admin-secret
+          - KAFKA_CLIENT_LISTENER_NAME=SECURE
+      - image: docker.io/bitnami/kafka:3.9.0
         name: kafka-1
         environment:
           - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
           - KAFKA_CFG_BROKER_ID=1
           - ALLOW_PLAINTEXT_LISTENER=yes
-          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-          - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
-          - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-1:9092,EXTERNAL://kafka-1:9093
+          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT,SECURE:SASL_PLAINTEXT
+          - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093,SECURE://:9094
+          - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-1:9092,EXTERNAL://kafka-1:9093,SECURE://kafka-1:9094
+          - KAFKA_CFG_SASL_ENABLED_MECHANISMS=PLAIN
           - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
           - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
-      - image: docker.io/bitnami/kafka:3
+          - KAFKA_CLIENT_USERS=admin
+          - KAFKA_CLIENT_PASSWORDS=admin-secret
+          - KAFKA_CLIENT_LISTENER_NAME=SECURE
+      - image: docker.io/bitnami/kafka:3.9.0
         name: kafka-2
         environment:
           - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
           - KAFKA_CFG_BROKER_ID=2
           - ALLOW_PLAINTEXT_LISTENER=yes
-          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-          - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
-          - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-2:9092,EXTERNAL://kafka-2:9093
+          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT,SECURE:SASL_PLAINTEXT
+          - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093,SECURE://:9094
+          - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-2:9092,EXTERNAL://kafka-2:9093,SECURE://kafka-2:9094
           - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
           - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
+          - KAFKA_CLIENT_USERS=admin
+          - KAFKA_CLIENT_PASSWORDS=admin-secret
+          - KAFKA_CLIENT_LISTENER_NAME=SECURE
       - image: serjs/go-socks5-proxy
         name: proxy
     resource_class: xlarge  # use of a smaller executor tends crashes on link
@@ -279,6 +289,7 @@ jobs:
       # Don't use the first node here since this is likely the controller and we want to ensure that we automatically
       # pick the controller for certain actions (e.g. topic creation) and don't just get lucky.
       KAFKA_CONNECT: "invalid:9093,kafka-1:9093"
+      KAFKA_SASL_CONNECT: kafka-1:9094
       SOCKS_PROXY: "proxy:1080"
     steps:
       - checkout

docker-compose-kafka.yml

@@ -11,7 +11,7 @@ services:
     volumes:
       - zookeeper_data:/bitnami/zookeeper
   kafka-0:
-    image: docker.io/bitnami/kafka:3
+    image: docker.io/bitnami/kafka:3.9.0
     ports:
       - "9010:9010"
       - "9096:9096"
@@ -24,13 +24,16 @@ services:
       - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-0:9000,EXTERNAL://localhost:9010,FOR_PROXY://kafka-0:9020,SECURE://localhost:9096
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
       - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
+      - KAFKA_CLIENT_USERS=admin
+      - KAFKA_CLIENT_PASSWORDS=admin-secret
+      - KAFKA_CLIENT_LISTENER_NAME=SECURE
     volumes:
       - kafka_0_data:/bitnami/kafka
       - ./kafka_jaas.conf:/opt/bitnami/kafka/config/kafka_jaas.conf
     depends_on:
       - zookeeper
   kafka-1:
-    image: docker.io/bitnami/kafka:3
+    image: docker.io/bitnami/kafka:3.9.0
     ports:
       - "9011:9011"
       - "9097:9097"
@@ -43,13 +46,16 @@ services:
       - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-1:9000,EXTERNAL://localhost:9011,FOR_PROXY://kafka-1:9021,SECURE://localhost:9097
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
       - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
+      - KAFKA_CLIENT_USERS=admin
+      - KAFKA_CLIENT_PASSWORDS=admin-secret
+      - KAFKA_CLIENT_LISTENER_NAME=SECURE
     volumes:
       - kafka_1_data:/bitnami/kafka
       - ./kafka_jaas.conf:/opt/bitnami/kafka/config/kafka_jaas.conf
     depends_on:
       - zookeeper
   kafka-2:
-    image: docker.io/bitnami/kafka:3
+    image: docker.io/bitnami/kafka:3.9.0
     ports:
       - "9012:9012"
       - "9098:9098"
@@ -62,6 +68,9 @@ services:
       - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-2:9000,EXTERNAL://localhost:9012,FOR_PROXY://kafka-2:9022,SECURE://localhost:9098
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
       - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
+      - KAFKA_CLIENT_USERS=admin
+      - KAFKA_CLIENT_PASSWORDS=admin-secret
+      - KAFKA_CLIENT_LISTENER_NAME=SECURE
     volumes:
       - kafka_2_data:/bitnami/kafka
       - ./kafka_jaas.conf:/opt/bitnami/kafka/config/kafka_jaas.conf

src/messenger.rs

@@ -610,7 +610,9 @@ where
                 let authentication_response =
                     self.sasl_authentication(to_sent.into_inner()).await?;
                 data_received = Some(authentication_response.auth_bytes.0);
-            } else {
+            }
+
+            if state.is_finished() {
                 break;
             }
         }

tests/client.rs

@@ -44,6 +44,10 @@ async fn test_sasl() {
         .sasl_config(rskafka::client::SaslConfig::Plain(
             rskafka::client::Credentials::new("admin".to_string(), "admin-secret".to_string()),
         ))
+        .backoff_config(BackoffConfig {
+            deadline: Some(Duration::from_secs(1)),
+            ..Default::default()
+        })
         .build()
         .await
         .unwrap();