Set the current host in the SASL configs (#2633)

Co-authored-by: Sebastian.Meinhardt <Sebastian.Meinhardt@vodafone.com>

Author: SeppMe

kafka/conn.py

@@ -256,6 +256,7 @@ def __init__(self, host, port, afi, **configs):
         for key in self.config:
             if key in configs:
                 self.config[key] = configs[key]
+        self.config['host'] = host
 
         self.node_id = self.config.pop('node_id')
 

test/test_conn.py

@@ -386,3 +386,13 @@ def test_maybe_throttle(conn):
 
         time.return_value = 3000
         assert not conn.throttled()
+
+
+def test_host_in_sasl_config():
+    hostname = 'example.org'
+    port = 9092
+    for security_protocol in ('SASL_PLAINTEXT', 'SASL_SSL'):
+        with mock.patch("kafka.conn.get_sasl_mechanism") as get_sasl_mechanism:
+            BrokerConnection(hostname, port, socket.AF_UNSPEC, security_protocol=security_protocol)
+            call_config = get_sasl_mechanism.mock_calls[1].kwargs
+            assert call_config['host'] == hostname