Reset SASL state on disconnect

kafka/conn.py

@@ -271,12 +271,10 @@ def __init__(self, host, port, afi, **configs):
         assert self.config['security_protocol'] in self.SECURITY_PROTOCOLS, (
             'security_protocol must be in ' + ', '.join(self.SECURITY_PROTOCOLS))
 
-        self._sasl_mechanism = None
         if self.config['security_protocol'] in ('SSL', 'SASL_SSL'):
             assert ssl_available, "Python wasn't built with SSL support"
 
-        if self.config['security_protocol'] in ('SASL_PLAINTEXT', 'SASL_SSL'):
-            self._sasl_mechanism = get_sasl_mechanism(self.config['sasl_mechanism'])(**self.config)
+        self._init_sasl_mechanism()
 
         # This is not a general lock / this class is not generally thread-safe yet
         # However, to avoid pushing responsibility for maintaining
@@ -312,6 +310,12 @@ def __init__(self, host, port, afi, **configs):
                                                     self.config['metric_group_prefix'],
                                                     self.node_id)
 
+    def _init_sasl_mechanism(self):
+        if self.config['security_protocol'] in ('SASL_PLAINTEXT', 'SASL_SSL'):
+            self._sasl_mechanism = get_sasl_mechanism(self.config['sasl_mechanism'])(**self.config)
+        else:
+            self._sasl_mechanism = None
+
     def _dns_lookup(self):
         self._gai = dns_lookup(self.host, self.port, self.afi)
         if not self._gai:
@@ -747,6 +751,7 @@ def _send_sasl_authenticate(self, sasl_auth_bytes):
             request = SaslAuthenticateRequest[0](sasl_auth_bytes)
             self._send(request, blocking=True)
         else:
+            log.debug('Sending %d raw sasl auth bytes to server', len(sasl_auth_bytes))
             try:
                 self._send_bytes_blocking(Int32.encode(len(sasl_auth_bytes)) + sasl_auth_bytes)
             except (ConnectionError, TimeoutError) as e:
@@ -787,6 +792,7 @@ def _recv_sasl_authenticate(self):
             return response.auth_bytes
         else:
             # unframed bytes w/ SaslHandhake v0
+            log.debug('Received %d raw sasl auth bytes from server', nbytes)
             return data[4:]
 
     def _sasl_authenticate(self, future):
@@ -930,6 +936,7 @@ def close(self, error=None):
             self._update_reconnect_backoff()
             self._api_versions_future = None
             self._sasl_auth_future = None
+            self._init_sasl_mechanism()
             self._protocol = KafkaProtocol(
                 client_id=self.config['client_id'],
                 api_version=self.config['api_version'])