unit-testing - KafkaTestUtils.getRecords() 只返回第一个发送的记录

Question

发送 4 条记录:

producer.send(new ProducerRecord<>("my-topic", 0, "key1", "value1"));
producer.send(new ProducerRecord<>("my-topic", 0, "key2", "value2"));
producer.send(new ProducerRecord<>("my-topic", 0, "key3", "value3"));
producer.send(new ProducerRecord<>("my-topic", 0, "key4", "value4"));

使用 KafkaTestUtils.getRecords() 只获取第一次发送的记录；

final ConsumerRecords<String, OrderBookViewItem> records = KafkaTestUtils.getRecords(consumer, 10000);

使用 consumer.poll() 获取所有 4 个；

final Iterable<ConsumerRecord<String, OrderBookViewItem>> records = consumer.poll(1000).records("my-topic");

如何调整 KafkaTestUtils 以返回所有记录或它是错误？

Answer 1

它只是返回 Kafka 在 poll() 上给我们的任何内容，该方法不知道要获取多少条记录。

您可以多次调用它，或者您可以尝试设置消费者属性 fetch.min.bytes 和 fetch.max.wait.ms 以便 poll() 等待更多数据。

参见 the kafka documentation有关这些属性的更多信息。