如何使用 php 函式庫從 Kafka 取得主題中的最後一個偏移量？

Question

我正在使用 php-rdkafka 函式庫為 API 專案編寫 Kafka 消費者。我需要找到主題中的最後一個偏移量並從中獲取值以進行進一步處理。例如，主題中的最後一個偏移 = 5，那麼我需要取得偏移量 5 並透過 API 發送它，直到添加新的偏移量。我正在嘗試運行的內容：

$conf = new RdKafka\Conf();

$settings = [
    'socket.keepalive.enable'  => true,
    'log_level'                => LOG_WARNING,
    'enable.auto.offset.store' => 'true',
    'auto.offset.reset'        => 'earliest',
    'enable.partition.eof'     => 'false',
    'enable.auto.commit'       => 'false',
    'max.poll.interval.ms'     => 300000,
    'session.timeout.ms'       => 45000,
    'group.id'                 => 'test-group',
    'group.instance.id'        => uniqid('', true),
    'metadata.broker.list'     => 'stat-kafka-1:9092,stat-kafka-2:9092,stat-kafka-3:9092',
];
foreach ($settings as $key => $value) {
    $conf->set($key, $value);
}

$topicName = 'userstatistics_12345';
$partition = 0;

$topicPartition = new RdKafka\TopicPartition($topicName, $partition);

$topicPartitionsWithOffsets = $consumer->getOffsetPositions([$topicPartition]);

var_dump($topicPartitionsWithOffsets);

但這會傳回帶有負偏移量的奇怪結果

array(1) { [0]=> object(RdKafka\TopicPartition)#6 (4) { ["topic"]=> string(20) "userstatistics_12345" ["partition"]=> int(0) ["offset"]=> int(-1001) ["err"]=> int(0) } }

雖然實際上目前的最後一個偏移量是59。我的想法是取得最後一個偏移量，然後使用以下方法來取得值：

$consumer->assign([
    new RdKafka\TopicPartition($topicName, $partition, $lastOffset)
]);

我也不想使用 while(true) 迴圈來快速執行腳本工作。

僅此而已。謝謝。

Answer 1

我找到了答案，對我來說效果很好：

$conf = new RdKafka\Conf();

// Configure the group.id. All consumer with the same group.id will consume
// different partitions.
$conf->set('group.id', 'test-group');

// Initial list of Kafka brokers
$conf->set('metadata.broker.list', 'kafka-1:9092');

// Set where to start consuming messages when there is no initial offset in
// offset store or the desired offset is out of range.
// 'earliest': start from the beginning
$conf->set('auto.offset.reset', 'latest');

// Emit EOF event when reaching the end of a partition
$conf->set('enable.partition.eof', 'true');

$kafkaConsumer = new RdKafka\KafkaConsumer($conf);
$topicName = 'topic_name';
$partition = 0;


$topicPartition = new RdKafka\TopicPartition($topicName, 0);
$timeoutMs = 100000;

$low = null;
$high = null;

$wm = $kafkaConsumer->queryWatermarkOffsets($topicName,$partition,$low,$high,$timeoutMs);

$offset = $high - 1;

$kafkaConsumer->assign([new RdKafka\TopicPartition($topicName, $partition, $offset)]);

$message = $kafkaConsumer->consume(1000);

if ($message !== null) {
    // Process the message
    $payload = $message->payload;
    echo "Message at offset $offset: $payload\n";
}

$kafkaConsumer->close();