Closed
Description
In producer/kafka.py in KafkaProducer._wait_on_metadata, if an exception was identified in the future, it is not propagated.
The revised code should probably be something like:
future = self._metadata.request_update()
future.add_both(lambda e, *args: e.set(), metadata_event)
self._sender.wakeup()
metadata_event.wait(max_wait - elapsed)
if future.exception: # Propagate exception
raise future.exception
Metadata
Metadata
Assignees
Labels
No labels