kafka消费不到数据问题

kafka集群搭建正常，通过console都能正常生产和消费消息，但是通过JAVA程序就是读取不到消息，更换group都尝试过了
package test;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class KafkaConsumer extends Thread {

private String topic;

public KafkaConsumer(String topic){
    super();
    this.topic=topic;
}

@Override
public void run(){
    //通过properties设置了Consumer的参数，并且创建了连接器，连接到Kafaka
    ConsumerConnector consumer = createConsumer();
    //Map作用指定获取的topic以及partition
    Map<String,Integer> topicCountMap = new HashMap<String,Integer>();
    topicCountMap.put(topic, 3);
    //consumer连接器获取消息
    Map<String,List<KafkaStream<byte[],byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
    //获取对应的topic中的某一个partition中的数据
    KafkaStream<byte[],byte[]> kafkaStream = messageStreams.get(topic).get(0);

    ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
    while(iterator.hasNext()){
        byte[] message = iterator.next().message();
        System.out.println("message is:"+new String(message));
    }

}

private ConsumerConnector createConsumer(){
    Properties properties = new Properties();
    properties.put("zookeeper.connect", "XXX:2181");
    properties.put("auto.offset.reset", "smallest");//读取旧数据  
    properties.put("group.id", "333fcdcd");
    return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
}

public static void main(String[] args) {
    new KafkaConsumer("testtest").start();
}

}