Method ConsumePartition

consumer.go:159–205

(topic string, partition int32, offset int64)

}

func (c *consumer) ConsumePartition(topic string, partition int32, offset int64) (PartitionConsumer, error) {
	child := &partitionConsumer{
		consumer:             c,
		conf:                 c.conf,
		topic:                topic,
		partition:            partition,
		messages:             make(chan *ConsumerMessage, c.conf.ChannelBufferSize),
		errors:               make(chan *ConsumerError, c.conf.ChannelBufferSize),
		feeder:               make(chan *partitionConsumerResponse, 1),
		leaderEpoch:          invalidLeaderEpoch,
		preferredReadReplica: invalidPreferredReplicaID,
		trigger:              make(chan none, 1),
		dying:                make(chan none),
		dispatcherStop:       make(chan none),
		fetchSize:            c.conf.Consumer.Fetch.Default,
	}

	if err := child.chooseStartingOffset(offset); err != nil {
		return nil, err
	}

	leader, epoch, err := c.client.LeaderAndEpoch(child.topic, child.partition)
	if err != nil {
		return nil, err
	}

	if err := c.addChild(child); err != nil {
		return nil, err
	}

	go withRecover(child.dispatcher)
	go withRecover(child.responseFeeder)

	child.leaderEpoch = epoch
	for {
		child.broker = c.refBrokerConsumer(leader)
		child.brokerSubscription = newBrokerSubscription(child)
		if child.broker.queueSubscription(child.brokerSubscription) {
			break
		}

		child.brokerSubscription.release()
		c.unrefBrokerConsumer(child.broker)
	}

	return child, nil
}

func (c *consumer) HighWaterMarks() map[string]map[int32]int64 {
	c.lock.Lock()