// DoGetKafkaConsumer returns a Kafka Consumer group
DoGetKafkaConsumer returns a Kafka Consumer group
// DoGetKafkaConsumer returns a Kafka Consumer group
func (e *Init) DoGetKafkaConsumer(ctx context.Context, kafkaCfg *config.KafkaConfig) (dpkafka.IConsumerGroup, error) {
	kafkaOffset := dpkafka.OffsetNewest
	if kafkaCfg.OffsetOldest {
		kafkaOffset = dpkafka.OffsetOldest
	}
	cgConfig := &dpkafka.ConsumerGroupConfig{
		KafkaVersion: &kafkaCfg.Version,
		Offset:       &kafkaOffset,
		Topic:        kafkaCfg.ReindexRequestedTopic,
		GroupName:    kafkaCfg.ReindexRequestedGroup,
		BrokerAddrs:  kafkaCfg.Brokers,
	}
	if kafkaCfg.SecProtocol == config.KafkaTLSProtocolFlag {
		cgConfig.SecurityConfig = dpkafka.GetSecurityConfig(
			kafkaCfg.SecCACerts,
			kafkaCfg.SecClientCert,
			kafkaCfg.SecClientKey,
			kafkaCfg.SecSkipVerify,
		)
	}
	kafkaConsumer, err := dpkafka.NewConsumerGroup(
		ctx,
		cgConfig,
	)
	if err != nil {
		return nil, err
	}
	return kafkaConsumer, nil
}
func ( e * Init ) DoGetKafkaConsumer ( ctx context . Context , kafkaCfg * config . KafkaConfig ) ( dpkafka . IConsumerGroup , error ) { kafkaOffset := dpkafka . OffsetNewest if kafkaCfg . OffsetOldest { kafkaOffset = dpkafka . OffsetOldest } cgConfig := & dpkafka . ConsumerGroupConfig { KafkaVersion : & kafkaCfg . Version , Offset : & kafkaOffset , Topic : kafkaCfg . ReindexRequestedTopic , GroupName : kafkaCfg . ReindexRequestedGroup , BrokerAddrs : kafkaCfg . Brokers , } if kafkaCfg . SecProtocol == config . KafkaTLSProtocolFlag { cgConfig . SecurityConfig = dpkafka . GetSecurityConfig ( kafkaCfg . SecCACerts , kafkaCfg . SecClientCert , kafkaCfg . SecClientKey , kafkaCfg . SecSkipVerify , ) } kafkaConsumer , err := dpkafka . NewConsumerGroup ( ctx , cgConfig , ) if err != nil { return nil , err } return kafkaConsumer , nil }
DoGetKafkaConsumer returns a Kafka Consumer group
DoGetKafkaConsumer returns a Kafka Consumer group