final long timeout, final TimeUnit timeUnit) throws MetaClientException, InterruptedException {
if (!this.publishedTopics.contains(topic)) {
this.producerZooKeeper.publishTopic(topic, this);
this.publishedTopics.add(topic);
}
final Broker broker =
new Broker(partition.getBrokerId(), this.producerZooKeeper.selectBroker(topic, partition));
final TopicPartitionRegInfo topicPartitionRegInfo = new TopicPartitionRegInfo(topic, partition, offset);
return this.fetch(new FetchRequest(broker, 0, topicPartitionRegInfo, maxSize), timeout, timeUnit);
}