下面是我的消费者和生产者脚本：消费者脚本

$config = ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('localhost:9092');
$config->setGroupId('test');
$config->setBrokerVersion('1.1.1');
$config->setTopics(['Request']);
$config->setOffsetReset('earliest');
$consumer = new Consumer();
$consumer->setLogger($logger);
$consumer->start(function($topic, $part, $message) {
var_dump($message);
});

制片人脚本

$config = ProducerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('localhost:9092');
$config->setBrokerVersion('1.1.1');
$config->setRequiredAck(1);
$config->setIsAsyn(false);
$config->setProduceInterval(500);
$producer = new \Kafka\Producer(
function() {
return [
[
'topic' => 'Request',
'value' => 'My Test Data',
],
];
}
);
$producer->success(function($result) {
var_dump($result);
});
$producer->error(function($errorCode) {
var_dump($errorCode);
});
$producer->send(true);

它工作得很好。当使用者正在运行并且生产者正在发布数据时，使用者能够读取数据。但一旦我停止消费者，然后发布数据，然后启动消费者，消费者就不会读取这些数据。我做错什么了？有人能帮忙吗？