Use class config for Kafka result bolt.

src/birding/bolt.py

@@ -2,7 +2,6 @@
 
 import json
 
-from pykafka import KafkaClient
 from streamparse.bolt import Bolt
 
 from .config import get_config, import_name
@@ -99,7 +98,8 @@ def initialize(self, conf, ctx):
         2. Prepare Kafka producer for `tweet` topic.
         """
         config = get_config()['ResultTopicBolt']
-        self.client = KafkaClient(hosts=config['hosts'])
+        kafka_class = import_name(config['kafka_class'])
+        self.client = kafka_class(**config['kafka_init'])
         self.topic = self.client.topics[config['topic']]
         self.producer = self.topic.get_producer()
 

src/birding/config.py

@@ -45,7 +45,9 @@
       index: tweet
       doc_type: tweet
     ResultTopicBolt:
-      hosts: 127.0.0.1:9092 # comma-separated list of hosts
+      kafka_class: pykafka.KafkaClient
+      kafka_init:
+        hosts: 127.0.0.1:9092 # comma-separated list of hosts
       topic: tweet
     Appendix: {}
 
@@ -80,7 +82,8 @@
         index = tv.String(),
         doc_type = tv.String()),
     ResultTopicBolt = tv.SchemaMapping().of(
-        hosts = tv.String(),
+        kafka_class = tv.String(),
+        kafka_init = tv.StrMapping().of(tv.Passthrough()),
         topic = tv.String()),
     Appendix = tv.Passthrough())
 

src/birding/follow.py

@@ -7,24 +7,24 @@
 from contextlib import contextmanager
 from time import sleep
 
-from pykafka import KafkaClient
 from pykafka.exceptions import KafkaException
 
-from .config import get_config
+from .config import get_config, import_name
 from .search import SearchManager
 
 
 def follow_topic_from_config():
     """Read kafka config, then dispatch to `follow_topic`."""
     config = get_config()['ResultTopicBolt']
-    return follow_topic(config['hosts'], config['topic'])
+    kafka_class = import_name(config['kafka_class'])
+    return follow_topic(kafka_class, config['topic'], **config['kafka_init'])
 
 
-def follow_topic(hosts, name, retry_interval=1):
+def follow_topic(kafka_class, name, retry_interval=1, **kafka_init):
     """Dump each message from kafka topic to stdio."""
     while True:
         try:
-            client = KafkaClient(hosts=hosts)
+            client = kafka_class(**kafka_init)
             topic = client.topics[name]
             consumer = topic.get_simple_consumer(reset_offset_on_start=True)
         except Exception as e: