我用的是Kafka0.9.0.1。
第一次启动应用程序时，从主题中检索“最新”消息需要20-30秒
我使用了不同的kafka代理（配置不同），但我仍然看到这种行为。后续消息通常没有延迟。
这是预期的行为吗？通过运行这个示例应用程序并将代理/主题名称更改为您自己的设置，您可以在下面清楚地看到这一点

public class KafkaProducerConsumerTest {

    public static final String KAFKA_BROKERS = "...";
    public static final String TOPIC = "...";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        new KafkaProducerConsumerTest().run();
    }

    public void run() throws ExecutionException, InterruptedException {
        Properties consumerProperties = new Properties();
        consumerProperties.setProperty("bootstrap.servers", KAFKA_BROKERS);
        consumerProperties.setProperty("group.id", "Test");
        consumerProperties.setProperty("auto.offset.reset", "latest");
        consumerProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        MyKafkaConsumer kafkaConsumer = new MyKafkaConsumer(consumerProperties, TOPIC);
        Executors.newFixedThreadPool(1).submit(() -> kafkaConsumer.consume());

        Properties producerProperties = new Properties();
        producerProperties.setProperty("bootstrap.servers", KAFKA_BROKERS);
        producerProperties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producerProperties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        MyKafkaProducer kafkaProducer = new MyKafkaProducer(producerProperties, TOPIC);
        kafkaProducer.publish("Test Message");
    }
}

class MyKafkaConsumer {
    private final Logger logger = LoggerFactory.getLogger(MyKafkaConsumer.class);
    private KafkaConsumer<String, Object> kafkaConsumer;

    public MyKafkaConsumer(Properties properties, String topic) {
        kafkaConsumer = new KafkaConsumer<String, Object>(properties);
        kafkaConsumer.subscribe(Lists.newArrayList(topic));
    }

    public void consume() {
        while (true) {
            logger.info("Started listening...");
            ConsumerRecords<String, Object> consumerRecords = kafkaConsumer.poll(Long.MAX_VALUE);
            logger.info("Received records {}", consumerRecords.iterator().next().value());
        }
    }
}

class MyKafkaProducer {
    private KafkaProducer<String, Object> kafkaProducer;
    private String topic;

    public MyKafkaProducer(Properties properties, String topic) {
        this.kafkaProducer = new KafkaProducer<String, Object>(properties);
        this.topic = topic;
    }

    public void publish(Object object) throws ExecutionException, InterruptedException {
        ProducerRecord<String, Object> producerRecord = new ProducerRecord<>(topic, "key", object);
        Future<RecordMetadata> response = kafkaProducer.send(producerRecord);
        response.get();
    }

}