Fix crash in producer when required_acks == 0

pykafka/producer.py

@@ -294,6 +294,10 @@ def _get_partition_msgs(partition_id, req):
 
         try:
             response = owned_broker.broker.produce_messages(req)
+            if self._required_acks == 0:  # and thus, `response` is None
+                owned_broker.increment_messages_pending(
+                    -1 * len(message_batch))
+                return
             to_retry = []
             for topic, partitions in iteritems(response.topics):
                 for partition, presponse in iteritems(partitions):

tests/pykafka/test_producer.py

@@ -93,5 +93,20 @@ def test_async_produce_thread_exception(self):
         while self.consumer.consume() is not None:
             time.sleep(.05)
 
+    def test_required_acks(self):
+        """Test with non-default values for `required_acks`
+
+        See #278 for a related bug.  Here, we only test that no exceptions
+        occur (hence `sync=True`, which would surface most exceptions)
+        """
+        kwargs = dict(linger_ms=1, sync=True, required_acks=0)
+        prod = self.client.topics[self.topic_name].get_producer(**kwargs)
+        prod.produce(uuid4().bytes)
+
+        kwargs["required_acks"] = -1
+        prod = self.client.topics[self.topic_name].get_producer(**kwargs)
+        prod.produce(uuid4().bytes)
+
+
 if __name__ == "__main__":
     unittest2.main()