java向kafka发送消息

/*** Kafka配置文件*/
public class KafkaProperties {public static final String ZOOKEEPER = "192.168.101.101:2181";public  static final String TOPIC = "test";public static final String BROKER_LIST = "192.168.101.101:9092";
}

/*** Kafka 生产者*/public class KafkaProducer extends Thread{private String topic;private Producer<Integer, String> producer;public KafkaProducer(String topic){this.topic = topic;Properties properties = new Properties();properties.put("metadata.broker.list", KafkaProperties.BROKER_LIST);properties.put("serializer.class", "kafka.serializer.StringEncoder");properties.put("request.required.acks", "1");producer = new Producer<Integer, String>(new ProducerConfig(properties));}@Overridepublic void run(){int count = 1;while (true){String message = "messate:" + count;this.producer.send(new KeyedMessage<Integer, String>(topic, message));System.out.println(message);count ++;try {Thread.sleep(2000);} catch (Exception e) {e.printStackTrace();}}}
}

/*** 测试发送消息*/
public class KafkaTest {public static void main(String[] args) {new KafkaProducer(KafkaProperties.TOPIC).start();}
}

发送失败出现：
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries

修改Kafka配置文件$KAFKA_HOME/config/server.properties