RdKafka\ConsumerTopic::consumeStart

(PECL rdkafka >= 0.9.1)

RdKafka\ConsumerTopic::consumeStart — Start consuming messages from a partition

Description

public RdKafka\ConsumerTopic::consumeStart ( integer $partition , integer $offset ) : void

Start consuming messages for partition at offset offset (see parameters for allowed values).

librdkafka will attempt to keep queued.min.messages (config property) messages in the local queue by repeatedly fetching batches of messages from the broker until the threshold is reached.

The application shall use the RdKafka\ConsumerTopic::consume() method to consume messages from the local queue, each kafka message being represented as a RdKafka\Message object.

Note:

RdKafka\ConsumerTopic::consumeStart() must not be called multiple times for the same topic and partition without stopping consumption first with RdKafka\ConsumerTopic::consumeStop().

Parameters

partition (integer)

Partition ID

offset (integer)

Can be either a proper offset (0..N), or one the the special offset:

  • RD_KAFKA_OFFSET_BEGINNING
  • RD_KAFKA_OFFSET_END
  • RD_KAFKA_OFFSET_STORED
  • The return value of rd_kafka_offset_tail()

Return Values

Returns no value.

Examples

Example #1 RdKafka\ConsumerTopic::consumeStart() example

<?php
$partition = 123;

// consume from the end
$topic->consumeStart($partition, RD_KAFKA_OFFSET_END);

// consume from the stored offset
$topic->consumeStart($partition, RD_KAFKA_OFFSET_STORED);

// consume 200 messages from the end
$topic->consumeStart($partition, rd_kafka_offset_tail(200));
?>