Method NewReader

pkg/acquisition/modules/kafka/config.go:164–212

(dialer *kafka.Dialer, logger *log.Entry)

}

func (c *Configuration) NewReader(dialer *kafka.Dialer, logger *log.Entry) (*kafka.Reader, error) {
	rConf := kafka.ReaderConfig{
		Brokers:     c.Brokers,
		Topic:       c.Topic,
		Dialer:      dialer,
		Logger:      kafka.LoggerFunc(logger.Debugf),
		ErrorLogger: kafka.LoggerFunc(logger.Errorf),
	}

	if c.GroupID != "" && c.Partition != 0 {
		return &kafka.Reader{}, errors.New("cannot specify both group_id and partition")
	}

	if c.GroupID != "" {
		rConf.GroupID = c.GroupID
		// kafka-go does not support calling SetOffset while using a consumer group
		rConf.StartOffset = kafka.LastOffset
	} else if c.Partition != 0 {
		rConf.Partition = c.Partition
	} else {
		logger.Warnf("no group_id specified, crowdsec will only read from the 1st partition of the topic")
	}

	if c.BatchConfiguration.BatchMinBytes != 0 {
		rConf.MinBytes = c.BatchConfiguration.BatchMinBytes
	}

	if c.BatchConfiguration.BatchMaxBytes != 0 {
		rConf.MaxBytes = c.BatchConfiguration.BatchMaxBytes
	}

	if c.BatchConfiguration.BatchMaxWait != 0 {
		rConf.MaxWait = c.BatchConfiguration.BatchMaxWait
	}

	if c.BatchConfiguration.BatchQueueSize != 0 {
		rConf.QueueCapacity = c.BatchConfiguration.BatchQueueSize
	}

	if c.BatchConfiguration.CommitInterval != 0 {
		rConf.CommitInterval = c.BatchConfiguration.CommitInterval
	}

	if err := rConf.Validate(); err != nil {
		return &kafka.Reader{}, fmt.Errorf("while validating reader configuration: %w", err)
	}

	return kafka.NewReader(rConf), nil
}