Function main

examples/exactly_once/main.go:60–177

()

}

func main() {
	keepRunning := true
	log.Println("Starting a new Sarama consumer")

	if verbose {
		sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
	}

	version, err := sarama.ParseKafkaVersion(version)
	if err != nil {
		log.Panicf("Error parsing Kafka version: %v", err)
	}

	/**
	 * Construct a new Sarama configuration.
	 * The Kafka cluster version has to be defined before the consumer/producer is initialized.
	 */
	config := sarama.NewConfig()
	config.Version = version

	switch assignor {
	case "sticky":
		config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategySticky()}
	case "roundrobin":
		config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategyRoundRobin()}
	case "range":
		config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategyRange()}
	default:
		log.Panicf("Unrecognized consumer group partition assignor: %s", assignor)
	}

	if oldest {
		config.Consumer.Offsets.Initial = sarama.OffsetOldest
	}

	config.Consumer.IsolationLevel = sarama.ReadCommitted
	config.Consumer.Offsets.AutoCommit.Enable = false

	producerProvider := newProducerProvider(strings.Split(brokers, ","), func() *sarama.Config {
		producerConfig := sarama.NewConfig()
		producerConfig.Version = version

		producerConfig.Net.MaxOpenRequests = 1
		producerConfig.Producer.RequiredAcks = sarama.WaitForAll
		producerConfig.Producer.Idempotent = true
		producerConfig.Producer.Transaction.ID = "sarama"
		return producerConfig
	})

	/**
	 * Setup a new Sarama consumer group
	 */
	consumer := Consumer{
		groupId:          group,
		brokers:          strings.Split(brokers, ","),
		producerProvider: producerProvider,
		ready:            make(chan bool),
	}