Cluster class does not support IPv6 raw address/port

[takaomag]

Environment

PyKafka version: 2.8.0 Kafka version: 2.0.0

Procedure

>>> from pykafka import KafkaClient
>>> 
>>> X_HOST_IPADDR='2406:da14:555:b701:247f:9587:88a4:f6d3'
>>> client = KafkaClient(hosts=f'[{X_HOST_IPADDR}]:9090,[{X_HOST_IPADDR}]:9091,[{X_HOST_IPADDR}]:9092')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/project/test-kafka-1/lib/python3.7/site-packages/pykafka/client.py", line 142, in __init__
    broker_version=broker_version)
  File "/opt/project/test-kafka-1/lib/python3.7/site-packages/pykafka/cluster.py", line 215, in __init__
    self.update()
  File "/opt/project/test-kafka-1/lib/python3.7/site-packages/pykafka/cluster.py", line 518, in update
    metadata = self._get_metadata()
  File "/opt/project/test-kafka-1/lib/python3.7/site-packages/pykafka/cluster.py", line 305, in _get_metadata
    'Unable to connect to a broker to fetch metadata. See logs.')
pykafka.exceptions.NoBrokersAvailableError: Unable to connect to a broker to fetch metadata. See logs.

Cause

https://github.com/Parsely/pykafka/blob/master/pykafka/cluster.py#L323-L325 Since the IPv6 address/port format is such like '[2406:da14:555:b701:247f:9587:88a4:f6d3]:9092', simple splitting ':' is not enough. I guess it should be as follows.

broker_connects = [
    (
        broker_str.split(']', 1)[0].lstrip('[') if '[' in broker_str else broker_str.split(':', 1)[0],
        broker_str.rsplit(':', 1)[-1].rstrip('/'),
    )
    for broker_str in self._seed_hosts.split(',')
]

I am not sure the other codes have same issue or not.

[emmettbutler]

Thank you for this report, @takaomag. This should definitely be fixed.