Error throw

tests/test_cons_add_remove.py

@@ -25,7 +25,7 @@ def new_start(key,value):
 
 print " Adding topic " + topic
 
-fc = FeedConsumer("localhost:9092")
+fc = FeedConsumer("localhost:9092", group ='TestGroup')
 fc.add_topic(topic,dummy)
 fc.start()
 fc.add_topic("test1",new_start)

tests/test_green_cons.py

@@ -15,7 +15,7 @@
 logger = logging.getLogger('kafka')
 logger.setLevel(logging.INFO)
 
-srv = os.getenv("GOMSG_SRV","localhost:9092")
+srv = os.getenv("GOMSG_SRV", "localhost:9092")
 
 def dummy(key, value):
     print "[GREEN]  %s" % (str(value))
@@ -24,7 +24,7 @@ def callback_xx(key, value):
     print "[GREEN_XX]  %s" % (str(value))
 
 
-fc = GreenFeedConsumer(srv,  group ='TestGroup')
+fc = GreenFeedConsumer(srv, group ='TestGroup')
 print srv
 print "listing to topic %s"%(topic)
 fc.add_topic(topic, dummy)

tests/test_green_cons_add_remove.py

@@ -25,7 +25,7 @@ def new_start(key,value):
 
 print " Adding topic " + topic
 
-fc = FeedConsumer("localhost:9092")
+fc = FeedConsumer("localhost:9092", group ='TestGroup')
 fc.add_topic(topic,dummy)
 fc.start()
 fc.add_topic("test1",new_start)

tests/test_partitioned_cons.py

@@ -34,9 +34,11 @@ def part2(key,value):
 # Or could have given
 # fc.add_topic(topic, callback_for_1_and_2, (1,2))
 
+# This will pass only if number of partitions for the topic >=2
 fc2 =  FeedConsumer(srv, group ='TestGroup')
 fc2.add_topic(topic,part1, (1,))
 
+# This will pass only if number of partitions for the topic is = 3.
 fc3 =  FeedConsumer(srv, group ='TestGroup')
 fc3.add_topic(topic,part2, (2,))
 

woof/green_consumer.py

@@ -2,7 +2,7 @@
 import threading, logging, gevent, time
 
 from kafka import KafkaConsumer
-
+from kafka.common import KafkaUnavailableError
 log = logging.getLogger("kafka")
 
 class GreenFeedConsumer(threading.Thread):

woof/partitioned_producer.py

@@ -63,6 +63,9 @@ def _next_partition(self, topic, key):
         return self.partitions[topic][self.hash_fn(key) % len(self.partitions[topic])]
 
     def send(self, topic, key, *msg):
+
+        if not msg:
+            raise TypeError("Message field cannot be empty.")
         try:
             topic = kafka_bytestring(topic)
             partition = self._next_partition(topic, key)