kafka producer发送消息send报错?

idea写的producer发送消息给linux中的消费者,zookeeper.kafka.消费者都以启动,但是send报错:

代码信息:

public class ProducerKafka extends Thread{
    private final Producer<Integer, String> producer;
    private final String topic;
    private final Properties prop = new Properties();

    public ProducerKafka(String topic) {
//        prop.put("zookeeper.connect", "192.168.111.128:2181");//声明zk
        prop.put("serializer.class", StringEncoder.class.getName());
        prop.put("metadata.broker.list", "192.168.111.128:9092");// 声明kafka broker
        producer = new Producer<Integer, String>(new ProducerConfig(prop));
        System.out.println(111);
        this.topic = topic;
    }
    public void run(){
        int messageNo = 1;
        while (true){
            String messageStr = new String("Message_"+messageNo);
            System.out.println("Send:"+messageStr);
            KeyedMessage<Integer, String> keyedMessage = new KeyedMessage<Integer, String>(topic,messageStr);
            producer.send(keyedMessage);
            messageNo++;
            try{
                sleep(3000);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        ProducerKafka producerKafka = new ProducerKafka("test");
        producerKafka.start();
    }
}

报错信息:

Exception in thread "Thread-0" java.util.EmptyStackException

    at ProducerKafka.run(ProducerKafka.java:27)
    Process finished with exit code 0