Fix thread-safety of Python VerifiableConsumer class
Author: Ewen Cheslack-Postava
Reviewers: Jason Gustafson
Closes from ewencp/kafka-4673-verifiable-consumer-thread-safety
@@ -158,10 +158,11 @@ class VerifiableConsumer(KafkaPathResolverMixin, BackgroundThreadService):
node.version = version
def _worker(self, idx, node):
+ with self.lock:
if node not in self.event_handlers:
self.event_handlers[node] = ConsumerEventHandler(node)
-
handler = self.event_handlers[node]
+
node.account.ssh("mkdir -p %s" % VerifiableConsumer.PERSISTENT_ROOT, allow_fail=False)
# Create and upload log properties
@@ -267,6 +268,7 @@ class VerifiableConsumer(KafkaPathResolverMixin, BackgroundThreadService):
for pid in self.pids(node):
node.account.signal(pid, sig, allow_fail)
+ with self.lock:
self.event_handlers[node].handle_kill_process(clean_shutdown)
def stop_node(self, node, clean_shutdown=True):
@@ -293,6 +295,7 @@ class VerifiableConsumer(KafkaPathResolverMixin, BackgroundThreadService):
return None
def owner(self, tp):
+ with self.lock:
for handler in self.event_handlers.itervalues():
if tp in handler.current_assignment():
return handler.node
