最近，在使用kafka时，我的应用程序需要从一开始就访问一个主题中的所有消息。因此，在编写kafka消费者（使用javaapi）时，我能够从头开始读取消息，但它只返回主题中的前500条消息。试图增加
props.put（consumerconfig.max\u poll\u records\u config，integer.max\u value）；props.put（consumerconfig.fetch\u max\u bytes\u config，long.max\u value）；
但它仍然不会返回所有消息，而在使用cli命令时，
kafka控制台使用者--引导服务器localhost:9092 --topic --从一开始
它会返回我所有的5000条记录。
伙计们，有什么配置丢失了吗？任何帮助都是值得感激的。。

[编辑：1]

消费者代码。

public ConsumerRecords<byte[], byte[]> pullFromKafka(String topicname, Map<String, Object> props) {
    KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props);
    consumer.subscribe(new ArrayList<String>(Collections.singletonList(topicname)));
    consumer.poll(0);
    // Reading topic offset from beginning
    consumer.seekToBeginning(consumer.assignment());
    // poll and time-out if no replies
    ConsumerRecords<byte[], byte[]> records = consumer.poll(1000);
    consumer.close();
    return records;
}

然而，我改变了消费者：

public Map<String, byte[]> pullFromKafka(String topicname, Map<String, Object> props) {
    KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<byte[], byte[]>(props);
    Map<String, byte[]> entityMap = new HashMap<String, byte[]>();
    boolean stop = false;
    consumer.subscribe(new ArrayList<String>(Collections.singletonList(topicname)));
    consumer.poll(0);
    // Reading topic offset from beginning
    consumer.seekToBeginning(consumer.assignment());
    while (!stop) {
        // Request unread messages from the topic.
        ConsumerRecords<byte[], byte[]> consumerRecords = consumer.poll(1000);
        Iterator<ConsumerRecord<byte[], byte[]>> iterator = consumerRecords.iterator();
        if (iterator.hasNext()) {
            while (iterator.hasNext()) {
                ConsumerRecord<byte[], byte[]> record = iterator.next();
                // Iterate through returned records, extract the value
                // of each message, and print the value to standard output.
                entityMap.put(new String(record.key()), record.value());
            }
        } else {
            stop = true;
        }
    }
    return entityMap;
}

虽然现在它正在获取所有记录，但我想知道是否有更好的方法。