search

Home  >  Q&A  >  body text

How to get the last offset in a topic from Kafka using php library?

I'm writing a Kafka consumer for an API project using the php-rdkafka library. I need to find the last offset in the topic and get the value from it for further processing. For example, last offset in topic = 5, then I need to get offset 5 and send it through API until new offset is added. What I'm trying to run:

$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);

But this returns strange results with negative offsets

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

Although in fact the current last offset is 59. My idea is to get the last offset and then get the value using:

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

I also don't want to use a while(true) loop to quickly perform script work.

That’s all. Thanks.

P粉763662390P粉763662390431 days ago577

reply all(1)I'll reply

  • P粉701491897

    P粉7014918972023-09-11 00:14:41

    I found the answer and it works fine for me:

    $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();

    reply
    0
  • Cancelreply