kafka-go

	// init kafka
	kafkaURL := fmt.Sprintf("%v:%v", configData.Kafka.Host, configData.Kafka.Port)
	kafkaReader := kafka.NewReader(
		kafka.ReaderConfig{
			Brokers:  []string{kafkaURL},
			Topic:    configData.Kafka.Topic,
			GroupID:  configData.Kafka.GroupID,
			MaxBytes: configData.Kafka.MaxBytes,
		},
	)

	for {
		msg, err := app.Kafka.ReadMessage(context.Background())
		if err != nil {
			panic(err)
		}
		var accepted *models.ConsumerData
		err = json.Unmarshal(msg.Key, &accepted)
		if err != nil {
			continue
		}
		status := handler.SendEmail(accepted)
		if !status {
			return errors.New("email wasn't sent")
		}
	}