Test reliability: Retry produce on retriable broker errors

[twm]

This should fix this failure:

ERROR: afkak.test.test_consumer_integration.TestConsumerIntegration.test_consumer_restart
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/internet/defer.py", line 151, in maybeDeferred
    result = f(*args, **kw)
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/internet/utils.py", line 201, in runWithWarningsSuppressed
    reraise(exc_info[1], exc_info[2])
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/internet/utils.py", line 197, in runWithWarningsSuppressed
    result = f(*a, **kw)
  File "/home/travis/build/ciena/afkak/afkak/test/testutil.py", line 88, in wrapper
    return func(self)
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/nose/twistedtools.py", line 143, in errback
    failure.raiseException()
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/internet/defer.py", line 1416, in _inlineCallbacks
    result = result.throwExceptionIntoGenerator(g)
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/python/failure.py", line 491, in throwExceptionIntoGenerator
    return g.throw(self.type, self.value, self.tb)
  File "/home/travis/build/ciena/afkak/afkak/test/test_consumer_integration.py", line 191, in test_consumer_restart
    sent_messages = yield self.send_messages(self.partition, range(0, 100))
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/internet/defer.py", line 1416, in _inlineCallbacks
    result = result.throwExceptionIntoGenerator(g)
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/python/failure.py", line 491, in throwExceptionIntoGenerator
    return g.throw(self.type, self.value, self.tb)
  File "/home/travis/build/ciena/afkak/afkak/test/test_consumer_integration.py", line 48, in send_messages
    resp, = yield self.client.send_produce_request([produce])
  File "/home/travis/build/ciena/afkak/.tox/py27-int-snappy-murmur/lib/python2.7/site-packages/twisted/internet/defer.py", line 1418, in _inlineCallbacks
    result = g.send(result)
  File "/home/travis/build/ciena/afkak/afkak/client.py", line 465, in send_produce_request
    returnValue(self._handle_responses(resps, fail_on_error, callback))
  File "/home/travis/build/ciena/afkak/afkak/client.py", line 566, in _handle_responses
    _check_error(resp)
  File "/home/travis/build/ciena/afkak/afkak/common.py", line 774, in _check_error
    raise error
UnknownTopicOrPartitionError

Seen on this build of #46.

[twm]