Handle New Topic Creation #174
Conversation
| if time.time() > start_time + timeout: | ||
| raise KafkaTimeoutError("Unable to create topic {}".format(topic)) | ||
| self.load_metadata_for_topics(topic) | ||
| time.sleep(.5) |
There was a problem hiding this comment.
sleep time should probably also be a param that users can override
There was a problem hiding this comment.
That's a great idea. I'll get back to this based on the style review in the other conversation as soon as the world isn't literally blowing up at work.
|
@wizzat: are you ok with just merging this as is? The timing parameter is not critical and would be good to clean this up. If you've got a revision in the works I'll wait for it. Otherwise I feel like we should merge now. |
|
There was some discussion about making this automatically happen as it does with the Java client. It turns out that you can tell whether the topic is created and whether or not autocreate is on based on the error returned from the Kafka server when you emit the produce request. The problem is that making a produce request will require some fairly significant restructuring of some internals. I started that work, but it's not anywhere near complete. |
| self.load_metadata_for_topics(topic) | ||
| while not self.has_metadata_for_topic(topic): | ||
| if time.time() > start_time + timeout: | ||
| raise KafkaTimeoutError("Unable to create topic {}".format(topic)) |
Adds ensure_topic_exists to KafkaClient, redirects test case to use
that. Fixes #113 and fixes #150.