我试图用java中的生产者代码发送消息，当我运行这段代码时，它给了我一个异常。

Exception in thread "Thread-0" kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
    at kafka.producer.Producer.send(Producer.scala:77)
    at kafka.javaapi.producer.Producer.send(Producer.scala:33)
    at javaFile.Producer.run(Producer.java:36)
    11:00 AM

这是我的制片人代码。。

import java.util.Properties;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class Producer extends Thread
{
      private final kafka.javaapi.producer.Producer<Integer, String> producer;
      private final String topic;
      private final Properties props = new Properties();
      public Producer(String topic)
      {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", "172.29.104.200:9092");
        //props.put("num.partitions", 1);
        props.put("producer.type", "sync");
        // Use random partitioner. Don't need the key type. Just set it to Integer.
        // The message is of type String.
        producer = new kafka.javaapi.producer.Producer<Integer, String>(new ProducerConfig(props));
        this.topic = topic;
      }

      public void run() {
        int messageNo = 1;
        while(true)
        {
          String messageStr = new String("Message_" + messageNo);
          producer.send(new KeyedMessage<Integer, String>(topic, messageStr));
          messageNo++;
        }
      }
    }

有人能告诉我什么时候会有这个例外吗？