openstack/deb-python-kafka
Code Issues Proposed changes

Try to bind to a new port on fixture failure retries

Browse Source
This commit is contained in:
Dana Powers
2016-08-04 15:16:10 -07:00
parent bccac7bca1
commit c061b9b648
1 changed files with 21 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
test/fixtures.py
View File

@@ -105,7 +105,7 @@ class ZookeeperFixture(Fixture):
(host, port) = (parse.hostname, parse.port) (host, port) = (parse.hostname, parse.port)
fixture = ExternalService(host, port) fixture = ExternalService(host, port)
else: else:
(host, port) = ("127.0.0.1", get_open_port()) (host, port) = ("127.0.0.1", None)
fixture = cls(host, port) fixture = cls(host, port)
fixture.open() fixture.open()
@@ -124,21 +124,18 @@ class ZookeeperFixture(Fixture):
return env return env
def out(self, message): def out(self, message):
log.info("*** Zookeeper [%s:%d]: %s", self.host, self.port, message) log.info("*** Zookeeper [%s:%s]: %s", self.host, self.port or '(auto)', message)
def open(self): def open(self):
self.tmp_dir = tempfile.mkdtemp() self.tmp_dir = tempfile.mkdtemp()
self.out("Running local instance...") self.out("Running local instance...")
log.info(" host = %s", self.host) log.info(" host = %s", self.host)
log.info(" port = %s", self.port) log.info(" port = %s", self.port or '(auto)')
log.info(" tmp_dir = %s", self.tmp_dir) log.info(" tmp_dir = %s", self.tmp_dir)
# Generate configs # Configure Zookeeper child process
template = self.test_resource("zookeeper.properties") template = self.test_resource("zookeeper.properties")
properties = os.path.join(self.tmp_dir, "zookeeper.properties") properties = os.path.join(self.tmp_dir, "zookeeper.properties")
self.render_template(template, properties, vars(self))
# Configure Zookeeper child process
args = self.kafka_run_class_args("org.apache.zookeeper.server.quorum.QuorumPeerMain", properties) args = self.kafka_run_class_args("org.apache.zookeeper.server.quorum.QuorumPeerMain", properties)
env = self.kafka_run_class_env() env = self.kafka_run_class_env()
@@ -148,13 +145,12 @@ class ZookeeperFixture(Fixture):
backoff = 1 backoff = 1
end_at = time.time() + max_timeout end_at = time.time() + max_timeout
tries = 1 tries = 1
auto_port = (self.port is None)
while time.time() < end_at: while time.time() < end_at:
self.out('Attempting to start (try #%d)' % tries) if auto_port:
try: self.port = get_open_port()
os.stat(properties) self.out('Attempting to start on port %d (try #%d)' % (self.port, tries))
except: self.render_template(template, properties, vars(self))
log.warning('Config %s not found -- re-rendering', properties)
self.render_template(template, properties, vars(self))
self.child = SpawnedService(args, env) self.child = SpawnedService(args, env)
self.child.start() self.child.start()
timeout = min(timeout, max(end_at - time.time(), 0)) timeout = min(timeout, max(end_at - time.time(), 0))
@@ -194,8 +190,6 @@ class KafkaFixture(Fixture):
(host, port) = (parse.hostname, parse.port) (host, port) = (parse.hostname, parse.port)
fixture = ExternalService(host, port) fixture = ExternalService(host, port)
else: else:
if port is None:
port = get_open_port()
# force IPv6 here because of a confusing point: # force IPv6 here because of a confusing point:
# #
# - if the string "localhost" is passed, Kafka will *only* bind to the IPv4 address of localhost # - if the string "localhost" is passed, Kafka will *only* bind to the IPv4 address of localhost
@@ -245,7 +239,7 @@ class KafkaFixture(Fixture):
return env return env
def out(self, message): def out(self, message):
log.info("*** Kafka [%s:%d]: %s", self.host, self.port, message) log.info("*** Kafka [%s:%s]: %s", self.host, self.port or '(auto)', message)
def open(self): def open(self):
if self.running: if self.running:
@@ -255,7 +249,7 @@ class KafkaFixture(Fixture):
self.tmp_dir = tempfile.mkdtemp() self.tmp_dir = tempfile.mkdtemp()
self.out("Running local instance...") self.out("Running local instance...")
log.info(" host = %s", self.host) log.info(" host = %s", self.host)
log.info(" port = %s", self.port) log.info(" port = %s", self.port or '(auto)')
log.info(" transport = %s", self.transport) log.info(" transport = %s", self.transport)
log.info(" broker_id = %s", self.broker_id) log.info(" broker_id = %s", self.broker_id)
log.info(" zk_host = %s", self.zk_host) log.info(" zk_host = %s", self.zk_host)
@@ -269,12 +263,6 @@ class KafkaFixture(Fixture):
os.mkdir(os.path.join(self.tmp_dir, "logs")) os.mkdir(os.path.join(self.tmp_dir, "logs"))
os.mkdir(os.path.join(self.tmp_dir, "data")) os.mkdir(os.path.join(self.tmp_dir, "data"))
# Generate configs
template = self.test_resource("kafka.properties")
properties = os.path.join(self.tmp_dir, "kafka.properties")
self.render_template(template, properties, vars(self))
# Party!
self.out("Creating Zookeeper chroot node...") self.out("Creating Zookeeper chroot node...")
args = self.kafka_run_class_args("org.apache.zookeeper.ZooKeeperMain", args = self.kafka_run_class_args("org.apache.zookeeper.ZooKeeperMain",
"-server", "%s:%d" % (self.zk_host, self.zk_port), "-server", "%s:%d" % (self.zk_host, self.zk_port),
@@ -292,6 +280,8 @@ class KafkaFixture(Fixture):
self.out("Done!") self.out("Done!")
# Configure Kafka child process # Configure Kafka child process
properties = os.path.join(self.tmp_dir, "kafka.properties")
template = self.test_resource("kafka.properties")
args = self.kafka_run_class_args("kafka.Kafka", properties) args = self.kafka_run_class_args("kafka.Kafka", properties)
env = self.kafka_run_class_env() env = self.kafka_run_class_env()
@@ -300,13 +290,15 @@ class KafkaFixture(Fixture):
backoff = 1 backoff = 1
end_at = time.time() + max_timeout end_at = time.time() + max_timeout
tries = 1 tries = 1
auto_port = (self.port is None)
while time.time() < end_at: while time.time() < end_at:
self.out('Attempting to start (try #%d)' % tries) # We have had problems with port conflicts on travis
try: # so we will try a different port on each retry
os.stat(properties) # unless the fixture was passed a specific port
except: if auto_port:
log.warning('Config %s not found -- re-rendering', properties) self.port = get_open_port()
self.render_template(template, properties, vars(self)) self.out('Attempting to start on port %d (try #%d)' % (self.port, tries))
self.render_template(template, properties, vars(self))
self.child = SpawnedService(args, env) self.child = SpawnedService(args, env)
self.child.start() self.child.start()
timeout = min(timeout, max(end_at - time.time(), 0)) timeout = min(timeout, max(end_at - time.time(), 0))