Send SASL message when state is Finished(Yes)

This was a bug in the existing implementation that sometimes caused
issues, but that some kafka brokers appear to tolerate.

Author: einarmo

src/messenger.rs

@@ -605,12 +605,16 @@ where
         loop {
             let mut to_sent = Cursor::new(Vec::new());
             let state = session.step(data_received.as_deref(), &mut to_sent)?;
+
+            if state.has_sent_message() {
+                let authentication_response =
+                    self.sasl_authentication(to_sent.into_inner()).await?;
+                data_received = Some(authentication_response.auth_bytes.0);
+            }
+
             if !state.is_running() {
                 break;
             }
-
-            let authentication_response = self.sasl_authentication(to_sent.into_inner()).await?;
-            data_received = Some(authentication_response.auth_bytes.0);
         }
 
         Ok(())