Function NewGroupReaderClient

pkg/ingest/reader_client.go:53–79

(kafkaCfg KafkaConfig, partitionRing ring.PartitionRingReader, metrics *kprom.Metrics, logger log.Logger, opts ...kgo.Opt)

}

func NewGroupReaderClient(kafkaCfg KafkaConfig, partitionRing ring.PartitionRingReader, metrics *kprom.Metrics, logger log.Logger, opts ...kgo.Opt) (*Client, error) {
	opts = append(opts,
		kgo.ConsumerGroup(kafkaCfg.ConsumerGroup),
		kgo.ConsumeTopics(kafkaCfg.Topic),
		kgo.SessionTimeout(3*time.Minute),
		kgo.RebalanceTimeout(5*time.Minute),
		kgo.Balancers(NewCooperativeActiveStickyBalancer(partitionRing)),
		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
	)

	client, err := NewReaderClient(kafkaCfg, metrics, logger, opts...)
	if err != nil {
		return nil, err
	}

	c := &Client{
		Client:        client,
		logger:        logger,
		stopCh:        make(chan struct{}),
		partitionRing: partitionRing,
	}
	// Start the partition monitor goroutine
	c.wg.Add(1)
	go c.monitorPartitions()

	return c, nil
}

func (c *Client) monitorPartitions() {
	defer c.wg.Done()