in zktraffic/base/util.py [0:0]
def __init__(self, sid, address_str):
"""
See org.apache.zookeeper.server.Quorum.QuorumPeer.QuorumServer
:param sid: int
:param address_str: str
"""
self.sid = sid
server_client_parts = address_str.split(';')
server_parts = server_client_parts[0].split(':')
if len(server_client_parts) > 2 or len(server_parts) < 3 \
or len(server_parts) > 4:
raise QuorumConfig.BadConfig(address_str)
if len(server_client_parts) == 2:
client_parts = server_client_parts[1].split(':')
if len(client_parts) > 2:
raise QuorumConfig.BadConfig(address_str)
self.zk_hostname = client_parts[0] if len(client_parts) == 2 else '0.0.0.0'
try:
self.zk_port = int(client_parts[-1])
except ValueError as e:
raise QuorumConfig.BadConfig(e)
self.zab_fle_hostname = server_parts[0]
try:
self.zab_port = int(server_parts[1])
self.fle_port = int(server_parts[2])
except ValueError as e:
raise QuorumConfig.BadConfig(e)
if len(server_parts) == 4:
self.learner_type = server_parts[3]
if not self.learner_type in ('participant', 'observer'):
raise QuorumConfig.BadConfig(address_str)