fix: SASL/PLAIN

The previous fix did not work. This is now definitely verified to be
working and also makes coherent sense.

.circleci/config.yml

@@ -243,9 +243,10 @@ jobs:
           - 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_CLIENT_SASL_MECHANISM=PLAIN
+          - 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_INTER_BROKER_LISTENER_NAME=CLIENT
           - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
       - image: docker.io/bitnami/kafka:3
@@ -279,6 +280,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

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();